From a2e3f29c1854494c66df45ec7db52f16ce775c66 Mon Sep 17 00:00:00 2001
From: Santiago Diaz <santiago.d.diaz@gmail.com>
Date: Mon, 6 Apr 2026 17:00:53 -0300
Subject: [PATCH] update execution context usage

---
 lib/mars/execution_context.rb    | 4 ++--
 lib/mars/workflows/parallel.rb   | 5 +++--
 lib/mars/workflows/sequential.rb | 3 ++-
 3 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/lib/mars/execution_context.rb b/lib/mars/execution_context.rb
index 2e4b7be..4904233 100644
--- a/lib/mars/execution_context.rb
+++ b/lib/mars/execution_context.rb
@@ -19,8 +19,8 @@ def record(step_name, output)
       @current_input = output
     end
 
-    def fork(input: current_input)
-      self.class.new(input: input, global_state: global_state)
+    def fork(input: current_input, state: {})
+      self.class.new(input: input, global_state: global_state.merge(state))
     end
 
     def merge(child_contexts)
diff --git a/lib/mars/workflows/parallel.rb b/lib/mars/workflows/parallel.rb
index 4e147d2..ebd76df 100644
--- a/lib/mars/workflows/parallel.rb
+++ b/lib/mars/workflows/parallel.rb
@@ -36,7 +36,7 @@ def aggregate_results(results)
       def execute_steps(context, errors, child_contexts)
         Async do |workflow|
           tasks = steps.map do |step|
-            child_ctx = context.fork
+            child_ctx = context.fork(state: step.state)
             child_contexts << child_ctx
 
             workflow.async do
@@ -54,7 +54,8 @@ def workflow_step(step, child_ctx)
         step.run_before_hooks(child_ctx)
 
         step_input = step.formatter.format_input(child_ctx)
-        result = step.run(step_input)
+        child_ctx.current_input = step_input
+        result = step.run(child_ctx)
 
         if result.is_a?(Halt)
           step.run_after_hooks(child_ctx, result)
diff --git a/lib/mars/workflows/sequential.rb b/lib/mars/workflows/sequential.rb
index 008dd18..9f250a8 100644
--- a/lib/mars/workflows/sequential.rb
+++ b/lib/mars/workflows/sequential.rb
@@ -16,7 +16,8 @@ def run(input)
           step.run_before_hooks(context)
 
           step_input = step.formatter.format_input(context)
-          result = step.run(step_input)
+          step_context = context.fork(input: step_input, state: step.state)
+          result = step.run(step_context)
 
           if result.is_a?(Halt)
             if result.global?