AstraZeneca
diff --git a/‎argo-pipeline.yaml‎
Lines changed: 83 additions & 105 deletions b/‎argo-pipeline.yaml‎
Lines changed: 83 additions & 105 deletions
diff --git a/‎docs/plans/branch-parameter-rollback.md‎
Lines changed: 440 additions & 0 deletions b/‎docs/plans/branch-parameter-rollback.md‎
Lines changed: 440 additions & 0 deletions
diff --git a/‎examples/02-sequential/conditional_rollback.py‎
Lines changed: 66 additions & 0 deletions b/‎examples/02-sequential/conditional_rollback.py‎
Lines changed: 66 additions & 0 deletions
diff --git a/‎examples/06-parallel/parallel_conflict.py‎
Lines changed: 41 additions & 0 deletions b/‎examples/06-parallel/parallel_conflict.py‎
Lines changed: 41 additions & 0 deletions
diff --git a/‎examples/06-parallel/parallel_rollback.py‎
Lines changed: 64 additions & 0 deletions b/‎examples/06-parallel/parallel_rollback.py‎
Lines changed: 64 additions & 0 deletions
diff --git a/‎examples/common/functions.py‎
Lines changed: 56 additions & 0 deletions b/‎examples/common/functions.py‎
Lines changed: 56 additions & 0 deletions
diff --git a/‎extensions/job_executor/__init__.py‎
Lines changed: 3 additions & 2 deletions b/‎extensions/job_executor/__init__.py‎
Lines changed: 3 additions & 2 deletions
@@ -0,0 +1,66 @@
+"""
+Demonstrate parameter rollback from conditional branches.
+
+When a conditional branch executes and succeeds, parameters set within
+that branch roll back to the parent scope.
+
+Execute with:
+    python examples/10-branch-parameters/conditional_rollback.py
+"""
+
+from examples.common.functions import (
+    set_conditional_heads_param,
+    set_conditional_tails_param,
+    verify_conditional_rollback,
+)
+from runnable import Conditional, Pipeline, PythonTask
+
+
+def decide_heads():
+    """Return 'heads' to select heads branch."""
+    return "heads"
+
+
+def main():
+    # Create branch pipelines that set parameters
+    heads_pipeline = PythonTask(
+        name="heads_task",
+        function=set_conditional_heads_param,
+        returns=["branch_param"],
+    ).as_pipeline()
+
+    tails_pipeline = PythonTask(
+        name="tails_task",
+        function=set_conditional_tails_param,
+        returns=["branch_param"],
+    ).as_pipeline()
+
+    # Conditional node selects branch based on 'choice' parameter
+    conditional = Conditional(
+        name="conditional",
+        branches={"heads": heads_pipeline, "tails": tails_pipeline},
+        parameter="choice",
+    )
+
+    # Task to set the choice parameter
+    decide_task = PythonTask(
+        name="decide",
+        function=decide_heads,
+        returns=["choice"],
+    )
+
+    # Task to verify the parameter rolled back from branch
+    verify_task = PythonTask(
+        name="verify",
+        function=verify_conditional_rollback,
+        terminate_with_success=True,
+    )
+
+    pipeline = Pipeline(steps=[decide_task, conditional, verify_task])
+    pipeline.execute()
+
+    return pipeline
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,41 @@
+"""
+Demonstrate parameter conflict resolution in parallel branches.
+
+When multiple branches set the same parameter, last write wins
+based on dictionary iteration order.
+
+Execute with:
+    python examples/10-branch-parameters/parallel_conflict.py
+"""
+
+from examples.common.functions import set_shared_param_a, set_shared_param_b
+from runnable import Parallel, Pipeline, PythonTask
+
+
+def main():
+    # Both branches set the same parameter name
+    branch1_pipeline = PythonTask(
+        name="branch1_task",
+        function=set_shared_param_a,
+        returns=["shared"],
+    ).as_pipeline()
+
+    branch2_pipeline = PythonTask(
+        name="branch2_task",
+        function=set_shared_param_b,
+        returns=["shared"],
+    ).as_pipeline()
+
+    parallel = Parallel(
+        name="parallel",
+        branches={"branch1": branch1_pipeline, "branch2": branch2_pipeline},
+    )
+
+    pipeline = Pipeline(steps=[parallel])
+    pipeline.execute()
+
+    return pipeline
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,64 @@
+"""
+Demonstrate parameter rollback from parallel branches.
+
+When parallel branches execute and succeed, parameters from all branches
+roll back to the parent scope.
+
+Execute with:
+    python examples/10-branch-parameters/parallel_rollback.py
+"""
+
+from examples.common.functions import (
+    set_parallel_branch1,
+    set_parallel_branch2,
+    set_parallel_branch3,
+    verify_parallel_rollback,
+)
+from runnable import Parallel, Pipeline, PythonTask
+
+
+def main():
+    # Create branch pipelines that set different parameters
+    branch1_pipeline = PythonTask(
+        name="branch1_task",
+        function=set_parallel_branch1,
+        returns=["result1"],
+    ).as_pipeline()
+
+    branch2_pipeline = PythonTask(
+        name="branch2_task",
+        function=set_parallel_branch2,
+        returns=["result2"],
+    ).as_pipeline()
+
+    branch3_pipeline = PythonTask(
+        name="branch3_task",
+        function=set_parallel_branch3,
+        returns=["result3"],
+    ).as_pipeline()
+
+    # Parallel node executes all branches
+    parallel = Parallel(
+        name="parallel",
+        branches={
+            "branch1": branch1_pipeline,
+            "branch2": branch2_pipeline,
+            "branch3": branch3_pipeline,
+        },
+    )
+
+    # Task to verify all parameters rolled back from branches
+    verify_task = PythonTask(
+        name="verify",
+        function=verify_parallel_rollback,
+        terminate_with_success=True,
+    )
+
+    pipeline = Pipeline(steps=[parallel, verify_task])
+    pipeline.execute()
+
+    return pipeline
+
+
+if __name__ == "__main__":
+    main()
@@ -225,3 +225,59 @@ async def async_process(result: str):
     processed = f"processed_{result}"
     print(f"Processed: {processed}")
     return processed
