feat: derive structured foreach runtime context

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",

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,
 }
 
 

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)

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)

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."""

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