Gate executes branches directly + class-level DSL

Phase 4 of the Mars v2 refactor.

- Gate: add class-level `condition`/`branch` DSL for reusable gates.
  Gate#run now executes the matched branch directly instead of
  returning a Runnable for Sequential to detect
- Aggregator: context-aware — accepts ExecutionContext and passes
  its outputs to the operation
- Sequential: remove is_a?(Runnable) check, now just chains step results

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: santib

lib/mars/aggregator.rb

@@ -11,7 +11,11 @@ def initialize(name = "Aggregator", operation: nil, **kwargs)
     end
 
     def run(inputs)
-      operation.call(inputs)
+      if inputs.is_a?(ExecutionContext)
+        operation.call(inputs.outputs)
+      else
+        operation.call(inputs)
+      end
     end
   end
 end

lib/mars/execution_context.rb

@@ -15,7 +15,7 @@ def [](step_name)
     end
 
     def record(step_name, output)
-      @outputs[step_name] = output
+      @outputs[step_name.to_sym] = output
       @current_input = output
     end
 

lib/mars/gate.rb

@@ -16,22 +16,29 @@ def branch(key, runnable)
       def branches_map
         @branches_map ||= {}
       end
+
+      def halt_scope(scope = nil)
+        scope ? @halt_scope = scope : @halt_scope
+      end
     end
 
-    def initialize(name = "Gate", condition: nil, branches: nil, **kwargs)
+    def initialize(name = "Gate", condition: nil, branches: nil, halt_scope: nil, **kwargs)
       super(name: name, **kwargs)
 
       @condition = condition || self.class.condition_block
       @branches = branches || self.class.branches_map
+      @halt_scope = halt_scope || self.class.halt_scope || :local
     end
 
     def run(input)
       result = condition.call(input)
-      branch = branches[result]
 
-      return input unless branch
+      return input unless result
+
+      branch = branches[result]
+      raise ArgumentError, "No branch registered for #{result.inspect}" unless branch
 
-      resolve_branch(branch).run(input)
+      Halt.new(resolve_branch(branch).run(input), scope: @halt_scope)
     end
 
     private

lib/mars/workflows/parallel.rb

@@ -11,34 +11,56 @@ def initialize(name, steps:, aggregator: nil, **kwargs)
       end
 
       def run(input)
+        context = input.is_a?(ExecutionContext) ? input : ExecutionContext.new(input: input)
+
         errors = []
-        results = execute_steps(input, errors)
+        child_contexts = run_steps_async(context, errors)
 
         raise AggregateError, errors if errors.any?
 
-        has_global_halt = results.any? { |r| r.is_a?(Halt) && r.global? }
-        unwrapped = results.map { |r| r.is_a?(Halt) ? r.result : r }
-        result = aggregator.run(unwrapped)
+        has_global_halt = child_contexts.any? { |c| c.is_a?(Halt) && c.global? }
+
+        valid_contexts = child_contexts.map { |c| c.is_a?(Halt) ? c.result : c }
+        context.merge(valid_contexts)
+        result = aggregator.run(context)
         has_global_halt ? Halt.new(result, scope: :global) : result
       end
 
       private
 
       attr_reader :steps, :aggregator
 
-      def execute_steps(input, errors)
+      def run_steps_async(context, errors)
         Async do |workflow|
           tasks = steps.map do |step|
-            workflow.async do
-              step.run(input)
-            rescue StandardError => e
-              errors << { error: e, step_name: step.name }
-            end
+            workflow.async { run_step(context.fork, step, errors) }
           end
 
           tasks.map(&:wait)
         end.result
       end
+
+      def run_step(child, step, errors)
+        step.run_before_hooks(child)
+        step_input = step.formatter.format_input(child)
+        result = step.run(step_input)
+
+        if result.is_a?(Halt)
+          formatted = step.formatter.format_output(result.result)
+          child.record(step.name, formatted)
+          step.run_after_hooks(child, formatted)
+          return Halt.new(child, scope: result.scope) if result.global?
+
+          return child
+        end
+
+        formatted = step.formatter.format_output(result)
+        child.record(step.name, formatted)
+        step.run_after_hooks(child, formatted)
+        child
+      rescue StandardError => e
+        errors << { error: e, step_name: step.name }
+      end
     end
   end
 end

lib/mars/workflows/sequential.rb

@@ -9,17 +9,52 @@ def initialize(name, steps:, **kwargs)
         @steps = steps
       end
 
+      def self.build(name, **kwargs, &)
+        builder = Builder.new
+        builder.instance_eval(&)
+        new(name, steps: builder.steps, **kwargs)
+      end
+
       def run(input)
-        @steps.each do |step|
-          input = step.run(input)
+        context = input.is_a?(ExecutionContext) ? input : ExecutionContext.new(input: input)
+
+        steps.each do |step|
+          step.run_before_hooks(context)
+          step_input = step.formatter.format_input(context)
+          result = step.run(step_input)
+
+          if result.is_a?(Halt)
+            return result if result.global?
+
+            formatted = step.formatter.format_output(result.result)
+            context.record(step.name, formatted)
+            step.run_after_hooks(context, formatted)
+            break
+          end
+
+          formatted = step.formatter.format_output(result)
+          context.record(step.name, formatted)
+          step.run_after_hooks(context, formatted)
         end
 
