frame New Primitive

readme.md

@@ -320,6 +320,12 @@ At minimum, run state should track:
 
 This gives the engine a clean path toward checkpointing, interrupt, and resume later.
 
+Run state should also carry execution frames.
+
+- v1 may only have a root workflow frame
+- future `foreach` and subgraph execution should attach work to child frames
+- interrupts should belong to a frame, not just to the run globally
+
 Two useful execution entry points fall out of this:
 
 - `step_workflow(...)` for one-node advancement

wf_core/__init__.py

@@ -19,7 +19,14 @@ from .runtime import (
     resume_workflow,
     step_workflow,
 )
-from .run_state import InterruptRequest, RunState, RunStatus, RuntimeContext, TraceEntry
+from .run_state import (
+    ExecutionFrame,
+    InterruptRequest,
+    RunState,
+    RunStatus,
+    RuntimeContext,
+    TraceEntry,
+)
 from .tokens import END, START
 from .validate import (
     ValidationIssue,
@@ -40,6 +47,7 @@ __all__ = [
     "StateField",
     "StateSchema",
     "NodeHandler",
+    "ExecutionFrame",
     "RunState",
     "RunStatus",
     "RuntimeContext",

wf_core/run_state.py

@@ -13,9 +13,21 @@ class RunStatus(StrEnum):
     INTERRUPTED = "interrupted"
 
 
+@dataclass(slots=True)
+class ExecutionFrame:
+    id: str
+    kind: str
+    node_id: str
+    parent_frame_id: str | None = None
+    prior_outcome: str | None = None
+    activated_incoming_edge: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
 @dataclass(slots=True)
 class RuntimeContext:
     current_node_id: str
+    frame_id: str = "root"
     retry_count: int = 0
     prior_outcome: str | None = None
     activated_incoming_edge: str | None = None
@@ -23,6 +35,7 @@ class RuntimeContext:
 
 @dataclass(slots=True)
 class TraceEntry:
+    frame_id: str
     node_id: str
     step_type: str
     resolved_input: dict[str, Any]
@@ -35,6 +48,7 @@ class TraceEntry:
 @dataclass(slots=True)
 class InterruptRequest:
     id: str
+    frame_id: str
     node_id: str
     kind: str
     payload: dict[str, Any] = field(default_factory=dict)
@@ -49,11 +63,24 @@ class RunState:
     state: dict[str, Any]
     output: dict[str, Any] = field(default_factory=dict)
     trace: list[TraceEntry] = field(default_factory=list)
+    frames: dict[str, ExecutionFrame] = field(default_factory=dict)
+    current_frame_id: str | None = None
     current_node_id: str | None = None
     prior_outcome: str | None = None
     activated_incoming_edge: str | None = None
     error: str | None = None
     interrupt: InterruptRequest | None = None
 
+    def current_frame(self) -> ExecutionFrame:
+        if self.current_frame_id is None:
+            raise ValueError("run has no current frame")
+        return self.frames[self.current_frame_id]
+
+    def sync_from_current_frame(self) -> None:
+        frame = self.current_frame()
+        self.current_node_id = frame.node_id
+        self.prior_outcome = frame.prior_outcome
+        self.activated_incoming_edge = frame.activated_incoming_edge
+
     def to_dict(self) -> dict[str, Any]:
         return asdict(self)

wf_core/runtime.py

@@ -15,7 +15,14 @@ from .model import (
     NodeUse,
     Workflow,
 )
-from .run_state import InterruptRequest, RunState, RunStatus, RuntimeContext, TraceEntry
+from .run_state import (
+    ExecutionFrame,
+    InterruptRequest,
+    RunState,
+    RunStatus,
+    RuntimeContext,
+    TraceEntry,
+)
 from .schema_tools import validate_payload_against_schema
 from .state_ops import apply_mapped_state, apply_output_map, project_output
 from .tokens import END
@@ -34,8 +41,17 @@ def execute_workflow(
         status=RunStatus.PENDING,
         workflow_input=dict(workflow_input),
         state=dict(workflow_input),
+        frames={
+            "root": ExecutionFrame(
+                id="root",
+                kind="workflow",
+                node_id=workflow.start,
+            )
+        },
+        current_frame_id="root",
         current_node_id=workflow.start,
     )
+    run.sync_from_current_frame()
 
     try:
         workflow.validate_structure().raise_for_errors()
@@ -62,6 +78,10 @@ def resume_workflow(
             f"run state belongs to workflow {run.workflow_name!r}, not {workflow.name!r}"
         )
 
+    if run.current_frame_id is None:
+        raise WorkflowExecutionError("run has no current frame")
+    run.sync_from_current_frame()
+
     if run.current_node_id is None:
         raise WorkflowExecutionError("run has no current node")
 
@@ -124,6 +144,10 @@ def step_workflow(
     nodes_by_id: dict[str, Any] | None = None,
     edge_map: dict[tuple[str, str], str] | None = None,
 ) -> RunState:
+    if run.current_frame_id is None:
+        raise WorkflowExecutionError("run has no current frame")
+
+    run.sync_from_current_frame()
     if run.current_node_id is None or run.current_node_id == END:
         return run
     if run.status == RunStatus.INTERRUPTED:
@@ -141,7 +165,8 @@ def step_workflow(
         (edge.from_, edge.outcome): edge.to for edge in workflow.edges
     }
 
-    step = nodes_by_id[run.current_node_id]
+    frame = run.current_frame()
+    step = nodes_by_id[frame.node_id]
 
     if isinstance(step, NodeUse):
         node_def = node_defs[step.node]
@@ -149,7 +174,7 @@ def step_workflow(
         outcome = step_result["outcome"]
     elif isinstance(step, ConditionNode):
         predicate = eval_condition(
-            step.check, run.state, run.workflow_input, run.prior_outcome
+            step.check, run.state, run.workflow_input, frame.prior_outcome
         )
         outcome = "true" if predicate else "false"
         step_result = {
@@ -167,18 +192,20 @@ def step_workflow(
     elif isinstance(step, InterruptNode):
         interrupt_request = _build_interrupt_request(
             step,
-            run.state,
+            frame_id=frame.id,
-            run.workflow_input,
+            state=run.state,
+            workflow_input=run.workflow_input,
         )
         run.interrupt = interrupt_request
         run.status = RunStatus.INTERRUPTED
         run.trace.append(
             TraceEntry(
-                node_id=run.current_node_id,
+                frame_id=frame.id,
+                node_id=frame.node_id,
                 step_type=step.type,
                 resolved_input=interrupt_request.payload,
                 outcome="interrupt",
-                next_node_id=run.current_node_id,
+                next_node_id=frame.node_id,
                 output=interrupt_request.payload,
                 state_changes={},
             )
@@ -189,15 +216,16 @@ def step_workflow(
     else:
         raise WorkflowExecutionError(f"unsupported step type {step.type!r}")
 
-    next_node_id = edge_map.get((run.current_node_id, outcome))
+    next_node_id = edge_map.get((frame.node_id, outcome))
     if next_node_id is None:
         raise WorkflowExecutionError(
-            f"no edge found for node {run.current_node_id!r} and outcome {outcome!r}"
+            f"no edge found for node {frame.node_id!r} and outcome {outcome!r}"
         )
 
     run.trace.append(
         TraceEntry(
-            node_id=run.current_node_id,
+            frame_id=frame.id,
+            node_id=frame.node_id,
             step_type=step.type,
             resolved_input=step_result["resolved_input"],
             outcome=outcome,
@@ -207,9 +235,10 @@ def step_workflow(
         )
     )
 
-    run.prior_outcome = outcome
+    frame.prior_outcome = outcome
-    run.activated_incoming_edge = run.current_node_id
+    frame.activated_incoming_edge = frame.node_id
-    run.current_node_id = next_node_id
+    frame.node_id = next_node_id
+    run.sync_from_current_frame()
     return run
 
 
@@ -239,10 +268,12 @@ def _execute_node_use(
         node_def.input_schema, resolved_input, f"node input for {node.id}"
     )
 
+    frame = run.current_frame()
     context = RuntimeContext(
         current_node_id=node.id,
-        prior_outcome=run.prior_outcome,
+        frame_id=frame.id,
-        activated_incoming_edge=run.activated_incoming_edge,
+        prior_outcome=frame.prior_outcome,
+        activated_incoming_edge=frame.activated_incoming_edge,
     )
     raw_result = handler(resolved_input, context)
     result = coerce_node_result(raw_result)
@@ -274,6 +305,8 @@ def coerce_node_result(raw_result: NodeResult | dict[str, Any]) -> NodeResult:
 
 def _build_interrupt_request(
     node: InterruptNode,
+    *,
+    frame_id: str,
     state: dict[str, Any],
     workflow_input: dict[str, Any],
 ) -> InterruptRequest:
@@ -288,6 +321,7 @@ def _build_interrupt_request(
     }
     return InterruptRequest(
         id=f"interrupt:{node.id}",
+        frame_id=frame_id,
         node_id=node.id,
         kind=node.kind,
         payload=payload,
@@ -303,12 +337,15 @@ def _resume_interrupt(
     resume_payload: dict[str, Any],
     resume_outcome: str,
 ) -> None:
+    if run.current_frame_id is None:
+        raise WorkflowExecutionError("interrupted run has no current frame")
     if run.current_node_id is None:
         raise WorkflowExecutionError("interrupted run has no current node")
     if run.interrupt is None:
         raise WorkflowExecutionError("run is interrupted but has no interrupt request")
 
-    step = nodes_by_id[run.current_node_id]
+    frame = run.current_frame()
+    step = nodes_by_id[frame.node_id]
     if not isinstance(step, InterruptNode):
         raise WorkflowExecutionError(
             f"interrupted run expected interrupt node, got {step.type!r}"
@@ -325,15 +362,16 @@ def _resume_interrupt(
         run.state,
         missing_field_message="interrupt resume payload is missing required field {field}",
     )
-    next_node_id = edge_map.get((run.current_node_id, resume_outcome))
+    next_node_id = edge_map.get((frame.node_id, resume_outcome))
     if next_node_id is None:
         raise WorkflowExecutionError(
-            f"no edge found for interrupt node {run.current_node_id!r} and outcome {resume_outcome!r}"
+            f"no edge found for interrupt node {frame.node_id!r} and outcome {resume_outcome!r}"
         )
 
     run.trace.append(
         TraceEntry(
-            node_id=run.current_node_id,
+            frame_id=frame.id,
+            node_id=frame.node_id,
             step_type=step.type,
             resolved_input=resume_payload,
             outcome=resume_outcome,
@@ -342,7 +380,8 @@ def _resume_interrupt(
             state_changes=state_changes,
         )
     )
-    run.prior_outcome = resume_outcome
+    frame.prior_outcome = resume_outcome
-    run.activated_incoming_edge = run.current_node_id
+    frame.activated_incoming_edge = frame.node_id
-    run.current_node_id = next_node_id
+    frame.node_id = next_node_id
     run.interrupt = None
+    run.sync_from_current_frame()