Task level queue registry

lib/minigun/execution/executor.rb

@@ -232,7 +232,6 @@ def fork_for_item(item, stage, user_context, output_queue)
               stage,
               [capture_queue],
               {},
-              {},
               stage_stats: stage_stats
             )
 
@@ -279,7 +278,6 @@ def fork_for_item(item, stage, user_context, output_queue)
             stage,
             [capture_queue],
             {},
-            {},
             stage_stats: stage_stats
           )
           if stage.respond_to?(:block) && stage.block

lib/minigun/pipeline.rb

@@ -5,7 +5,7 @@ module Minigun
   # A Pipeline can be standalone or part of a multi-pipeline Task
   class Pipeline
     attr_reader :name, :config, :stages, :hooks, :dag, :output_queues, :stats,
-                :context, :stage_hooks, :stage_input_queues, :runtime_edges, :input_queues, :parent_pipeline, :task
+                :context, :stage_hooks, :runtime_edges, :input_queues, :parent_pipeline, :task
 
     def initialize(name, task, parent_pipeline, config = {}, stages: nil, hooks: nil, stage_hooks: nil, dag: nil, stats: nil)
       @name = name
@@ -283,8 +283,8 @@ def run_pipeline(_context)
       # Insert router stages for fan-out
       insert_router_stages_for_fan_out
 
-      # Create one input queue per stage (except producers)
-      @stage_input_queues = build_stage_input_queues
+      # Create one input queue per stage (except producers) and register with Task
+      build_stage_input_queues
       @produced_count = Concurrent::AtomicFixnum.new(0)
       @stage_threads = []
 
@@ -303,10 +303,8 @@ def run_pipeline(_context)
       @stage_threads.each(&:join)
     end
 
-    # Build one input queue per stage (except producers)
+    # Build one input queue per stage (except producers) and register with Task
     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?
@@ -318,26 +316,25 @@ def build_stage_input_queues
 
         # Entrance router uses PipelineStage's queue
         if stage == @entrance_router && @input_queues && @input_queues[:input]
-          queues[stage] = @input_queues[:input]
+          register_queue(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]
+          register_queue(stage, @input_queues[:input])
           next
         end
 
         # Use stage's queue_size setting (bounded SizedQueue or unbounded Queue)
         size = stage.queue_size
-        queues[stage] = if size.nil?
-                          Queue.new # Unbounded queue
-                        else
-                          SizedQueue.new(size) # Bounded queue with backpressure
-                        end
+        queue = if size.nil?
+                  Queue.new # Unbounded queue
+                else
+                  SizedQueue.new(size) # Bounded queue with backpressure
+                end
+        register_queue(stage, queue)
       end
-
-      queues
     end
 
     # Insert RouterStage instances for fan-out patterns
@@ -599,6 +596,11 @@ def log_debug(msg)
       Minigun.logger.debug(msg)
     end
 
+    # Register a queue with the task's queue registry
+    def register_queue(stage, queue)
+      @task&.register_stage_queue(stage, queue)
+    end
+
     def log_error(msg)
       Minigun.logger.error(msg)
     end

lib/minigun/queue_wrappers.rb

@@ -40,10 +40,9 @@ def pop
 
   # Wrapper around stage output that routes to downstream queues
   class OutputQueue
-    def initialize(stage, downstream_queues, all_stage_queues, runtime_edges, stage_stats: nil)
+    def initialize(stage, downstream_queues, runtime_edges, stage_stats: nil)
       @stage = stage
       @downstream_queues = downstream_queues # Array of Queue objects
-      @all_stage_queues = all_stage_queues   # Hash keyed by Stage objects
       @runtime_edges = runtime_edges         # Track dynamic routing (keyed by Stage objects)
       @stage_stats = stage_stats             # Stats object for tracking (optional)
       @to_cache = {}                         # Memoization cache for .to() results
@@ -64,11 +63,12 @@ def to(target)
       return @to_cache[target] if @to_cache.key?(target)
 
       # Resolve target to Stage object if it's a name
