Replace pipeline entrance with a router

.gitignore

@@ -49,6 +49,7 @@ pickle-email-*.html
 .envrc
 .ruby-gemset
 .rspec_status
+.claude
 
 # Redis DB
 dump.rdb

TODOS.md

@@ -22,10 +22,6 @@ TODO: Refactor so more things are moved from Stage to Worker, e.g.
 - sending of end signals?
 - rename #run_worker_loop as its not a loop. Maybe #run_in_worker? other ideas? --> DONE: renamed to #run_stage
 
-=======================================
-
-@stage_name == :_entrance and :_exit (YUCK)
-
 ================================
 
 This needs to be in all stage:

lib/minigun/pipeline.rb

@@ -20,6 +20,7 @@ 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_router = nil # Router stage for nested pipeline entrance (if multiple entry stages)
 
       # Pipeline-level hooks (run once per pipeline)
       @hooks = hooks || {
@@ -306,13 +307,24 @@ def run_pipeline(_context)
     def build_stage_input_queues
       queues = {}
 
+      # Find entrance infrastructure (router or single entry stage)
+      entry_stages = @stages.select do |s|
+        s.run_mode != :autonomous && @dag.upstream(s).empty?
+      end
+
       @stages.each do |stage|
         # Skip autonomous stages - they don't have input queues
         next if stage.run_mode == :autonomous
 
-        # Special case: EntranceStage uses the parent pipeline's input queue
-        if stage.is_a?(EntranceStage) && @input_queues && @input_queues[:input]
-          queues[stage] = @input_queues[:input]  # Key by object
+        # Entrance router uses PipelineStage's queue
+        if stage == @entrance_router && @input_queues && @input_queues[:input]
+          queues[stage] = @input_queues[:input]
+          next
+        end
+
+        # Single entry stage uses PipelineStage's queue directly (no router)
+        if entry_stages.size == 1 && stage == entry_stages.first && @input_queues && @input_queues[:input]
+          queues[stage] = @input_queues[:input]
           next
         end
 
@@ -425,7 +437,7 @@ def build_dag_routing!
       # Fill any remaining sequential gaps (handles fan-out, siblings, cycles)
       fill_sequential_gaps_by_definition_order!
 
-      # If this pipeline has input_queues (nested pipeline), add :_entrance distributor
+      # If this pipeline has input_queues (nested pipeline), add entrance distributor
       insert_entrance_distributor_for_inputs! if @input_queues && !@input_queues.empty?
 
       # If this pipeline has output_queues, add :_exit collector for terminal stages
@@ -504,40 +516,44 @@ def fill_sequential_gaps_by_definition_order!
       end
     end
 
-    # Insert an :_entrance distributor stage for nested pipelines
-    # This receives items from the parent pipeline and distributes to entry stages
+    # Insert an entrance distributor for nested pipelines
+    # Uses RouterStage for multiple entries, direct connection for single entry
     def insert_entrance_distributor_for_inputs!
       # Find stages that have no upstream (would be entry points)
-      # After normalization, DAG uses Stage objects
       entry_stages = @stages.select do |stage|
-        # Skip autonomous stages (they're producers)
         next false if stage.run_mode == :autonomous
-        # Entry stages have no upstream
         @dag.upstream(stage).empty?
       end
 
       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]
-      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, {})
+      # Single entry stage: it uses PipelineStage's queue directly (no router needed)
+      # This is handled in build_stage_input_queues
+      return if entry_stages.size == 1
+
+      # Multiple entry stages: create a router to distribute items
+      # Get routing strategy from parent_pipeline's config (where PipelineStage lives)
+      pipeline_stage = @parent_pipeline&.stages&.find { |s| s.is_a?(PipelineStage) && s.nested_pipeline == self }
+      routing_strategy = pipeline_stage&.options&.[](:routing) || :broadcast
+
+      # Create anonymous router stage
+      @entrance_router = if routing_strategy == :round_robin
+                           RouterRoundRobinStage.new(nil, self, entry_stages.dup, {})
+                         else
+                           RouterBroadcastStage.new(nil, self, entry_stages.dup, {})
+                         end
 
