From 3bfb3f440bcc4244639c9a53544f7e7c62f339f7 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 23:30:51 +0700 Subject: [PATCH] fix: evaluate conditions against structured frame context --- src/wf_core/conditions.py | 32 +++- src/wf_core/runtime/ops/handlers.py | 3 + tests/core/test_structured_runtime_context.py | 149 ++++++++++++++++++ 3 files changed, 178 insertions(+), 6 deletions(-) diff --git a/src/wf_core/conditions.py b/src/wf_core/conditions.py index fa70d310..b2b26b36 100644 --- a/src/wf_core/conditions.py +++ b/src/wf_core/conditions.py @@ -21,25 +21,43 @@ def eval_condition( state: Mapping[str, Any], workflow_input: Mapping[str, Any], context_data: str | None, + *, + context: Mapping[str, Any] | None = None, ) -> bool: + """Evaluate a condition against state, input, and context. + + ``context`` is the frame's structured context graph (see + ``frame_context_view``); when omitted, evaluation falls back to the + legacy ``{"prior_outcome": context_data}`` stub so direct unit callers + keep working. The graph always carries ``prior_outcome``, so passing it + preserves legacy behavior while also resolving structured paths such as + ``context.foreach..item``. + """ + resolved = context if context is not None else {"prior_outcome": context_data} if isinstance(condition, ExistsCondition): return path_exists( condition.path, state=state, workflow_input=workflow_input, - context={"prior_outcome": context_data}, + context=resolved, ) if isinstance(condition, NotCondition): - return not eval_condition(condition.arg, state, workflow_input, context_data) + return not eval_condition( + condition.arg, state, workflow_input, context_data, context=resolved + ) if isinstance(condition, VariadicCondition): values = [ - eval_condition(arg, state, workflow_input, context_data) + eval_condition(arg, state, workflow_input, context_data, context=resolved) for arg in condition.args ] return all(values) if condition.op == "and" else any(values) if isinstance(condition, BinaryCondition): - left = resolve_operand(condition.left, state, workflow_input, context_data) - right = resolve_operand(condition.right, state, workflow_input, context_data) + left = resolve_operand( + condition.left, state, workflow_input, context_data, context=resolved + ) + right = resolve_operand( + condition.right, state, workflow_input, context_data, context=resolved + ) if condition.op == "eq": return left == right if condition.op == "ne": @@ -60,6 +78,8 @@ def resolve_operand( state: Mapping[str, Any], workflow_input: Mapping[str, Any], context_data: str | None, + *, + context: Mapping[str, Any] | None = None, ) -> Any: if isinstance(operand, LiteralOperand): return operand.value @@ -67,7 +87,7 @@ def resolve_operand( str(operand.path), state=state, workflow_input=workflow_input, - context={"prior_outcome": context_data}, + context=context if context is not None else {"prior_outcome": context_data}, ) diff --git a/src/wf_core/runtime/ops/handlers.py b/src/wf_core/runtime/ops/handlers.py index 1df0f77d..89e9216d 100644 --- a/src/wf_core/runtime/ops/handlers.py +++ b/src/wf_core/runtime/ops/handlers.py @@ -29,6 +29,9 @@ def handle_condition_step( state_view_for_frame(run, frame), scope_input_for_frame(run, frame), frame.prior_outcome, + # Structured context must match what validation blesses: conditions + # inside a foreach body can read context.foreach.* / loop aliases. + context=frame_context_view(run, frame).graph, ) outcome = "true" if predicate else "false" return StepExecutionResult( diff --git a/tests/core/test_structured_runtime_context.py b/tests/core/test_structured_runtime_context.py index 188395e2..bc797e49 100644 --- a/tests/core/test_structured_runtime_context.py +++ b/tests/core/test_structured_runtime_context.py @@ -923,3 +923,152 @@ def test_bool_loop_index_metadata_fails_closed() -> None: ) with pytest.raises(WorkflowExecutionError, match="malformed foreach loop index"): frame_context_view(run, run.frames["bad"]) + + +def test_condition_exists_reads_structured_foreach_item() -> None: + from wf_core import END, Edge, ForeachNode, Workflow, execute_workflow + from wf_core.models.schemas import SchemaRef, StateField, StateSchema + from wf_core.models.steps import ConditionNode + + pick = ConditionNode.model_validate( + { + "id": "pick", + "type": "condition", + "check": {"op": "exists", "path": "context.foreach.each.item"}, + } + ) + workflow = Workflow( + name="condition_structured_exists", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map({"items": StateField(type="array")}), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[], + start="each", + nodes=[ + ForeachNode.model_validate( + { + "id": "each", + "type": "foreach", + "over": "state.items", + "as": "item", + "mode": "serial", + } + ), + pick, + ], + edges=[ + Edge.model_validate({"from": "each", "outcome": "loop", "to": "pick"}), + Edge.model_validate({"from": "pick", "outcome": "true", "to": "each"}), + Edge.model_validate({"from": "pick", "outcome": "false", "to": "each"}), + Edge.model_validate({"from": "each", "outcome": "done", "to": END}), + ], + ) + run = execute_workflow(workflow, {"items": ["a"]}, {}) + assert run.status == RunStatus.COMPLETED + # The item exists, so the validated structured path must take true. + assert ("pick", "true") in [(t.node_id, t.outcome) for t in run.trace] + + +def test_condition_eq_reads_loop_alias_per_item() -> None: + from wf_core import END, Edge, ForeachNode, Workflow, execute_workflow + from wf_core.models.schemas import SchemaRef, StateField, StateSchema + from wf_core.models.steps import ConditionNode + + pick = ConditionNode.model_validate( + { + "id": "pick", + "type": "condition", + "check": { + "op": "eq", + "left": {"path": "context.item"}, + "right": {"value": "a"}, + }, + } + ) + workflow = Workflow( + name="condition_structured_eq", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map({"items": StateField(type="array")}), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[], + start="each", + nodes=[ + ForeachNode.model_validate( + { + "id": "each", + "type": "foreach", + "over": "state.items", + "as": "item", + "mode": "serial", + } + ), + pick, + ], + edges=[ + Edge.model_validate({"from": "each", "outcome": "loop", "to": "pick"}), + Edge.model_validate({"from": "pick", "outcome": "true", "to": "each"}), + Edge.model_validate({"from": "pick", "outcome": "false", "to": "each"}), + Edge.model_validate({"from": "each", "outcome": "done", "to": END}), + ], + ) + run = execute_workflow(workflow, {"items": ["a", "b"]}, {}) + assert run.status == RunStatus.COMPLETED + pick_outcomes = [t.outcome for t in run.trace if t.node_id == "pick"] + assert pick_outcomes == ["true", "false"] + + +def test_condition_still_reads_prior_outcome() -> None: + from wf_core import END, Edge, NodeDef, NodeUse, Workflow, execute_workflow + from wf_core.models.schemas import SchemaRef, StateSchema + from wf_core.models.steps import ConditionNode + + pick = ConditionNode.model_validate( + { + "id": "pick", + "type": "condition", + "check": { + "op": "eq", + "left": {"path": "context.prior_outcome"}, + "right": {"value": "ok"}, + }, + } + ) + workflow = Workflow( + name="condition_prior_outcome", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema(fields={}), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[ + NodeDef( + name="record", + input_schema=SchemaRef(type="object", properties={"value": {}}), + output_schema=SchemaRef(type="object", properties={}), + outcomes=["ok"], + ) + ], + start="work", + nodes=[ + NodeUse.model_validate( + { + "id": "work", + "type": "node", + "node": "record", + "input": [{"target": "value", "value": 1}], + "output": [], + } + ), + pick, + ], + edges=[ + Edge.model_validate({"from": "work", "outcome": "ok", "to": "pick"}), + Edge.model_validate({"from": "pick", "outcome": "true", "to": END}), + Edge.model_validate({"from": "pick", "outcome": "false", "to": END}), + ], + ) + + def record(payload: dict[str, object], ctx: RuntimeContext) -> dict[str, object]: + return {"outcome": "ok", "output": {}} + + run = execute_workflow(workflow, {}, {"record": record}) + assert run.status == RunStatus.COMPLETED + assert ("pick", "true") in [(t.node_id, t.outcome) for t in run.trace]