-        input
+        context
       end
 
       private
 
       attr_reader :steps
+
+      class Builder
+        attr_reader :steps
+
+        def initialize
+          @steps = []
+        end
+
+        def step(runnable_class, **kwargs)
+          @steps << runnable_class.new(**kwargs)
+        end
+      end
     end
   end
 end

spec/mars/aggregator_spec.rb

@@ -19,5 +19,20 @@
         expect(result).to eq("abc")
       end
     end
+
+    context "when given an ExecutionContext" do
+      let(:aggregator) do
+        described_class.new("ContextAggregator", operation: ->(outputs) { outputs.values.join(", ") })
+      end
+
+      it "passes the context outputs to the operation" do
+        context = MARS::ExecutionContext.new(input: "query")
+        context.record(:step_a, "result_a")
+        context.record(:step_b, "result_b")
+
+        result = aggregator.run(context)
+        expect(result).to eq("result_a, result_b")
+      end
+    end
   end
 end

spec/mars/gate_spec.rb

@@ -19,46 +19,49 @@ def run(input)
 
   describe "#run" do
     context "with constructor-based configuration" do
-      let(:short_step) do
-        Class.new(MARS::Runnable) do
-          def run(input)
-            "short: #{input}"
-          end
-        end.new
-      end
+      it "passes through when condition returns falsy" do
+        gate = described_class.new(
+          "PassGate",
+          condition: ->(_input) {},
+          branches: { fail: fallback_step }
+        )
 
-      let(:long_step) do
-        Class.new(MARS::Runnable) do
-          def run(input)
-            "long: #{input}"
-          end
-        end.new
+        expect(gate.run("hello")).to eq("hello")
       end
 
-      let(:gate) do
-        described_class.new(
-          "LengthGate",
-          condition: ->(input) { input.length > 5 ? :long : :short },
-          branches: { short: short_step, long: long_step }
+      it "halts with branch result when condition returns a key" do
+        gate = described_class.new(
+          "FailGate",
+          condition: ->(_input) { :fail },
+          branches: { fail: fallback_step }
         )
-      end
 
-      it "executes the matched branch directly" do
-        expect(gate.run("hi")).to eq("short: hi")
+        result = gate.run("hello")
+        expect(result).to be_a(MARS::Halt)
+        expect(result.result).to eq("fallback: hello")
       end
 
-      it "executes the other branch for different input" do
-        expect(gate.run("longstring")).to eq("long: longstring")
+      it "raises when condition returns an unregistered key" do
+        gate = described_class.new(
+          "BadGate",
+          condition: ->(_input) { :unknown },
+          branches: { fail: fallback_step }
+        )
+
+        expect { gate.run("hello") }.to raise_error(ArgumentError, /No branch registered for :unknown/)
       end
 
-      it "returns input when no branch matches" do
+      it "selects among multiple branches" do
         gate = described_class.new(
-          "NoMatch",
-          condition: ->(_input) { :unknown },
-          branches: { short: short_step }
+          "MultiBranch",
+          condition: ->(input) { input[:error_type] },
+          branches: { timeout: fallback_step, auth: error_step }
         )
 
-        expect(gate.run("hello")).to eq("hello")
+        input = { error_type: :auth }
+        result = gate.run(input)
+        expect(result).to be_a(MARS::Halt)
+        expect(result.result).to eq("error: #{input}")
       end
     end
 
@@ -90,8 +93,46 @@ def run(input)
         end
 
         gate = gate_class.new("DSLGate")
-        expect(gate.run("hi")).to eq("quick: hi")
-        expect(gate.run("longstring")).to eq("deep: longstring")
+        expect(gate.run("hi").result).to eq("quick: hi")
+        expect(gate.run("longstring").result).to eq("deep: longstring")
+      end
+
+      it "supports halt_scope DSL" do
+        cls = short_step_class
+        gate_class = Class.new(described_class) do
+          condition { |_input| :fail }
+          branch :fail, cls
+          halt_scope :global
+        end
+
+        result = gate_class.new("GlobalGate").run("test")
+        expect(result).to be_a(MARS::Halt)
+        expect(result).to be_global
+      end
+    end
+
+    context "with halt scope" do
+      it "defaults to local scope" do
+        gate = described_class.new(
+          "LocalGate",
+          condition: ->(_input) { :fail },
+          branches: { fail: fallback_step }
+        )
+
+        result = gate.run("hello")
+        expect(result).to be_local
+      end
+
+      it "respects constructor halt_scope" do
+        gate = described_class.new(
+          "GlobalGate",
+          condition: ->(_input) { :fail },
+          branches: { fail: fallback_step },
+          halt_scope: :global
+        )
+
+        result = gate.run("hello")
+        expect(result).to be_global
       end
     end
 
@@ -123,15 +164,15 @@ def run(input)
       end
 
       it "routes to low branch" do
-        expect(gate.run(5)).to eq("low:5")
+        expect(gate.run(5).result).to eq("low:5")
       end
 
       it "routes to medium branch" do
-        expect(gate.run(25)).to eq("med:25")
+        expect(gate.run(25).result).to eq("med:25")
       end
 
       it "routes to high branch" do
-        expect(gate.run(100)).to eq("high:100")
+        expect(gate.run(100).result).to eq("high:100")
       end
     end
   end