-      # Add the :_entrance stage to the pipeline (at the beginning)
-      @stages.unshift(entrance_stage)
-      @dag.add_node(entrance_stage)
+      # Add router to pipeline
+      @stages.unshift(@entrance_router)
+      @dag.add_node(@entrance_router)
 
-      # Connect :_entrance to entry stages (all objects)
+      # Connect router to entry stages
       entry_stages.each do |stage|
-        @dag.add_edge(entrance_stage, stage)
+        @dag.add_edge(@entrance_router, stage)
       end
 
       entry_names = entry_stages.map(&:name).join(', ')
-      log_debug "[Pipeline:#{@name}] Added :_entrance distributor for entry stages: #{entry_names}"
+      log_debug "[Pipeline:#{@name}] Added #{routing_strategy} entrance router for entry stages: #{entry_names}"
     end
 
     # Insert an :_exit collector stage that terminal stages drain into

lib/minigun/stage.rb

@@ -444,15 +444,6 @@ def run_stage(worker_ctx)
     end
   end
 
-  # Special entrance stage for nested pipelines
-  # Automatically created when a pipeline has input from parent
-  class EntranceStage < ConsumerStage
-    # Positional constructor: EntranceStage.new(name, pipeline, block, options)
-    def initialize(name, pipeline, block, options = {})
-      super(name, pipeline, block, options)
-    end
-  end
-
   # Special exit stage for nested pipelines
   # Automatically created when a pipeline has output to parent
   class ExitStage < ConsumerStage
@@ -482,10 +473,10 @@ def run_stage(stage_ctx)
       return unless @nested_pipeline
 
       # Set up input/output queues for the nested pipeline
-      # The pipeline will create :_entrance and :_exit stages based on these
+      # Pass the PipelineStage's input queue to nested pipeline so entry stages can use it
       if !stage_ctx.sources_expected.empty?
-        # Has upstream: set input queue so pipeline creates :_entrance
-        # Also pass the expected source count for proper END signal handling
+        # Has upstream: pass input queue to nested pipeline
+        # The nested pipeline will handle distributing to its entry stages
         @nested_pipeline.instance_variable_set(:@input_queues, {
           input: stage_ctx.input_queue,
           sources_expected: stage_ctx.sources_expected
@@ -495,7 +486,7 @@ def run_stage(stage_ctx)
       # Always set output queue so pipeline creates :_exit
       @nested_pipeline.instance_variable_set(:@output_queues, { output: create_output_queue(stage_ctx) })
 
-      # Run the nested pipeline (it will automatically create :_entrance/:_exit as needed)
+      # Run the nested pipeline (it will handle input distribution to entry stages)
       @nested_pipeline.run(stage_ctx.root_pipeline.context)
     ensure
       send_end_signals(stage_ctx)

lib/minigun/worker.rb

@@ -83,11 +83,20 @@ def create_stage_context
       # DAG now uses Stage objects instead of names
       sources_expected = if @stage.run_mode == :autonomous
                            Set.new
-                         elsif @stage.is_a?(Minigun::EntranceStage) && @pipeline.input_queues
-                           # For EntranceStage, use sources from parent pipeline if available
-                           @pipeline.input_queues[:sources_expected] || Set.new
                          else
-                           Set.new(dag.upstream(@stage))
+                           # Check if this stage is an entrance router or single entry stage for nested pipeline
+                           input_queues = @pipeline.instance_variable_get(:@input_queues)
+                           entrance_router = @pipeline.instance_variable_get(:@entrance_router)
+
+                           if @stage == entrance_router && input_queues
+                             # For entrance router, use sources from parent pipeline
+                             input_queues[:sources_expected] || Set.new
+                           elsif dag.upstream(@stage).empty? && input_queues && input_queues[:sources_expected]
+                             # For single entry stage with no upstream, use sources from parent pipeline if available
+                             input_queues[:sources_expected]
+                           else
+                             Set.new(dag.upstream(@stage))
+                           end
                          end
 
       # Create stats object for this specific stage