-      target_stage = pipeline.find_stage(target)
+      # Use StageRegistry for cross-pipeline lookup
+      target_stage = task.stage_registry.find(target, from_pipeline: pipeline)
       raise ArgumentError, "Unknown target stage: #{target}" unless target_stage
 
-      # Look up queue by Stage object
-      target_queue = @all_stage_queues[target_stage]
+      # Look up queue by Stage object using Task's queue registry
+      target_queue = task.find_queue(target_stage)
       raise ArgumentError, "Unknown target stage: #{target} (resolved to #{target_stage.name})" unless target_queue
 
       # Track this as a runtime edge for END signal handling
@@ -78,7 +78,6 @@ def to(target)
       @to_cache[target] = OutputQueue.new(
         @stage,
         [target_queue],
-        @all_stage_queues,
         @runtime_edges,
         stage_stats: @stage_stats
       )
@@ -103,6 +102,10 @@ def to_proc
     def pipeline
       @stage.pipeline
     end
+
+    def task
+      pipeline&.task
+    end
   end
 
   # IPC-backed input queue that reads items from parent via IPC pipe

lib/minigun/stage.rb

@@ -9,7 +9,6 @@ module Minigun
     :stage,
     :dag,
     :runtime_edges,
-    :stage_input_queues,
     :stage_stats,
     # Worker-specific (nil/empty for producers)
     :worker,
@@ -162,11 +161,11 @@ def create_input_queue(stage_ctx)
     def create_output_queue(stage_ctx)
       # DAG and queues now use Stage objects
       downstream = stage_ctx.dag.downstream(stage_ctx.stage)
-      downstream_queues = downstream.filter_map { |ds| stage_ctx.stage_input_queues[ds] }
+      task = stage_ctx.stage.task
+      downstream_queues = downstream.filter_map { |ds| task&.find_queue(ds) }
       OutputQueue.new(
         stage_ctx.stage,
         downstream_queues,
-        stage_ctx.stage_input_queues,
         stage_ctx.runtime_edges,
         stage_stats: stage_ctx.stage_stats
       )
@@ -177,11 +176,13 @@ def send_end_signals(stage_ctx)
       dag_downstream = stage_ctx.dag.downstream(stage_ctx.stage)
       dynamic_targets = stage_ctx.runtime_edges[stage_ctx.stage].to_a
       all_targets = (dag_downstream + dynamic_targets).uniq
+      task = stage_ctx.stage.task
 
       all_targets.each do |target|
-        next unless stage_ctx.stage_input_queues[target]
+        queue = task&.find_queue(target)
+        next unless queue
 
-        stage_ctx.stage_input_queues[target] << EndOfSource.new(stage_ctx.stage)
+        queue << EndOfSource.new(stage_ctx.stage)
       end
     end
 
@@ -388,15 +389,19 @@ def initialize(name, pipeline, targets, options = {})
 
     def send_end_signals(worker_ctx)
       # Broadcast EndOfSource to ALL router targets
+      task = worker_ctx.stage.task
       @targets.each do |target|
-        worker_ctx.stage_input_queues[target] << EndOfSource.new(worker_ctx.stage)
+        queue = task&.find_queue(target)
+        queue&.<< EndOfSource.new(worker_ctx.stage)
       end
     end
   end
 
   # Broadcast router - sends each item to ALL downstream stages
   class RouterBroadcastStage < RouterStage
     def run_stage(worker_ctx)
+      task = worker_ctx.stage.task
+
       loop do
         item = worker_ctx.input_queue.pop
 
@@ -410,7 +415,8 @@ def run_stage(worker_ctx)
 
         # Broadcast to all downstream stages (fan-out semantics)
         @targets.each do |target|
-          worker_ctx.stage_input_queues[target] << item
+          queue = task&.find_queue(target)
+          queue&.<< item
         end
       end
     ensure
@@ -421,7 +427,8 @@ def run_stage(worker_ctx)
   # Round-robin router - distributes items across downstream stages
   class RouterRoundRobinStage < RouterStage
     def run_stage(worker_ctx)
-      target_queues = @targets.map { |target| worker_ctx.stage_input_queues[target] }
+      task = worker_ctx.stage.task
+      target_queues = @targets.map { |target| task&.find_queue(target) }.compact
       round_robin_index = 0
 
       loop do