+
+
+# Helper functions for conditional parameter rollback examples
+def set_conditional_heads_param():
+    """Set parameter when heads branch executes."""
+    return "heads_value"
+
+
+def set_conditional_tails_param():
+    """Set parameter when tails branch executes."""
+    return "tails_value"
+
+
+def set_conditional_multiple():
+    """Set multiple parameters in conditional branch."""
+    return "param1_value", "param2_value", "param3_value"
+
+
+def verify_conditional_rollback(branch_param: str):
+    """Verify rolled back parameter from conditional branch."""
+    assert branch_param in ["heads_value", "tails_value"]
+    return branch_param
+
+
+# Helper functions for parallel parameter rollback examples
+def set_parallel_branch1():
+    """Set parameter in parallel branch 1."""
+    return "branch1_value"
+
+
+def set_parallel_branch2():
+    """Set parameter in parallel branch 2."""
+    return "branch2_value"
+
+
+def set_parallel_branch3():
+    """Set parameter in parallel branch 3."""
+    return "branch3_value"
+
+
+def verify_parallel_rollback(result1: str, result2: str, result3: str):
+    """Verify all parallel branch parameters rolled back."""
+    assert result1 == "branch1_value"
+    assert result2 == "branch2_value"
+    assert result3 == "branch3_value"
+    return "verified"
+
+
+def set_shared_param_a():
+    """Set shared parameter to value A."""
+    return "value_a"
+
+
+def set_shared_param_b():
+    """Set shared parameter to value B."""
+    return "value_b"
@@ -12,6 +12,7 @@
     utils,
 )
 from runnable.datastore import DataCatalog, JobLog, JsonParameter, StepAttempt
+from runnable.defaults import IterableParameterModel
 from runnable.executor import BaseJobExecutor
 from runnable.tasks import BaseTaskType
 
@@ -174,11 +175,11 @@ def _sync_catalog(
         return data_catalogs
 
     def add_task_log_to_catalog(
-        self, name: str, map_variable: Dict[str, str | int | float] | None = None
+        self, name: str, iter_variable: Optional[IterableParameterModel] = None
     ):
         log_file_name = utils.make_log_file_name(
             name=name,
-            map_variable=map_variable,
+            iter_variable=iter_variable,
         )
         task_console.save_text(log_file_name)
         # Put the log file in the catalog