Fix routing to pipelines wip

lib/minigun/pipeline.rb

@@ -21,6 +21,9 @@ def initialize(name, task, parent_pipeline, config = {}, stages: nil, hooks: nil
       @stages = stages || []
       @deferred_edges = [] # Store edges with forward references: [{from: stage, to: :name}, ...]
 
+      # Entrance stage will be created lazily in insert_entrance_distributor_for_inputs!
+      @entrance_stage = nil
+
       # Pipeline-level hooks (run once per pipeline)
       @hooks = hooks || {
         before_run: [],
@@ -513,19 +516,18 @@ def insert_entrance_distributor_for_inputs!
         # Skip autonomous stages (they're producers)
         next false if stage.run_mode == :autonomous
         # Entry stages have no upstream
-        @dag.upstream(stage).empty?
+        @dag.upstream(stage).empty()
       end
 
-      return if entry_stages.empty?
+      return if entry_stages.empty()
 
-      # Create a consumer stage that reads from parent input and emits to nested pipeline
-      # Use ConsumerStage (not ProducerStage) so it properly tracks multiple END signals
-      parent_input = @input_queues[:input]
+      # Create entrance stage lazily (now that @input_queues is set)
       entrance_block = proc do |item, output|
         # Just forward items from parent to nested pipeline
         output << item
       end
-      entrance_stage = Minigun::EntranceStage.new(nil, self, entrance_block, {})
+      entrance_stage = Minigun::EntranceStage.new(@name, self, entrance_block, {})
+      @entrance_stage = entrance_stage
 
       # Add the :_entrance stage to the pipeline (at the beginning)
       @stages.unshift(entrance_stage)

lib/minigun/stage.rb

@@ -389,7 +389,13 @@ def initialize(name, pipeline, targets, options = {})
     def send_end_signals(worker_ctx)
       # Broadcast EndOfSource to ALL router targets
       @targets.each do |target|
-        worker_ctx.stage_input_queues[target] << EndOfSource.new(worker_ctx.stage)
+        queue = worker_ctx.stage_input_queues[target]
+        if queue.nil?
+          puts "[DEBUG send_end_signals] Target: #{target.inspect} (#{target.class}) - queue is nil!"
+          puts "[DEBUG] Available queues: #{worker_ctx.stage_input_queues.keys.map(&:inspect).join(', ')}"
+          raise "No queue for target #{target.inspect}"
+        end
+        queue << EndOfSource.new(worker_ctx.stage)
       end
     end
   end

lib/minigun/stage_registry.rb

@@ -112,6 +112,21 @@ def clear
 
     private
 
+    # If stage is a PipelineStage, resolve to its EntranceStage for routing
+    # Otherwise return the stage as-is
+    def resolve_to_entrance(stage)
+      return nil unless stage
+
+      # Check if this is a PipelineStage with a nested pipeline
+      if stage.is_a?(Minigun::PipelineStage) && stage.nested_pipeline
+        # Find the entrance stage (first stage in nested pipeline)
+        entrance = stage.nested_pipeline.stages.first
+        return entrance if entrance.is_a?(Minigun::EntranceStage)
+      end
+
+      stage
+    end
+
     # Normalize name to string (symbols and strings both work)
     def normalize_name(name)
       return nil if name.nil?

lib/minigun/task.rb

@@ -108,6 +108,8 @@ def define_pipeline(name, options = {})
         pipeline = pipeline_stage.nested_pipeline
       else
         # Create new PipelineStage and add to root_pipeline (pipeline-first positional style)
+        # Both PipelineStage and EntranceStage get the pipeline's name
+        # Routing to the PipelineStage will be redirected to the EntranceStage
         pipeline = Pipeline.new(name, self, @root_pipeline, @config)
         pipeline_stage = PipelineStage.new(name, @root_pipeline, pipeline, nil, options)
 

spec/integration/examples_spec.rb

@@ -175,8 +175,6 @@
 
   describe '07_multi_pipeline_data_processing.rb' do
     it 'processes data through validation and routing pipelines' do
-      skip 'Priority-based routing from within pipeline stages not yet supported'
-
       load File.expand_path('../../examples/07_multi_pipeline_data_processing.rb', __dir__)
 
       processor = DataProcessingPipeline.new
@@ -1396,7 +1394,7 @@
     end
   end
 
-  describe '66_cow_and_ipc_fork_executors.rb', skip: Gem.win_platform? do
+  describe '66_cow_and_ipc_fork_executors.rb', skip: !Process.respond_to?(:fork) do
     it 'demonstrates COW and IPC fork executors' do
       load File.expand_path('../../examples/66_cow_and_ipc_fork_executors.rb', __dir__)
 

spec/unit/execution/executor_spec.rb

@@ -315,7 +315,7 @@
   end
 end
 
-RSpec.describe Minigun::Execution::CowForkPoolExecutor, skip: Gem.win_platform? do
+RSpec.describe Minigun::Execution::CowForkPoolExecutor, skip: !Process.respond_to?(:fork) do
   let(:stage_ctx) do
     dag = double('dag', terminal?: false)
     pipeline = double('pipeline', name: 'test_pipeline', dag: dag, send: nil)
@@ -398,7 +398,7 @@
   end
 end
 
-RSpec.describe Minigun::Execution::CowForkPoolExecutor, skip: Gem.win_platform? do
+RSpec.describe Minigun::Execution::CowForkPoolExecutor, skip: !Process.respond_to?(:fork) do
   let(:stage_ctx) do
     dag = double('dag', terminal?: false)
     pipeline = double('pipeline', name: 'test_pipeline', dag: dag, send: nil)
@@ -496,7 +496,7 @@
   end
 end
 
-RSpec.describe Minigun::Execution::IpcForkPoolExecutor, skip: Gem.win_platform? do
+RSpec.describe Minigun::Execution::IpcForkPoolExecutor, skip: !Process.respond_to?(:fork) do
   let(:stage_ctx) do
     dag = double('dag', terminal?: false)
     pipeline = double('pipeline', name: 'test_pipeline', dag: dag, send: nil)

spec/unit/execution/fork_executors_jepsen_spec.rb

@@ -12,7 +12,7 @@
 # - Edge cases and chaos scenarios
 # - Resource cleanup
 
-RSpec.describe 'Fork Executors - Jepsen-style Tests', skip: Gem.win_platform? do
+RSpec.describe 'Fork Executors - Jepsen-style Tests', skip: !Process.respond_to?(:fork) do
   let(:dag) { double('dag', terminal?: false) }
   let(:pipeline) do
     double('pipeline',

spec/unit/yield_syntax_spec.rb

@@ -207,7 +207,7 @@ def call(input_queue, _output_queue)
   describe 'yield with routing' do
     # Known limitation: stages with only dynamically-routed inputs don't wait for input
     # This is tracked separately as a general dynamic routing limitation
-    it 'supports yield(item, to: :stage_name) - known limitation with dynamic routing', skip: 'Known limitation with dynamic routing' do
+    it 'supports yield(item, to: :stage_name)' do
       router_stage = Class.new(Minigun::ConsumerStage) do
         def call(item, _output)
           if item.even?