From 0c0dcbce49445a8a3363ce70d4796b0e4b59195d Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 4 Sep 2026 21:20:46 +0700 Subject: [PATCH] feat: expose structured context during execution --- src/wf_core/runtime/ops/flow.py | 12 + src/wf_core/runtime/ops/foreach.py | 4 +- src/wf_core/runtime/ops/handlers.py | 4 +- src/wf_core/runtime/ops/nodes.py | 6 +- src/wf_core/runtime/subgraphs.py | 6 +- src/wf_core/validation/steps.py | 7 +- tests/core/test_structured_runtime_context.py | 408 +++++++++++++++++- 7 files changed, 436 insertions(+), 11 deletions(-) diff --git a/src/wf_core/runtime/ops/flow.py b/src/wf_core/runtime/ops/flow.py index 237d498d..ef6b644e 100644 --- a/src/wf_core/runtime/ops/flow.py +++ b/src/wf_core/runtime/ops/flow.py @@ -161,10 +161,22 @@ def _foreach_ancestor_ids(run: RunState, frame: ExecutionFrame) -> list[str]: def finalize_run(workflow: Workflow, run: RunState) -> RunState: if run.outcome is None: run.outcome = "ok" + # Root workflow output keeps standard root facts consistent by projecting + # against the root frame's derived context rather than an empty mapping. + from wf_core.run_state import ROOT_FRAME_ID + from wf_core.runtime.ops.frames import frame_context_view + + root_frame = run.frames.get(ROOT_FRAME_ID) + root_context: dict[str, Any] = ( + dict(frame_context_view(run, root_frame).graph) + if root_frame is not None + else {} + ) run.output = project_output( workflow, run.state, workflow_input=run.workflow_input, + context=root_context, ) validate_payload_against_schema( workflow.output_schema, run.output, "workflow output" diff --git a/src/wf_core/runtime/ops/foreach.py b/src/wf_core/runtime/ops/foreach.py index 9876dbf4..6e65e96f 100644 --- a/src/wf_core/runtime/ops/foreach.py +++ b/src/wf_core/runtime/ops/foreach.py @@ -23,7 +23,7 @@ from wf_core.runtime.lineage import ( scope_input_for_frame, ) from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace -from wf_core.runtime.ops.frames import frame_context_values +from wf_core.runtime.ops.frames import frame_context_view from wf_core.runtime.ops.index import WorkflowIndex from wf_core.runtime.ops.merges import ReducerDefinition from wf_core.runtime.ops.overlays import state_view_for_frame @@ -174,7 +174,7 @@ def _resolve_foreach_iterable( str(step.over), state=state_view_for_frame(run, frame), workflow_input=scope_input_for_frame(run, frame), - context=frame_context_values(frame), + context=frame_context_view(run, frame).graph, ) if not isinstance(iterable, list): raise WorkflowExecutionError( diff --git a/src/wf_core/runtime/ops/handlers.py b/src/wf_core/runtime/ops/handlers.py index 14422b18..1df0f77d 100644 --- a/src/wf_core/runtime/ops/handlers.py +++ b/src/wf_core/runtime/ops/handlers.py @@ -14,7 +14,7 @@ from wf_core.run_state import ( ) from wf_core.runtime.lineage import scope_input_for_frame from wf_core.runtime.ops.flow import append_trace -from wf_core.runtime.ops.frames import frame_context_values +from wf_core.runtime.ops.frames import frame_context_view from wf_core.runtime.ops.interrupts import build_interrupt_request from wf_core.runtime.ops.overlays import state_view_for_frame @@ -75,7 +75,7 @@ def handle_interrupt_step( frame_id=frame.id, state=state_view_for_frame(run, frame), workflow_input=scope_input_for_frame(run, frame), - context=frame_context_values(frame), + context=dict(frame_context_view(run, frame).graph), public_frame_id=public_frame.id, public_node_id=public_frame.node_id, route=route, diff --git a/src/wf_core/runtime/ops/nodes.py b/src/wf_core/runtime/ops/nodes.py index 199b4df8..9e02d7e4 100644 --- a/src/wf_core/runtime/ops/nodes.py +++ b/src/wf_core/runtime/ops/nodes.py @@ -20,7 +20,7 @@ from wf_core.runtime.lineage import ( commit_foreach_aware_patch, scope_input_for_frame, ) -from wf_core.runtime.ops.frames import frame_context_values +from wf_core.runtime.ops.frames import frame_context_view from wf_core.runtime.ops.merges import ReducerDefinition from wf_core.runtime.ops.overlays import state_view_for_frame from wf_core.runtime.ops.schemas import validate_payload_against_schema @@ -54,7 +54,8 @@ def _resolve_node_execution( node_def: NodeDef, platform: object | None = None, ) -> tuple[dict[str, Any], RuntimeContext, dict[str, Any]]: - context_values = frame_context_values(frame) + context_view = frame_context_view(run, frame) + context_values = context_view.graph state_view = state_view_for_frame(run, frame) resolved_input = resolve_step_input_bindings( node.input, @@ -76,6 +77,7 @@ def _resolve_node_execution( prior_outcome=frame.prior_outcome, activated_incoming_edge=frame.activated_incoming_edge, metadata=dict(frame.metadata), + foreach=dict(context_view.foreach), platform=platform, ) return resolved_input, context, state_view diff --git a/src/wf_core/runtime/subgraphs.py b/src/wf_core/runtime/subgraphs.py index 49daabb1..42c5bcff 100644 --- a/src/wf_core/runtime/subgraphs.py +++ b/src/wf_core/runtime/subgraphs.py @@ -18,7 +18,7 @@ from wf_core.run_state import ( ) from wf_core.runtime.input_bindings import resolve_step_input_bindings from wf_core.runtime.lineage import commit_foreach_aware_patch -from wf_core.runtime.ops.frames import frame_context_values +from wf_core.runtime.ops.frames import frame_context_view from wf_core.runtime.ops.merges import ReducerDefinition from wf_core.runtime.ops.overlays import state_view_for_frame from wf_core.runtime.ops.runs import initial_state @@ -140,7 +140,7 @@ def _start_subgraph( step.input, state=state_view_for_frame(run, frame), workflow_input=parent_scope.workflow_input, - context=frame_context_values(frame), + context=frame_context_view(run, frame).graph, label=f"subgraph {step.id!r}", ) validate_payload_against_schema( @@ -218,7 +218,7 @@ def _finish_subgraph( prepared.workflow, child_scope.committed_state, workflow_input=child_scope.workflow_input, - context=frame_context_values(child_frame), + context=frame_context_view(run, child_frame).graph, ) validate_payload_against_schema( prepared.workflow.output_schema, diff --git a/src/wf_core/validation/steps.py b/src/wf_core/validation/steps.py index 812cc81a..7e829fae 100644 --- a/src/wf_core/validation/steps.py +++ b/src/wf_core/validation/steps.py @@ -329,7 +329,12 @@ def validate_foreach_node( input_root_fields: set[str], workflow: Workflow, ) -> None: - if not is_valid_source_path(node.over, state_root_fields, input_root_fields): + # Interim permissive gate: context-rooted `over` paths reach runtime, where + # structured ancestry resolution handles them. Task 5 replaces this with + # location-aware validation against the consuming node's context schema. + if not is_valid_source_path( + node.over, state_root_fields, input_root_fields, allow_context=True + ): report.add( ValidationIssueCode.INVALID_FOREACH_SOURCE, f"nodes[{index}].over", diff --git a/tests/core/test_structured_runtime_context.py b/tests/core/test_structured_runtime_context.py index 59a64a79..e4c0dd07 100644 --- a/tests/core/test_structured_runtime_context.py +++ b/tests/core/test_structured_runtime_context.py @@ -3,7 +3,13 @@ from __future__ import annotations import pytest from wf_core.errors import WorkflowExecutionError -from wf_core.run_state import ExecutionFrame, ForeachContext, RunState, RunStatus +from wf_core.run_state import ( + ExecutionFrame, + ForeachContext, + RunState, + RunStatus, + RuntimeContext, +) from wf_core.runtime.ops.frames import frame_context_view @@ -299,3 +305,403 @@ def test_context_read_does_not_mutate_run_state() -> None: before = run.to_dict() frame_context_view(run, run.frames["outer-item"]) assert run.to_dict() == before + + +def _nested_workflow(*, inner_over: str = "state.orders_list"): + from wf_core import END, Edge, ForeachNode, NodeDef, NodeUse, SchemaRef, Workflow + from wf_core.models.schemas import StateField, StateSchema + + customers = ForeachNode.model_validate( + { + "id": "customers", + "type": "foreach", + "over": "state.customers", + "as": "customer", + "mode": "serial", + } + ) + orders = ForeachNode.model_validate( + { + "id": "orders", + "type": "foreach", + "over": inner_over, + "as": "order", + "mode": "serial", + } + ) + return Workflow( + name="nested_structured", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map( + { + "customers": StateField(type="array"), + "orders_list": StateField(type="array"), + } + ), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[ + NodeDef( + name="record", + input_schema=SchemaRef( + type="object", properties={"value": {}}, required=["value"] + ), + output_schema=SchemaRef(type="object", properties={}), + outcomes=["ok"], + ) + ], + start="customers", + nodes=[ + customers, + orders, + NodeUse.model_validate( + { + "id": "work", + "type": "node", + "node": "record", + "input": [{"target": "value", "path": "context.order"}], + "output": [], + } + ), + ], + edges=[ + Edge.model_validate( + {"from": "customers", "outcome": "loop", "to": "orders"} + ), + Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "ok", "to": "orders"}), + Edge.model_validate({"from": "orders", "outcome": "done", "to": "customers"}), + Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), + ], + ) + + +def test_nested_handler_receives_outer_and_inner_typed_entries() -> None: + from wf_core import execute_workflow + + seen: list[tuple[tuple[str, ...], object, object, int]] = [] + + def record(_payload: dict[str, object], ctx: RuntimeContext) -> dict[str, object]: + seen.append( + ( + tuple(ctx.foreach), + ctx.foreach["customers"].item, + ctx.foreach["orders"].item, + ctx.foreach["orders"].index, + ) + ) + return {"outcome": "ok", "output": {}} + + workflow = _nested_workflow() + run = execute_workflow( + workflow, + {"customers": [{"name": "Ada"}], "orders_list": [{"sku": "A-17"}]}, + {"record": record}, + ) + assert run.status == RunStatus.COMPLETED + assert seen == [ + (("customers", "orders"), {"name": "Ada"}, {"sku": "A-17"}, 0) + ] + + +def test_nested_graph_bindings_resolve_outer_and_inner_items() -> None: + from wf_core import ( + END, + Edge, + ForeachNode, + NodeDef, + NodeUse, + SchemaRef, + Workflow, + execute_workflow, + ) + from wf_core.models.schemas import StateField, StateSchema + + customers = ForeachNode.model_validate( + { + "id": "customers", + "type": "foreach", + "over": "state.customers", + "as": "customer", + "mode": "serial", + } + ) + orders = ForeachNode.model_validate( + { + "id": "orders", + "type": "foreach", + "over": "state.orders_list", + "as": "order", + "mode": "serial", + } + ) + workflow = Workflow( + name="nested_bindings", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map( + { + "customers": StateField(type="array"), + "orders_list": StateField(type="array"), + } + ), + output_schema=SchemaRef(type="object", properties={}), + node_defs=[ + NodeDef( + name="record", + input_schema=SchemaRef( + type="object", + properties={"outer": {}, "inner": {}}, + required=["outer", "inner"], + ), + output_schema=SchemaRef(type="object", properties={}), + outcomes=["ok"], + ) + ], + start="customers", + nodes=[ + customers, + orders, + NodeUse.model_validate( + { + "id": "work", + "type": "node", + "node": "record", + "input": [ + { + "target": "outer", + "path": "context.foreach.customers.item", + }, + { + "target": "inner", + "path": "context.foreach.orders.item", + }, + ], + "output": [], + } + ), + ], + edges=[ + Edge.model_validate( + {"from": "customers", "outcome": "loop", "to": "orders"} + ), + Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "ok", "to": "orders"}), + Edge.model_validate( + {"from": "orders", "outcome": "done", "to": "customers"} + ), + Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), + ], + ) + captured: list[dict[str, object]] = [] + + def record(payload: dict[str, object], _ctx: RuntimeContext) -> dict[str, object]: + captured.append(dict(payload)) + return {"outcome": "ok", "output": {}} + + run = execute_workflow( + workflow, + {"customers": [{"name": "Ada"}], "orders_list": [{"sku": "A-17"}]}, + {"record": record}, + ) + assert run.status == RunStatus.COMPLETED + assert captured == [{"outer": {"name": "Ada"}, "inner": {"sku": "A-17"}}] + + +def test_inner_completion_restores_outer_context() -> None: + from wf_core import ( + END, + Edge, + ForeachNode, + NodeDef, + NodeUse, + SchemaRef, + Workflow, + execute_workflow, + ) + from wf_core.models.schemas import StateField, StateSchema + + workflow = Workflow( + name="restore_outer", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map( + { + "customers": StateField(type="array"), + "orders_list": StateField(type="array"), + } + ), + 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="customers", + nodes=[ + ForeachNode.model_validate( + { + "id": "customers", + "type": "foreach", + "over": "state.customers", + "as": "customer", + "mode": "serial", + } + ), + ForeachNode.model_validate( + { + "id": "orders", + "type": "foreach", + "over": "state.orders_list", + "as": "order", + "mode": "serial", + } + ), + NodeUse.model_validate( + { + "id": "work", + "type": "node", + "node": "record", + "input": [{"target": "value", "path": "context.order"}], + "output": [], + } + ), + NodeUse.model_validate( + { + "id": "after_inner", + "type": "node", + "node": "record", + "input": [{"target": "value", "path": "context.customer"}], + "output": [], + } + ), + ], + edges=[ + Edge.model_validate( + {"from": "customers", "outcome": "loop", "to": "orders"} + ), + Edge.model_validate({"from": "orders", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "ok", "to": "orders"}), + Edge.model_validate( + {"from": "orders", "outcome": "done", "to": "after_inner"} + ), + Edge.model_validate( + {"from": "after_inner", "outcome": "ok", "to": "customers"} + ), + Edge.model_validate({"from": "customers", "outcome": "done", "to": END}), + ], + ) + keys: list[tuple[str, tuple[str, ...]]] = [] + + def record(_payload: dict[str, object], ctx: RuntimeContext) -> dict[str, object]: + keys.append((ctx.current_node_id, tuple(ctx.foreach))) + return {"outcome": "ok", "output": {}} + + run = execute_workflow( + workflow, + {"customers": [{"name": "Ada"}], "orders_list": [{"sku": "A-17"}]}, + {"record": record}, + ) + assert run.status == RunStatus.COMPLETED + by_node = {node: keys_tuple for node, keys_tuple in keys} + assert by_node["work"] == ("customers", "orders") + assert by_node["after_inner"] == ("customers",) + + +def test_concurrent_items_receive_distinct_frame_and_lineage_context() -> None: + from wf_core import ( + END, + Edge, + ForeachNode, + NodeDef, + NodeUse, + SchemaRef, + Workflow, + execute_workflow, + ) + from wf_core.models.schemas import StateField, StateSchema + + workflow = Workflow( + name="concurrent_ctx", + input_schema=SchemaRef(type="object", properties={}), + state_schema=StateSchema.from_field_map( + { + "items": StateField(type="array"), + } + ), + 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="each", + nodes=[ + ForeachNode.model_validate( + { + "id": "each", + "type": "foreach", + "over": "state.items", + "as": "item", + "mode": "concurrent", + "concurrent": {"max_active": 2, "max_outstanding": 2}, + } + ), + NodeUse.model_validate( + { + "id": "work", + "type": "node", + "node": "record", + "input": [{"target": "value", "path": "context.item"}], + "output": [], + } + ), + ], + edges=[ + Edge.model_validate({"from": "each", "outcome": "loop", "to": "work"}), + Edge.model_validate({"from": "work", "outcome": "ok", "to": "each"}), + Edge.model_validate({"from": "each", "outcome": "done", "to": END}), + ], + ) + contexts: list[ForeachContext] = [] + + def record(_payload: dict[str, object], ctx: RuntimeContext) -> dict[str, object]: + contexts.append(ctx.foreach["each"]) + return {"outcome": "ok", "output": {}} + + run = execute_workflow( + workflow, {"items": ["a", "b"]}, {"record": record} + ) + assert run.status == RunStatus.COMPLETED + assert len(contexts) == 2 + assert contexts[0].activation_id == contexts[1].activation_id + assert contexts[0].frame_id != contexts[1].frame_id + assert contexts[0].lineage_id != contexts[1].lineage_id + assert sorted([c.item for c in contexts]) == ["a", "b"] + + +def test_nested_foreach_over_resolves_structured_outer_item_path() -> None: + from wf_core import execute_workflow + + workflow = _nested_workflow( + inner_over="context.foreach.customers.item.orders" + ) + captured: list[object] = [] + + def record(payload: dict[str, object], _ctx: RuntimeContext) -> dict[str, object]: + captured.append(payload["value"]) + return {"outcome": "ok", "output": {}} + + run = execute_workflow( + workflow, + { + "customers": [{"name": "Ada", "orders": [{"sku": "A-17"}]}], + "orders_list": [], + }, + {"record": record}, + ) + assert run.status == RunStatus.COMPLETED + assert captured == [{"sku": "A-17"}]