lib/minigun/task.rb

@@ -20,10 +20,23 @@ def initialize(config: nil, root_pipeline: nil)
       # Initialize the stage_registry for stage management
       @stage_registry = StageRegistry.new
 
+      # Queue registry for cross-pipeline routing (Stage => Queue)
+      @stage_queues = {}
+
       # Root pipeline - all stages and nested pipelines live here
       @root_pipeline = root_pipeline || Pipeline.new(:default, self, nil, @config)
     end
 
+    # Register a stage's input queue for cross-pipeline routing
+    def register_stage_queue(stage, queue)
+      @stage_queues[stage] = queue
+    end
+
+    # Find a stage's input queue for cross-pipeline routing
+    def find_queue(stage)
+      @stage_queues[stage]
+    end
+
     # Set config value (applies to all pipelines)
     def set_config(key, value)
       @config[key] = value

lib/minigun/worker.rb

@@ -63,9 +63,11 @@ def handle_disconnected_stage(stage_ctx) # rubocop:disable Naming/PredicateMetho
 
         # Send EndOfSource to all downstream stages so they don't deadlock
         # DAG and queues now use Stage objects
+        task = stage_ctx.stage.task
         downstream = stage_ctx.dag.downstream(stage_ctx.stage)
         downstream.each do |target|
-          stage_ctx.stage_input_queues[target] << EndOfSource.new(stage_ctx.stage)
+          queue = task&.find_queue(target)
+          queue&.<< EndOfSource.new(stage_ctx.stage)
         end
 
         log_debug 'Done'
@@ -77,7 +79,7 @@ def handle_disconnected_stage(stage_ctx) # rubocop:disable Naming/PredicateMetho
 
     def create_stage_context
       dag = @pipeline.dag
-      stage_input_queues = @pipeline.stage_input_queues
+      task = @pipeline.task
 
       # Calculate sources for workers (empty for autonomous stages)
       # DAG now uses Stage objects instead of names
@@ -109,10 +111,9 @@ def create_stage_context
         stage: @stage,
         dag: dag,
         runtime_edges: @pipeline.runtime_edges,
-        stage_input_queues: stage_input_queues,
         stage_stats: stage_stats,
         # Worker-specific (nil/empty for producers)
-        input_queue: stage_input_queues[@stage],
+        input_queue: task&.find_queue(@stage),
         sources_expected: sources_expected,
         sources_done: Set.new
       )

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

spec/minigun/pipeline_spec.rb

@@ -4,7 +4,7 @@
 
 RSpec.describe Minigun::Pipeline do
   let(:config) { { max_threads: 3, max_processes: 2 } }
-  let(:task) { instance_double(Minigun::Task, stage_registry: nil) }
+  let(:task) { Minigun::Task.new }
   let(:pipeline) { described_class.new(:test_pipeline, task, nil, config) }
 
   describe '#initialize' do
@@ -78,15 +78,15 @@
 
       expect do
         pipeline.add_stage(:producer, :fetch) { |output| output << 'second' }
-      end.to raise_error(Minigun::Error, /Stage name collision.*fetch/)
+      end.to raise_error(Minigun::StageNameConflict, /Stage name.*fetch/)
     end
 
     it 'raises error on duplicate stage name across different types' do
       pipeline.add_stage(:producer, :my_stage) { |output| output << 'data' }
 
       expect do
         pipeline.add_stage(:consumer, :my_stage) { |item| puts item }
-      end.to raise_error(Minigun::Error, /Stage name collision.*my_stage/)
+      end.to raise_error(Minigun::StageNameConflict, /Stage name.*my_stage/)
     end
   end
 
@@ -371,7 +371,7 @@ def initialize
       end.new
 
       # Create PipelineStage that acts as a producer
-      source_pipeline = described_class.new(:source, nil, pipeline, config)
+      source_pipeline = described_class.new(:source, task, pipeline, config)
       pipeline_stage = Minigun::PipelineStage.new(:source_pipeline, pipeline, source_pipeline, nil, {})
       source_pipeline.add_stage(:producer, :gen) { |output| 3.times { |i| output << i } }
       source_pipeline.add_stage(:processor, :double) { |item, output| output << (item * 2) }
