diff --git a/src/wf_core/__init__.py b/src/wf_core/__init__.py index f1769402..2915214e 100644 --- a/src/wf_core/__init__.py +++ b/src/wf_core/__init__.py @@ -35,6 +35,7 @@ from .models import ( from .run_codec import PersistedRunState, dump_run_state, load_run_state from .run_state import ( ExecutionFrame, + ForeachContext, FrameStatus, InterruptRequest, InterruptRoute, @@ -76,6 +77,7 @@ __all__ = [ "Edge", "EndNode", "ExecutionFrame", + "ForeachContext", "ForeachConcurrentPolicy", "ForeachItemErrorPolicy", "ForeachNode", diff --git a/src/wf_core/context_contracts.py b/src/wf_core/context_contracts.py index 62845819..11063000 100644 --- a/src/wf_core/context_contracts.py +++ b/src/wf_core/context_contracts.py @@ -13,6 +13,7 @@ LINEAGE_ID_CONTEXT_KEY = "lineage_id" PARENT_LINEAGE_ID_CONTEXT_KEY = "parent_lineage_id" LOOP_ITEM_CONTEXT_KEY = "loop_item" LOOP_INDEX_CONTEXT_KEY = "loop_index" +FOREACH_CONTEXT_KEY = "foreach" @dataclass(frozen=True, slots=True) @@ -57,6 +58,7 @@ STANDARD_CONTEXT_FIELD_NAMES = frozenset( RESERVED_CONTEXT_KEYS = STANDARD_CONTEXT_FIELD_NAMES | { LOOP_ITEM_CONTEXT_KEY, LOOP_INDEX_CONTEXT_KEY, + FOREACH_CONTEXT_KEY, } diff --git a/src/wf_core/run_state.py b/src/wf_core/run_state.py index 9aeb5918..e3283ed8 100644 --- a/src/wf_core/run_state.py +++ b/src/wf_core/run_state.py @@ -86,6 +86,25 @@ class ExecutionFrame: finished_at_node_id: str | None = None +@dataclass(frozen=True, slots=True) +class ForeachContext: + """Typed view of one active same-scope foreach iteration. + + The mapping key in ``RuntimeContext.foreach`` is the static + ``ForeachNode.id`` within the current workflow scope. Identity fields + disclose the dynamic activation/frame/scope/lineage when advanced + runtime code needs them; normal node authors should use mapped inputs. + """ + + node_id: str + activation_id: str + frame_id: str + scope_id: str + lineage_id: str + index: int + item: Any + + @dataclass(slots=True) class RuntimeContext: current_node_id: str @@ -98,6 +117,7 @@ class RuntimeContext: activated_incoming_edge: str | None = None metadata: dict[str, Any] = field(default_factory=dict) platform: object | None = None + foreach: dict[str, ForeachContext] = field(default_factory=dict) @dataclass(slots=True) diff --git a/src/wf_core/runtime/ops/frames.py b/src/wf_core/runtime/ops/frames.py index 22422927..6a658cbf 100644 --- a/src/wf_core/runtime/ops/frames.py +++ b/src/wf_core/runtime/ops/frames.py @@ -1,7 +1,11 @@ from __future__ import annotations +from collections.abc import Mapping +from dataclasses import dataclass + from wf_core.context_contracts import ( ACTIVATED_INCOMING_EDGE_CONTEXT_KEY, + FOREACH_CONTEXT_KEY, LINEAGE_ID_CONTEXT_KEY, LOOP_INDEX_CONTEXT_KEY, LOOP_ITEM_CONTEXT_KEY, @@ -10,7 +14,8 @@ from wf_core.context_contracts import ( RESERVED_CONTEXT_KEYS, SCOPE_ID_CONTEXT_KEY, ) -from wf_core.run_state import ExecutionFrame +from wf_core.errors import WorkflowExecutionError +from wf_core.run_state import ExecutionFrame, ForeachContext, RunState def frame_context_values(frame: ExecutionFrame) -> dict[str, object | None]: @@ -34,3 +39,114 @@ def frame_context_values(frame: ExecutionFrame) -> dict[str, object | None]: ): context[loop_alias] = loop_item return context + + +@dataclass(frozen=True, slots=True) +class FrameContextView: + """Typed handler context and graph values from one ancestry walk.""" + + foreach: Mapping[str, ForeachContext] + graph: Mapping[str, object | None] + + +def frame_context_view(run: RunState, frame: ExecutionFrame) -> FrameContextView: + """Derive structured foreach context from persisted frame ancestry. + + The walk starts at the selected frame and follows ``parent_frame_id`` + while ancestors remain in the same ``scope_id``. Each foreach item frame + contributes one typed entry from its validated metadata. Frame ancestry + describes scheduling ownership, so traversal stops at a runtime-scope + boundary even though a subgraph root frame has a scheduling parent in + the caller: subgraphs receive caller values only through declared input + bindings. + + The walk is fail-closed: malformed foreach item metadata, a missing + parent frame, a parent cycle, duplicate active foreach ids, and empty, + reserved, or duplicated active aliases all raise + ``WorkflowExecutionError``. Corrupt persisted state is not equivalent to + a missing context value. ``RunState`` is never mutated while reading. + """ + # Local import avoids a cycle: scheduler owns typed foreach metadata on + # top of run_state, while this module owns context derivation. + from wf_core.runtime.scheduler import ForeachIterationMetadata + + selected_scope_id = frame.scope_id + current: ExecutionFrame | None = frame + seen: set[str] = set() + # Collected innermost-first; reversed into outermost-to-innermost order. + inner_to_outer: list[tuple[str, ForeachContext, str]] = [] + + while current is not None and current.scope_id == selected_scope_id: + if current.id in seen: + raise WorkflowExecutionError( + f"cyclic execution frame ancestry at frame {current.id!r}" + ) + seen.add(current.id) + # Fail-closed typed decode: corrupt persisted item metadata raises + # here rather than surfacing as a missing context value. + metadata = ForeachIterationMetadata.from_frame(current) + if metadata is not None: + inner_to_outer.append( + ( + metadata.foreach_node_id, + metadata.to_context(current), + metadata.loop_alias, + ) + ) + if current.parent_frame_id is None: + break + parent = run.frames.get(current.parent_frame_id) + if parent is None: + raise WorkflowExecutionError( + f"missing parent frame {current.parent_frame_id!r} " + f"for frame {current.id!r}" + ) + current = parent + + foreach: dict[str, ForeachContext] = {} + alias_by_node_id: dict[str, str] = {} + seen_aliases: set[str] = set() + for node_id, entry, alias in reversed(inner_to_outer): + if node_id in foreach: + raise WorkflowExecutionError( + f"duplicate active foreach id {node_id!r} for frame {frame.id!r}" + ) + if not alias or alias in RESERVED_CONTEXT_KEYS: + raise WorkflowExecutionError( + f"foreach alias {alias!r} for node {node_id!r} collides with " + f"reserved context keys for frame {frame.id!r}" + ) + if alias in seen_aliases: + raise WorkflowExecutionError( + f"duplicate active foreach alias {alias!r} for frame {frame.id!r}" + ) + foreach[node_id] = entry + alias_by_node_id[node_id] = alias + seen_aliases.add(alias) + + graph: dict[str, object | None] = { + PRIOR_OUTCOME_CONTEXT_KEY: frame.prior_outcome, + ACTIVATED_INCOMING_EDGE_CONTEXT_KEY: frame.activated_incoming_edge, + SCOPE_ID_CONTEXT_KEY: frame.scope_id, + LINEAGE_ID_CONTEXT_KEY: frame.lineage_id, + PARENT_LINEAGE_ID_CONTEXT_KEY: frame.parent_lineage_id, + } + graph[FOREACH_CONTEXT_KEY] = { + node_id: { + "node_id": entry.node_id, + "activation_id": entry.activation_id, + "frame_id": entry.frame_id, + "scope_id": entry.scope_id, + "lineage_id": entry.lineage_id, + "index": entry.index, + "item": entry.item, + } + for node_id, entry in foreach.items() + } + for node_id, entry in foreach.items(): + graph[alias_by_node_id[node_id]] = entry.item + if foreach: + innermost = next(reversed(foreach.values())) + graph[LOOP_ITEM_CONTEXT_KEY] = innermost.item + graph[LOOP_INDEX_CONTEXT_KEY] = innermost.index + return FrameContextView(foreach=foreach, graph=graph) diff --git a/src/wf_core/runtime/scheduler.py b/src/wf_core/runtime/scheduler.py index 35cc8648..badd1df4 100644 --- a/src/wf_core/runtime/scheduler.py +++ b/src/wf_core/runtime/scheduler.py @@ -5,7 +5,13 @@ from dataclasses import dataclass from typing import Any from wf_core.errors import WorkflowExecutionError -from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, RunStatus +from wf_core.run_state import ( + ExecutionFrame, + ForeachContext, + FrameStatus, + RunState, + RunStatus, +) @dataclass(slots=True, frozen=True) @@ -97,6 +103,23 @@ class ForeachIterationMetadata: "loop_alias": self.loop_alias, } + def to_context(self, frame: ExecutionFrame) -> ForeachContext: + """Convert validated item metadata into the typed runtime context value. + + Field names are mapped once here so ancestry-traversal call sites do + not copy them. ``frame`` supplies the dynamic scope/lineage/frame + identities for the entry. + """ + return ForeachContext( + node_id=self.foreach_node_id, + activation_id=self.activation_id, + frame_id=frame.id, + scope_id=frame.scope_id, + lineage_id=frame.lineage_id, + index=self.loop_index, + item=self.loop_item, + ) + def add_frame(run: RunState, frame: ExecutionFrame, *, ready: bool = False) -> None: """Add a frame once; frame id reuse is always a runtime invariant error.""" diff --git a/tests/core/test_structured_runtime_context.py b/tests/core/test_structured_runtime_context.py new file mode 100644 index 00000000..59a64a79 --- /dev/null +++ b/tests/core/test_structured_runtime_context.py @@ -0,0 +1,301 @@ +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.runtime.ops.frames import frame_context_view + + +def _run_with_frames(frames: list[ExecutionFrame]) -> RunState: + run = RunState( + workflow_name="demo", + status=RunStatus.RUNNING, + workflow_input={}, + state={}, + ) + for frame in frames: + run.frames[frame.id] = frame + return run + + +def _item_frame( + *, + frame_id: str, + parent_id: str | None, + node_id: str, + activation_id: str, + index: int, + item: object, + alias: str, + scope_id: str = "root", + lineage_id: str = "lineage", +) -> ExecutionFrame: + return ExecutionFrame( + id=frame_id, + kind="foreach_iteration", + node_id=node_id, + parent_frame_id=parent_id, + scope_id=scope_id, + lineage_id=lineage_id, + metadata={ + "foreach_node_id": node_id, + "activation_id": activation_id, + "loop_index": index, + "loop_item": item, + "loop_alias": alias, + }, + ) + + +def test_root_frame_has_empty_structured_foreach_context() -> None: + run = _run_with_frames( + [ExecutionFrame(id="root", kind="root", node_id="start", scope_id="root")] + ) + view = frame_context_view(run, run.frames["root"]) + assert dict(view.foreach) == {} + assert view.graph["foreach"] == {} + assert "loop_item" not in view.graph + assert "loop_index" not in view.graph + + +def test_nested_same_scope_frames_expose_outermost_to_innermost_context() -> None: + run = _run_with_frames( + [ + ExecutionFrame(id="root", kind="root", node_id="customers", scope_id="root"), + _item_frame( + frame_id="outer-item", + parent_id="root", + node_id="customers", + activation_id="customers:activation:1", + index=0, + item={"name": "Ada"}, + alias="customer", + lineage_id="customers:lineage:0", + ), + ExecutionFrame( + id="inner-controller", + kind="foreach", + node_id="orders", + parent_frame_id="outer-item", + scope_id="root", + lineage_id="customers:lineage:0", + ), + _item_frame( + frame_id="inner-item", + parent_id="inner-controller", + node_id="orders", + activation_id="orders:activation:1", + index=2, + item={"sku": "A-17"}, + alias="order", + lineage_id="orders:lineage:2", + ), + ] + ) + view = frame_context_view(run, run.frames["inner-item"]) + contexts = view.foreach + + assert tuple(contexts) == ("customers", "orders") + assert contexts["customers"] == ForeachContext( + node_id="customers", + activation_id="customers:activation:1", + frame_id="outer-item", + scope_id="root", + lineage_id="customers:lineage:0", + index=0, + item={"name": "Ada"}, + ) + assert contexts["orders"].index == 2 + assert contexts["orders"].item == {"sku": "A-17"} + + +def test_graph_context_values_keep_all_aliases_and_innermost_loop_keys() -> None: + run = _run_with_frames( + [ + ExecutionFrame(id="root", kind="root", node_id="customers", scope_id="root"), + _item_frame( + frame_id="outer-item", + parent_id="root", + node_id="customers", + activation_id="customers:activation:1", + index=0, + item={"name": "Ada"}, + alias="customer", + lineage_id="customers:lineage:0", + ), + _item_frame( + frame_id="inner-item", + parent_id="outer-item", + node_id="orders", + activation_id="orders:activation:1", + index=2, + item={"sku": "A-17"}, + alias="order", + lineage_id="orders:lineage:2", + ), + ] + ) + view = frame_context_view(run, run.frames["inner-item"]) + graph = dict(view.graph) + assert graph["customer"] == {"name": "Ada"} + assert graph["order"] == {"sku": "A-17"} + assert graph["loop_item"] == {"sku": "A-17"} + assert graph["loop_index"] == 2 + foreach_map = graph["foreach"] + assert isinstance(foreach_map, dict) + assert set(foreach_map) == {"customers", "orders"} + assert foreach_map["orders"]["index"] == 2 + assert foreach_map["orders"]["item"] == {"sku": "A-17"} + + +def test_context_ancestry_stops_at_runtime_scope_boundary() -> None: + run = _run_with_frames( + [ + ExecutionFrame(id="root", kind="root", node_id="customers", scope_id="root"), + _item_frame( + frame_id="outer-item", + parent_id="root", + node_id="customers", + activation_id="customers:activation:1", + index=0, + item={"name": "Ada"}, + alias="customer", + lineage_id="customers:lineage:0", + ), + ExecutionFrame( + id="child-root", + kind="subgraph_root", + node_id="start", + parent_frame_id="outer-item", + scope_id="child", + lineage_id="child:root", + ), + ] + ) + view = frame_context_view(run, run.frames["child-root"]) + assert dict(view.foreach) == {} + assert view.graph["foreach"] == {} + + +def test_structured_context_rejects_malformed_foreach_metadata() -> None: + run = _run_with_frames( + [ + ExecutionFrame( + id="bad", + kind="foreach_iteration", + node_id="body", + scope_id="root", + metadata={"foreach_node_id": "", "activation_id": "a"}, + ) + ] + ) + with pytest.raises(WorkflowExecutionError, match="malformed|missing"): + frame_context_view(run, run.frames["bad"]) + + +def test_structured_context_rejects_missing_parent_frame() -> None: + run = _run_with_frames( + [ + ExecutionFrame( + id="child", + kind="node", + node_id="body", + parent_frame_id="missing", + scope_id="root", + ) + ] + ) + with pytest.raises(WorkflowExecutionError, match="missing parent frame"): + frame_context_view(run, run.frames["child"]) + + +def test_structured_context_rejects_parent_cycle() -> None: + run = _run_with_frames( + [ + ExecutionFrame( + id="a", kind="node", node_id="x", parent_frame_id="b", scope_id="root" + ), + ExecutionFrame( + id="b", kind="node", node_id="y", parent_frame_id="a", scope_id="root" + ), + ] + ) + with pytest.raises(WorkflowExecutionError, match="cyclic"): + frame_context_view(run, run.frames["a"]) + + +def test_structured_context_rejects_duplicate_active_foreach_id() -> None: + run = _run_with_frames( + [ + ExecutionFrame(id="root", kind="root", node_id="each", scope_id="root"), + _item_frame( + frame_id="outer-item", + parent_id="root", + node_id="each", + activation_id="act-1", + index=0, + item="a", + alias="first", + ), + _item_frame( + frame_id="inner-item", + parent_id="outer-item", + node_id="each", + activation_id="act-2", + index=1, + item="b", + alias="second", + ), + ] + ) + with pytest.raises(WorkflowExecutionError, match="duplicate active foreach id"): + frame_context_view(run, run.frames["inner-item"]) + + +def test_structured_context_rejects_duplicate_active_alias() -> None: + run = _run_with_frames( + [ + ExecutionFrame(id="root", kind="root", node_id="a", scope_id="root"), + _item_frame( + frame_id="outer-item", + parent_id="root", + node_id="customers", + activation_id="act-1", + index=0, + item="a", + alias="same", + ), + _item_frame( + frame_id="inner-item", + parent_id="outer-item", + node_id="orders", + activation_id="act-2", + index=0, + item="b", + alias="same", + ), + ] + ) + with pytest.raises(WorkflowExecutionError, match="duplicate active foreach alias"): + frame_context_view(run, run.frames["inner-item"]) + + +def test_context_read_does_not_mutate_run_state() -> None: + run = _run_with_frames( + [ + ExecutionFrame(id="root", kind="root", node_id="customers", scope_id="root"), + _item_frame( + frame_id="outer-item", + parent_id="root", + node_id="customers", + activation_id="act-1", + index=0, + item={"name": "Ada"}, + alias="customer", + ), + ] + ) + before = run.to_dict() + frame_context_view(run, run.frames["outer-item"]) + assert run.to_dict() == before