@@ -402,7 +402,7 @@ def initialize
       pipeline.add_stage(:producer, :source) { |output| 3.times { |i| output << i } }
 
       # PipelineStage as processor
-      proc_pipeline = described_class.new(:processor, nil, pipeline, config)
+      proc_pipeline = described_class.new(:processor, task, pipeline, config)
       pipeline_stage = Minigun::PipelineStage.new(:processor_pipeline, pipeline, proc_pipeline, nil, {})
       proc_pipeline.add_stage(:processor, :multiply) { |item, output| output << (item * 10) }
       proc_pipeline.add_stage(:processor, :add_one) { |item, output| output << (item + 1) }
@@ -433,7 +433,7 @@ def initialize
       end.new
 
       # First PipelineStage producer
-      p1 = described_class.new(:pa, nil, pipeline, config)
+      p1 = described_class.new(:pa, task, pipeline, config)
       ps1 = Minigun::PipelineStage.new(:pipeline_a, pipeline, p1, nil, {})
       p1.add_stage(:producer, :gen) { |output| output << 10 }
       p1.add_stage(:processor, :double) { |item, output| output << (item * 2) }
@@ -442,7 +442,7 @@ def initialize
       pipeline.dag.add_node(ps1)  # Use Stage object
 
       # Second PipelineStage producer
-      p2 = described_class.new(:pb, nil, pipeline, config)
+      p2 = described_class.new(:pb, task, pipeline, config)
       ps2 = Minigun::PipelineStage.new(:pipeline_b, pipeline, p2, nil, {})
       p2.add_stage(:producer, :gen) { |output| output << 5 }
       p2.add_stage(:processor, :triple) { |item, output| output << (item * 3) }

spec/minigun/stage_spec.rb

@@ -4,7 +4,7 @@
 
 RSpec.describe Minigun::Stage do
   let(:mock_registry) { instance_double(Minigun::StageRegistry, register: nil) }
-  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry) }
+  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry, find_queue: nil) }
   let(:mock_pipeline) { instance_double(Minigun::Pipeline, name: 'test_pipeline', task: mock_task) }
 
   describe 'base class' do
@@ -31,7 +31,7 @@
 
 RSpec.describe Minigun::ProducerStage do
   let(:mock_registry) { instance_double(Minigun::StageRegistry, register: nil) }
-  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry) }
+  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry, find_queue: nil) }
   let(:mock_pipeline) { instance_double(Minigun::Pipeline, name: 'test_pipeline', task: mock_task) }
 
   describe 'producer behavior' do
@@ -60,7 +60,7 @@
 
 RSpec.describe Minigun::ConsumerStage do
   let(:mock_registry) { instance_double(Minigun::StageRegistry, register: nil) }
-  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry) }
+  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry, find_queue: nil) }
   let(:mock_pipeline) { instance_double(Minigun::Pipeline, name: 'test_pipeline', task: mock_task) }
 
   describe 'processor behavior' do
@@ -117,7 +117,7 @@
 
 RSpec.describe Minigun::AccumulatorStage do
   let(:mock_registry) { instance_double(Minigun::StageRegistry, register: nil) }
-  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry) }
+  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry, find_queue: nil) }
   let(:mock_pipeline) { instance_double(Minigun::Pipeline, name: 'test_pipeline', task: mock_task) }
 
   it 'is a special batching stage' do
@@ -128,7 +128,7 @@
 
 RSpec.describe 'Stage common behavior' do
   let(:mock_registry) { instance_double(Minigun::StageRegistry, register: nil) }
-  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry) }
+  let(:mock_task) { instance_double(Minigun::Task, stage_registry: mock_registry, find_queue: nil) }
   let(:mock_pipeline) { instance_double(Minigun::Pipeline, name: 'test_pipeline', task: mock_task) }
   let(:stage) { Minigun::ConsumerStage.new(:test, mock_pipeline, proc { |x, _output| x * 2 }, { foo: 'bar' }) }