child graphs can interrupt/resume

This commit is contained in:
lda
2026-05-25 05:11:16 +07:00 Verified
parent 7477a12b51
commit 7aa562c619
14 changed files with 397 additions and 95 deletions
+7 -5
View File
@@ -52,8 +52,10 @@ implementation state.
reference conversion helpers. Core can now execute a prepared local child reference conversion helpers. Core can now execute a prepared local child
workflow through an isolated child scope/lineage, preserve its trace entries, workflow through an isolated child scope/lineage, preserve its trace entries,
map child output through the boundary, and route by the child's terminal map child output through the boundary, and route by the child's terminal
outcome. Interrupt bubbling/resume and saved/deployed child resolution remain outcome. Prepared child interrupts now bubble through a typed internal route
next. Wrapper helpers currently run child workflows as ordinary nodes; native and resume inside child scope while the public request identifies the parent
subgraph boundary. Saved/deployed child resolution remains next. Wrapper
helpers currently run child workflows as ordinary nodes; native
`SubgraphNode` is now the graph-as-node path for prepared children. `SubgraphNode` is now the graph-as-node path for prepared children.
- **Concurrent foreach**: implemented in core with explicit scheduling, - **Concurrent foreach**: implemented in core with explicit scheduling,
reducer/merge semantics, item error policy, async handler batching, and reducer/merge semantics, item error policy, async handler batching, and
@@ -64,9 +66,9 @@ implementation state.
`RuntimeScope` / `LineageState` storage, scope-aware reads, and non-root write `RuntimeScope` / `LineageState` storage, scope-aware reads, and non-root write
buffering. New concurrent foreach item writes are stored in buffering. New concurrent foreach item writes are stored in
`RunState.lineages`, while `ForeachBarrierState` keeps scheduling/result `RunState.lineages`, while `ForeachBarrierState` keeps scheduling/result
metadata and compatibility patches. Current direct commits are still metadata and compatibility patches. Scope-root commits now apply to both the
root-frame-only via an explicit helper; native subgraph completion should root workflow and prepared native child scopes through the explicit
replace that shortcut with an explicit scope/lineage commit target. scope/lineage commit helper.
- **Persistent run history**: add a run store before adding stable `run_id`, - **Persistent run history**: add a run store before adding stable `run_id`,
`inspect_run`, or `read_run_trace(run_id, range)` APIs. Current traces are `inspect_run`, or `read_run_trace(run_id, range)` APIs. Current traces are
returned directly from immediate run responses. returned directly from immediate run responses.
@@ -0,0 +1,105 @@
# Native Subgraph Interrupt Resume Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Let a prepared native child workflow interrupt its parent run and later resume inside the original child scope.
**Architecture:** Add a typed internal interrupt route that distinguishes the public parent-subgraph identity from the actual interrupted child frame. Make interrupt request/resume operations scope-aware, then have engine resume select the prepared child workflow and reducers before applying child resume bindings. The parent `SubgraphNode` remains blocked until the child reaches its ordinary terminal outcome.
**Tech Stack:** Python 3.14, dataclasses, Pydantic workflow models, pytest, Ruff, basedpyright.
---
### Task 1: Child Interrupt Contract
**Files:**
- Modify: `src/wf_core/run_state.py`
- Modify: `tests/core/test_subgraph_step.py`
- [x] **Step 1: Write failing child interrupt/resume tests**
Add a child workflow containing an `InterruptNode` followed by a node or terminal step. Execute it through a parent `SubgraphNode` and assert:
```python
assert run.status == RunStatus.INTERRUPTED
assert run.interrupt is not None
assert run.interrupt.node_id == "child"
assert run.interrupt.payload["question"] == "confirm?"
assert run.frames["root"].status == FrameStatus.BLOCKED
resumed = resume_workflow(
parent,
run,
{},
resume_payload={"answer": "yes"},
subgraphs={"child.workflow": prepared_child},
)
assert resumed.status == RunStatus.COMPLETED
assert resumed.output["answer"] == "yes"
```
- [x] **Step 2: Run tests and observe the existing explicit rejection**
Run: `uv run pytest -q tests/core/test_subgraph_step.py`
Expected: FAIL because child `InterruptNode` execution currently raises that child interrupts are unsupported.
- [x] **Step 3: Add structural route state**
Add `InterruptRoute` to `run_state.py` containing the interrupted child
`frame_id`, `node_id`, `scope_id`, `lineage_id`, and `workflow_ref`. Add an
optional `route` field to `InterruptRequest`; root interrupt requests continue
to use `route=None`.
### Task 2: Scope-Aware Interrupt Creation and Resume
**Files:**
- Modify: `src/wf_core/runtime/ops/handlers.py`
- Modify: `src/wf_core/runtime/ops/interrupts.py`
- Modify: `src/wf_core/runtime/preparation.py`
- Modify: `src/wf_core/runtime/step.py`
- Modify: `src/wf_core/runtime/engine.py`
- [x] **Step 1: Permit child interrupts and build their payload from child scope**
Remove the explicit child rejection. Build child interrupt request bindings
from `state_view_for_frame(...)` and `scope_input_for_frame(...)`, not from the
root compatibility dictionaries. For a non-root scope, find the owning parent
subgraph frame for public identity and attach `InterruptRoute` for resume.
- [x] **Step 2: Resume through the routed child workflow**
When an interrupted request has `route`, restore the routed child as the
current frame, resolve its `PreparedSubgraph`, build the child workflow index,
and apply `resume` output bindings into the child scope using normal
scope-aware patch commit logic. Root interrupts retain the existing path.
- [x] **Step 3: Verify parent completion behavior**
After child resume, scheduling must continue child execution first. Only after
the child finishes may the blocked parent subgraph frame wake and map child
output into parent state.
### Task 3: Verification and Documentation
**Files:**
- Modify: `docs/wf_core_architecture.md`
- Modify: `docs/current_roadmap.md`
- [x] **Step 1: Update documented limitations**
Document that prepared child interrupts now bubble and resume locally, while
artifact/deployment resolution for nested children remains outside core.
- [x] **Step 2: Run focused verification**
Run:
```powershell
uv run pytest -q tests/core/test_subgraph_step.py tests/core/test_concurrent_foreach_interrupts.py
uvx ruff check src/wf_core tests/core/test_subgraph_step.py tests/core/test_concurrent_foreach_interrupts.py
uvx ruff format --check src/wf_core tests/core/test_subgraph_step.py tests/core/test_concurrent_foreach_interrupts.py
uv run basedpyright --level error src/wf_core tests/core/test_subgraph_step.py tests/core/test_concurrent_foreach_interrupts.py
```
Expected: all commands exit successfully.
@@ -1,6 +1,6 @@
# Native Subgraphs Design # Native Subgraphs Design
Status: prepared-child execution implemented; interrupts/artifact resolution planned Status: prepared-child execution and interrupt resume implemented; artifact resolution planned
Native subgraphs should make a workflow usable as a workflow step without Native subgraphs should make a workflow usable as a workflow step without
collapsing the child run into one opaque Python node call. The current collapsing the child run into one opaque Python node call. The current
@@ -8,9 +8,9 @@ collapsing the child run into one opaque Python node call. The current
compatibility wrappers, but they hide the child trace, child frames, and child compatibility wrappers, but they hide the child trace, child frames, and child
interrupt lifecycle from `wf_core`. interrupt lifecycle from `wf_core`.
This design defines the core runtime shape. The boundary model and This design defines the core runtime shape. The boundary model, prepared-child
non-interrupting prepared-child execution are implemented; interruption and execution, and routed child interrupt resume are implemented; saved-workflow
saved-workflow resolution remain planned. resolution remains planned.
## Goals ## Goals
@@ -81,7 +81,8 @@ Runtime execution now accepts caller-supplied `PreparedSubgraph` dependencies
for local workflow refs. It creates a child scope/lineage, schedules child for local workflow refs. It creates a child scope/lineage, schedules child
frames in the parent run, retains child trace entries, and applies mapped frames in the parent run, retains child trace entries, and applies mapped
output only at boundary completion. Saved artifact refs are not loaded by output only at boundary completion. Saved artifact refs are not loaded by
`wf_core`, and child interrupts fail explicitly until resume routing exists. `wf_core`. Prepared child interrupts bubble through a typed internal route and
resume inside their original child scope.
`WorkflowRef` should be structural, not a dotted string parser: `WorkflowRef` should be structural, not a dotted string parser:
@@ -228,15 +229,16 @@ If a child frame reaches an `InterruptNode`:
4. `RunState.interrupt` points to the child interrupt, with enough route data 4. `RunState.interrupt` points to the child interrupt, with enough route data
to resume into the child. to resume into the child.
The parent-facing interrupt request should include: The client-facing interrupt identity should remain the parent subgraph
boundary: its frame/step identify the reusable graph node that asked for
interaction, while `kind` and `payload` describe what the caller must supply.
The runtime must separately retain the actual child route required for resume:
- parent frame id - parent frame id and subgraph parent step id for the public request identity
- subgraph parent step id
- child workflow reference - child workflow reference
- child frame id - child scope and lineage ids
- child interrupt node id - interrupted child frame id and interrupt node id
- interrupt kind - interrupt kind and payload
- payload
Current `InterruptRequest` only has `id`, `frame_id`, `node_id`, `kind`, Current `InterruptRequest` only has `id`, `frame_id`, `node_id`, `kind`,
`payload`, and `resumable`. Native subgraphs need either: `payload`, and `resumable`. Native subgraphs need either:
@@ -245,9 +247,11 @@ Current `InterruptRequest` only has `id`, `frame_id`, `node_id`, `kind`,
`lineage_id`, `parent_frame_id`, and `workflow_ref`, or `lineage_id`, `parent_frame_id`, and `workflow_ref`, or
- a typed nested route object, such as `InterruptRoute`. - a typed nested route object, such as `InterruptRoute`.
The preferred direction is explicit route structure. A generic metadata field The preferred direction is a typed `InterruptRoute` stored by
would recreate the ad hoc frame metadata problem, while string parsing is `InterruptRequest`. A generic metadata field would recreate the ad hoc frame
exactly what the project has been moving away from. metadata problem, while string parsing is exactly what the project has been
moving away from. Root-workflow interrupts may omit the route and retain their
existing direct frame/node identity.
## Resume Semantics ## Resume Semantics
@@ -393,7 +397,7 @@ selection out of `wf_core`.
- Artifact conversion helpers bridge saved workflow identities to core - Artifact conversion helpers bridge saved workflow identities to core
`WorkflowRef` values. `WorkflowRef` values.
### Completed Slice 1: Non-Interrupting Prepared Subgraph Runtime ### Completed Slice 1: Prepared Subgraph Runtime
- Local/prepared child `WorkflowRef` dependencies resolve through - Local/prepared child `WorkflowRef` dependencies resolve through
`PreparedSubgraph`; `wf_core` does not load saved artifacts. `PreparedSubgraph`; `wf_core` does not load saved artifacts.
@@ -404,9 +408,10 @@ selection out of `wf_core`.
records the parent `subgraph` trace entry. records the parent `subgraph` trace entry.
- Child output maps to parent state through existing output binding machinery. - Child output maps to parent state through existing output binding machinery.
- The parent step routes through the child's terminal workflow outcome. - The parent step routes through the child's terminal workflow outcome.
- Child interrupts reject explicitly until structural resume routing exists. - Child interrupts route structurally and preserve the blocked parent boundary
until the child resumes and completes.
### Slice 2: Interrupt Bubbling and Resume ### Completed Slice 2: Interrupt Bubbling and Resume
- Extend `InterruptRequest` with explicit route structure. - Extend `InterruptRequest` with explicit route structure.
- Bubble child interrupts to the parent run. - Bubble child interrupts to the parent run.
+16 -14
View File
@@ -131,29 +131,31 @@ limits and intended adapter seam.
strategies beyond exact-path mergeable reducers. strategies beyond exact-path mergeable reducers.
- Interrupt lifecycle is still node-level and run-state-level. Long-lived - Interrupt lifecycle is still node-level and run-state-level. Long-lived
external subscriptions or notification streams need a separate lifecycle external subscriptions or notification streams need a separate lifecycle
design. Interrupt `request` and `resume` are canonical binding lists; nested design. Interrupt `request` and `resume` are canonical binding lists;
child-workflow resume is still future work. prepared child workflows retain a typed internal interrupt route so resume
continues in child scope while exposing the parent subgraph boundary to the
caller.
- Native subgraphs use `SubgraphNode` plus caller-supplied `PreparedSubgraph` - Native subgraphs use `SubgraphNode` plus caller-supplied `PreparedSubgraph`
dependencies. A prepared local child executes through a child runtime scope dependencies. A prepared local child executes through a child runtime scope
and lineage; child output commits only through declared boundary bindings and and lineage; child output commits only through declared boundary bindings and
the parent routes by the child's terminal workflow outcome. Saved/deployed the parent routes by the child's terminal workflow outcome. Saved/deployed
workflow resolution remains outside core and is not implemented at this workflow resolution remains outside core and is not implemented at this
boundary yet. boundary yet.
- Nested subgraph interruption is not first-class yet. The current - The current `wf_authoring` wrapper helpers still run child workflows as
`wf_authoring` subgraph helpers wrap a child workflow as an ordinary sync or ordinary sync or async nodes and therefore do not preserve native child
async node and validate the child output; they do not preserve a child run state or resumable interrupts. Native `SubgraphNode` plus
state that can interrupt, bubble to the parent, and later resume inside the `PreparedSubgraph` is the first-class path: prepared child interrupts now
child. Native `SubgraphNode` preserves prepared-child execution and trace, bubble to the parent run and resume inside the original child scope. See
but rejects child interrupts until route-aware resume is implemented. See `examples/authoring_workflow_as_node.py` for the compatibility wrapper shape
`examples/authoring_workflow_as_node.py` for the wrapper-node shape. and `examples/authoring_native_subgraph.py` for the native path.
- Saved workflow-as-node execution with interrupts requires a core runtime - Saved workflow-as-node execution with interrupts still requires platform
upgrade: nested run state, child-frame trace preservation, interrupt bubbling resolution of artifact/deployment references into prepared child
with path metadata, and resume back into the child workflow. dependencies before core execution begins.
- Frames are no longer only a serial execution stack: the runtime has a ready - Frames are no longer only a serial execution stack: the runtime has a ready
queue, `BLOCKED` frame state, lineage isolation, barrier merge semantics, and queue, `BLOCKED` frame state, lineage isolation, barrier merge semantics, and
pending child results for concurrent foreach. Native prepared subgraphs now pending child results for concurrent foreach. Native prepared subgraphs now
use child-scope execution; saved/deployed child resolution and nested use child-scope execution and typed routed child interruption; saved/deployed
interruption remain outstanding. child resolution remains outstanding.
Concurrent foreach is the primary current use case for async concurrent node Concurrent foreach is the primary current use case for async concurrent node
handler execution. handler execution.
- Runtime errors are still ordinary exceptions plus failed run status. A richer - Runtime errors are still ordinary exceptions plus failed run status. A richer
+14 -12
View File
@@ -648,20 +648,22 @@ document, `WorkflowCapabilityRef` is the public callable capability name, and
The first implementation should prefer artifact validation and dependency The first implementation should prefer artifact validation and dependency
diagnostics before attempting persistent nested resume. diagnostics before attempting persistent nested resume.
Native subgraphs now have a core `SubgraphNode` placeholder. It validates the Native subgraphs now have a core `SubgraphNode` boundary. It validates the
parent-side contract: child workflow reference, declared child input/output parent-side contract: child workflow reference, declared child input/output
schemas, binding lists, and declared outcomes. Runtime execution is still not schemas, binding lists, and declared outcomes. When callers resolve a local
implemented; reaching a subgraph step raises a clear runtime error. The current child into `PreparedSubgraph`, core executes it in child scope, preserves its
`wf_authoring.subgraph_node` and `async_subgraph_node` helpers still execute a trace, and can bubble and resume child interrupts without exposing child state
child workflow as a plain node and validate the child output. The async helper as parent state. The current `wf_authoring.subgraph_node` and
is explicit because hiding `asyncio.run()` inside the sync wrapper would break `async_subgraph_node` helpers still execute a child workflow as a plain node
inside already-running event loops. Future saved-workflow-as-node execution and validate the child output. The async helper is explicit because hiding
needs a real child run state if child interrupts should pause the parent and `asyncio.run()` inside the sync wrapper would break inside already-running
later resume the child. event loops. Saved workflow-as-node execution still needs platform-level
artifact/deployment resolution into prepared children before core can run it.
See `examples/authoring_workflow_as_node.py` for the current wrapper-node See `examples/authoring_workflow_as_node.py` for the compatibility wrapper-node
approach. In that example the parent trace sees one node call; the child approach and `examples/authoring_native_subgraph.py` for native prepared-child
workflow's internal trace is not embedded in the parent run state. execution. In the wrapper example the parent trace sees one node call; in the
native example child trace entries remain in the parent run state.
Until that core upgrade exists, artifact tooling must not assume that an Until that core upgrade exists, artifact tooling must not assume that an
interrupting saved workflow can safely be used as a child node. Top-level saved interrupting saved workflow can safely be used as a child node. Top-level saved
+2
View File
@@ -37,6 +37,7 @@ from .runtime import (
from .run_state import ( from .run_state import (
ExecutionFrame, ExecutionFrame,
FrameStatus, FrameStatus,
InterruptRoute,
InterruptRequest, InterruptRequest,
RunState, RunState,
RunStatus, RunStatus,
@@ -81,6 +82,7 @@ __all__ = [
"RuntimeContext", "RuntimeContext",
"StepExecutionResult", "StepExecutionResult",
"TraceEntry", "TraceEntry",
"InterruptRoute",
"InterruptRequest", "InterruptRequest",
"START", "START",
"END", "END",
+2 -2
View File
@@ -161,8 +161,8 @@ class SubgraphNode(BaseModel):
"""Workflow boundary step for native prepared-child execution. """Workflow boundary step for native prepared-child execution.
The runtime can execute an already-prepared local child graph through a The runtime can execute an already-prepared local child graph through a
child scope/lineage and commit only its mapped boundary output. Resolving child scope/lineage, bubble and resume child interrupts, and commit only
saved artifacts and resuming child interrupts remain platform/runtime work. its mapped boundary output. Resolving saved artifacts remains platform work.
""" """
id: str id: str
+18
View File
@@ -119,6 +119,23 @@ class StepExecutionResult:
state_changes: dict[str, Any] = field(default_factory=dict) state_changes: dict[str, Any] = field(default_factory=dict)
@dataclass(slots=True)
class InterruptRoute:
"""Internal resume route for an interrupt raised below a graph boundary.
`InterruptRequest.frame_id` and `.node_id` may describe the public parent
subgraph boundary. This route retains the actual interrupted child frame so
resume can continue inside its original workflow scope.
"""
frame_id: str
node_id: str
scope_id: str
lineage_id: str
parent_frame_id: str
workflow_ref: WorkflowRef
@dataclass(slots=True) @dataclass(slots=True)
class InterruptRequest: class InterruptRequest:
id: str id: str
@@ -127,6 +144,7 @@ class InterruptRequest:
kind: str kind: str
payload: dict[str, Any] = field(default_factory=dict) payload: dict[str, Any] = field(default_factory=dict)
resumable: bool = True resumable: bool = True
route: InterruptRoute | None = None
@dataclass(slots=True) @dataclass(slots=True)
+25
View File
@@ -81,12 +81,17 @@ def resume_workflow(
subgraphs: Mapping[str, PreparedSubgraph[NodeHandler]] | None = None, subgraphs: Mapping[str, PreparedSubgraph[NodeHandler]] | None = None,
) -> RunState: ) -> RunState:
"""Resume a synchronous run from its current state.""" """Resume a synchronous run from its current state."""
interrupted_workflow, interrupted_reducers = _interrupt_resume_target(
workflow, reducers, run, subgraphs, resuming=resume_payload is not None
)
index = prepare_resume( index = prepare_resume(
workflow, workflow,
run, run,
resume_payload=resume_payload, resume_payload=resume_payload,
resume_outcome=resume_outcome, resume_outcome=resume_outcome,
reducers=reducers, reducers=reducers,
interrupted_workflow=interrupted_workflow,
interrupted_reducers=interrupted_reducers,
) )
if index is None: if index is None:
if run.current_node_id == END: if run.current_node_id == END:
@@ -128,12 +133,17 @@ async def resume_workflow_async(
subgraphs: Mapping[str, PreparedSubgraph[AsyncNodeHandler]] | None = None, subgraphs: Mapping[str, PreparedSubgraph[AsyncNodeHandler]] | None = None,
) -> RunState: ) -> RunState:
"""Resume an async run from its current state.""" """Resume an async run from its current state."""
interrupted_workflow, interrupted_reducers = _interrupt_resume_target(
workflow, reducers, run, subgraphs, resuming=resume_payload is not None
)
index = prepare_resume( index = prepare_resume(
workflow, workflow,
run, run,
resume_payload=resume_payload, resume_payload=resume_payload,
resume_outcome=resume_outcome, resume_outcome=resume_outcome,
reducers=reducers, reducers=reducers,
interrupted_workflow=interrupted_workflow,
interrupted_reducers=interrupted_reducers,
) )
if index is None: if index is None:
if run.current_node_id == END: if run.current_node_id == END:
@@ -164,6 +174,21 @@ async def resume_workflow_async(
return finalize_run(workflow, run) return finalize_run(workflow, run)
def _interrupt_resume_target(
root_workflow: Workflow,
root_reducers: Mapping[str, ReducerDefinition] | None,
run: RunState,
subgraphs: Mapping[str, PreparedSubgraph[Any]] | None,
*,
resuming: bool,
) -> tuple[Workflow | None, Mapping[str, ReducerDefinition] | None]:
"""Resolve the workflow that owns an outstanding routed child interrupt."""
if not resuming or run.interrupt is None or run.interrupt.route is None:
return None, root_reducers
child = resolve_prepared_subgraph(run.interrupt.route.workflow_ref, subgraphs)
return child.workflow, child.reducers
def _sync_execution_target( def _sync_execution_target(
root_workflow: Workflow, root_workflow: Workflow,
root_registry: Mapping[str, NodeHandler], root_registry: Mapping[str, NodeHandler],
+50 -3
View File
@@ -1,8 +1,17 @@
from __future__ import annotations from __future__ import annotations
from wf_core.conditions import eval_condition from wf_core.conditions import eval_condition
from wf_core.errors import WorkflowExecutionError
from wf_core.models.steps import ConditionNode, InterruptNode from wf_core.models.steps import ConditionNode, InterruptNode
from wf_core.run_state import FrameStatus, RunState, RunStatus, StepExecutionResult from wf_core.run_state import (
ROOT_SCOPE_ID,
ExecutionFrame,
FrameStatus,
InterruptRoute,
RunState,
RunStatus,
StepExecutionResult,
)
from wf_core.runtime.lineage import scope_input_for_frame from wf_core.runtime.lineage import scope_input_for_frame
from wf_core.runtime.ops.flow import append_trace 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_values
@@ -44,12 +53,32 @@ def handle_interrupt_step(
step: InterruptNode, step: InterruptNode,
) -> RunState: ) -> RunState:
frame = run.current_frame() frame = run.current_frame()
public_frame = frame
route = None
if frame.scope_id != ROOT_SCOPE_ID:
public_frame = _owning_subgraph_frame(run, frame)
scope = run.scopes.get(frame.scope_id)
if scope is None or scope.workflow_ref is None:
raise WorkflowExecutionError(
f"child interrupt frame {frame.id!r} has no workflow scope"
)
route = InterruptRoute(
frame_id=frame.id,
node_id=frame.node_id,
scope_id=frame.scope_id,
lineage_id=frame.lineage_id,
parent_frame_id=public_frame.id,
workflow_ref=scope.workflow_ref,
)
interrupt_request = build_interrupt_request( interrupt_request = build_interrupt_request(
step, step,
frame_id=frame.id, frame_id=frame.id,
state=run.state, state=state_view_for_frame(run, frame),
workflow_input=run.workflow_input, workflow_input=scope_input_for_frame(run, frame),
context=frame_context_values(frame), context=frame_context_values(frame),
public_frame_id=public_frame.id,
public_node_id=public_frame.node_id,
route=route,
) )
run.interrupt = interrupt_request run.interrupt = interrupt_request
run.status = RunStatus.INTERRUPTED run.status = RunStatus.INTERRUPTED
@@ -66,3 +95,21 @@ def handle_interrupt_step(
state_changes={}, state_changes={},
) )
return run return run
def _owning_subgraph_frame(run: RunState, frame: ExecutionFrame) -> ExecutionFrame:
"""Return the graph-boundary frame that owns one child-scope interrupt."""
cursor = frame
while cursor.parent_frame_id is not None:
parent = run.frames.get(cursor.parent_frame_id)
if parent is None:
raise WorkflowExecutionError(
f"child interrupt frame {frame.id!r} references missing parent "
f"{cursor.parent_frame_id!r}"
)
if parent.scope_id != frame.scope_id:
return parent
cursor = parent
raise WorkflowExecutionError(
f"child interrupt frame {frame.id!r} has no parent subgraph"
)
+39 -12
View File
@@ -8,11 +8,19 @@ from wf_core.errors import WorkflowExecutionError
from wf_core.local_paths import LocalPathError, set_local_value from wf_core.local_paths import LocalPathError, set_local_value
from wf_core.models.steps import InputPathBinding, InputValueBinding, InterruptNode from wf_core.models.steps import InputPathBinding, InputValueBinding, InterruptNode
from wf_core.models.workflow import Workflow from wf_core.models.workflow import Workflow
from wf_core.run_state import InterruptRequest, RunState, StepExecutionResult from wf_core.run_state import (
FrameStatus,
InterruptRequest,
InterruptRoute,
RunState,
StepExecutionResult,
)
from wf_core.runtime.lineage import commit_patch_for_frame
from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace from wf_core.runtime.ops.flow import advance_frame, append_step_result_trace
from wf_core.runtime.ops.index import WorkflowIndex from wf_core.runtime.ops.index import WorkflowIndex
from wf_core.runtime.ops.merges import ReducerDefinition from wf_core.runtime.ops.merges import ReducerDefinition
from wf_core.runtime.ops.state import apply_output_bindings from wf_core.runtime.ops.overlays import state_view_for_frame
from wf_core.runtime.ops.state import build_output_patch
def build_interrupt_request( def build_interrupt_request(
@@ -22,6 +30,9 @@ def build_interrupt_request(
state: dict[str, Any], state: dict[str, Any],
workflow_input: dict[str, Any], workflow_input: dict[str, Any],
context: dict[str, Any], context: dict[str, Any],
public_frame_id: str | None = None,
public_node_id: str | None = None,
route: InterruptRoute | None = None,
) -> InterruptRequest: ) -> InterruptRequest:
payload: dict[str, Any] = {} payload: dict[str, Any] = {}
for binding in node.request: for binding in node.request:
@@ -43,11 +54,12 @@ def build_interrupt_request(
except LocalPathError as exc: except LocalPathError as exc:
raise WorkflowExecutionError(str(exc)) from exc raise WorkflowExecutionError(str(exc)) from exc
return InterruptRequest( return InterruptRequest(
id=f"interrupt:{node.id}", id=f"interrupt:{public_node_id or node.id}",
frame_id=frame_id, frame_id=public_frame_id or frame_id,
node_id=node.id, node_id=public_node_id or node.id,
kind=node.kind, kind=node.kind,
payload=payload, payload=payload,
route=route,
) )
@@ -60,14 +72,28 @@ def resume_interrupt(
resume_outcome: str, resume_outcome: str,
reducers: Mapping[str, ReducerDefinition] | None = None, reducers: Mapping[str, ReducerDefinition] | None = None,
) -> None: ) -> 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: if run.interrupt is None:
raise WorkflowExecutionError("run is interrupted but has no interrupt request") raise WorkflowExecutionError("run is interrupted but has no interrupt request")
frame = run.current_frame() route = run.interrupt.route
if route is not None:
frame = run.frames.get(route.frame_id)
if (
frame is None
or frame.scope_id != route.scope_id
or frame.lineage_id != route.lineage_id
or frame.node_id != route.node_id
or frame.status != FrameStatus.INTERRUPTED
):
raise WorkflowExecutionError("child interrupt route is no longer resumable")
run.current_frame_id = frame.id
run.sync_from_current_frame()
else:
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")
frame = run.current_frame()
step = index.nodes_by_id[frame.node_id] step = index.nodes_by_id[frame.node_id]
if not isinstance(step, InterruptNode): if not isinstance(step, InterruptNode):
raise WorkflowExecutionError( raise WorkflowExecutionError(
@@ -78,14 +104,15 @@ def resume_interrupt(
f"interrupt node {step.id!r} does not declare resume outcome {resume_outcome!r}" f"interrupt node {step.id!r} does not declare resume outcome {resume_outcome!r}"
) )
state_changes = apply_output_bindings( patch = build_output_patch(
workflow, workflow,
step.resume, step.resume,
resume_payload, resume_payload,
run.state, state_view_for_frame(run, frame),
reducers=reducers, reducers=reducers,
missing_field_message="interrupt resume payload is missing required field {field}", missing_field_message="interrupt resume payload is missing required field {field}",
) )
state_changes = commit_patch_for_frame(run, frame, patch)
next_node_id = index.next_node_id(frame.node_id, resume_outcome) next_node_id = index.next_node_id(frame.node_id, resume_outcome)
append_step_result_trace( append_step_result_trace(
run, run,
+13 -3
View File
@@ -34,6 +34,8 @@ def prepare_resume(
resume_payload: dict[str, Any] | None, resume_payload: dict[str, Any] | None,
resume_outcome: str, resume_outcome: str,
reducers: Mapping[str, ReducerDefinition] | None = None, reducers: Mapping[str, ReducerDefinition] | None = None,
interrupted_workflow: Workflow | None = None,
interrupted_reducers: Mapping[str, ReducerDefinition] | None = None,
) -> WorkflowIndex | None: ) -> WorkflowIndex | None:
"""Validate and normalize a run state before resume execution.""" """Validate and normalize a run state before resume execution."""
if run.workflow_name != workflow.name: if run.workflow_name != workflow.name:
@@ -55,13 +57,21 @@ def prepare_resume(
if run.status == RunStatus.INTERRUPTED: if run.status == RunStatus.INTERRUPTED:
if resume_payload is None: if resume_payload is None:
return None return None
resume_workflow = interrupted_workflow or workflow
resume_index = (
index
if resume_workflow is workflow
else build_workflow_index(resume_workflow)
)
resume_interrupt( resume_interrupt(
workflow, resume_workflow,
run, run,
index=index, index=resume_index,
resume_payload=resume_payload, resume_payload=resume_payload,
resume_outcome=resume_outcome, resume_outcome=resume_outcome,
reducers=reducers, reducers=(
reducers if interrupted_workflow is None else interrupted_reducers
),
) )
if run.current_frame_id is not None: if run.current_frame_id is not None:
frame = run.current_frame() frame = run.current_frame()
-9
View File
@@ -40,7 +40,6 @@ from wf_core.runtime.scheduler import (
) )
from wf_core.runtime.subgraphs import PreparedSubgraph, step_subgraph from wf_core.runtime.subgraphs import PreparedSubgraph, step_subgraph
from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, StepExecutionResult from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, StepExecutionResult
from wf_core.run_state import ROOT_SCOPE_ID
from wf_core.tokens import END from wf_core.tokens import END
from .preparation import prepare_step from .preparation import prepare_step
@@ -155,10 +154,6 @@ def step_workflow(
outcome=step.outcome, outcome=step.outcome,
) )
elif isinstance(step, InterruptNode): elif isinstance(step, InterruptNode):
if frame.kind == "subgraph_root" or frame.scope_id != ROOT_SCOPE_ID:
raise WorkflowExecutionError(
"child interrupts are not supported until native subgraph resume routing exists"
)
return handle_interrupt_step(run, step) return handle_interrupt_step(run, step)
elif isinstance(step, ForeachNode): elif isinstance(step, ForeachNode):
return step_foreach(workflow, run, step, index, reducers=reducers) return step_foreach(workflow, run, step, index, reducers=reducers)
@@ -271,10 +266,6 @@ async def step_workflow_async(
outcome=step.outcome, outcome=step.outcome,
) )
elif isinstance(step, InterruptNode): elif isinstance(step, InterruptNode):
if frame.kind == "subgraph_root" or frame.scope_id != ROOT_SCOPE_ID:
raise WorkflowExecutionError(
"child interrupts are not supported until native subgraph resume routing exists"
)
return handle_interrupt_step(run, step) return handle_interrupt_step(run, step)
elif isinstance(step, ForeachNode): elif isinstance(step, ForeachNode):
return step_foreach(workflow, run, step, index, reducers=reducers) return step_foreach(workflow, run, step, index, reducers=reducers)
+83 -17
View File
@@ -21,6 +21,8 @@ from wf_core import (
WorkflowExecutionError, WorkflowExecutionError,
execute_workflow_async, execute_workflow_async,
execute_workflow, execute_workflow,
resume_workflow_async,
resume_workflow,
) )
from wf_core.validation.issues import ValidationIssueCode from wf_core.validation.issues import ValidationIssueCode
from wf_core.models.steps import Step from wf_core.models.steps import Step
@@ -194,28 +196,70 @@ def test_subgraph_step_routes_through_child_terminal_outcome() -> None:
) )
def test_subgraph_step_rejects_child_interrupt_until_resume_route_exists() -> None: def test_subgraph_step_interrupts_and_resumes_inside_prepared_child() -> None:
child = Workflow( child = _interrupting_child_workflow()
name="child.workflow", parent = _workflow(
input_schema=_schema({"text": {"type": "string"}}), node=_subgraph_node(input_bindings=[{"target": "text", "value": "child-only"}]),
state_schema=StateSchema.from_field_map({}), output_schema=_schema({"answer": {"type": "string"}}),
output_schema=_schema({}), )
start="ask", prepared = PreparedSubgraph(workflow=child, registry={})
nodes=[
InterruptNode.model_validate( run = execute_workflow(
{"id": "ask", "type": "interrupt", "kind": "input"} parent,
) {"text": "hello"},
], {},
edges=[Edge.model_validate({"from": "ask", "outcome": "submitted", "to": END})], subgraphs={"child.workflow": prepared},
) )
with pytest.raises(WorkflowExecutionError, match="child interrupts"): assert run.status == "interrupted"
execute_workflow( assert run.interrupt is not None
_workflow(), assert run.interrupt.node_id == "child"
assert run.interrupt.payload["question"] == "child-only"
assert run.interrupt.route is not None
assert run.interrupt.route.node_id == "ask"
assert run.frames["root"].status == "blocked"
paused = resume_workflow(parent, run, {})
assert paused.status == "interrupted"
assert paused.interrupt is not None
resumed = resume_workflow(
parent,
run,
{},
resume_payload={"answer": "yes"},
subgraphs={"child.workflow": prepared},
)
assert resumed.status == "completed"
assert resumed.output["answer"] == "yes"
assert resumed.state["answer"] == "yes"
def test_subgraph_step_resumes_interrupted_async_prepared_child() -> None:
async def run_child() -> RunState:
child = _interrupting_child_workflow()
parent = _workflow(output_schema=_schema({"answer": {"type": "string"}}))
prepared = PreparedSubgraph(workflow=child, registry={})
run = await execute_workflow_async(
parent,
{"text": "hello"}, {"text": "hello"},
{}, {},
subgraphs={"child.workflow": PreparedSubgraph(workflow=child, registry={})}, subgraphs={"child.workflow": prepared},
) )
return await resume_workflow_async(
parent,
run,
{},
resume_payload={"answer": "async yes"},
subgraphs={"child.workflow": prepared},
)
resumed = asyncio.run(run_child())
assert resumed.status == "completed"
assert resumed.output["answer"] == "async yes"
def _workflow( def _workflow(
@@ -299,5 +343,27 @@ def _child_workflow(
) )
def _interrupting_child_workflow() -> Workflow:
return Workflow(
name="child.workflow",
input_schema=_schema({"text": {"type": "string"}}),
state_schema=StateSchema.from_field_map({"answer": StateField(type="string")}),
output_schema=_schema({"answer": {"type": "string"}}),
start="ask",
nodes=[
InterruptNode.model_validate(
{
"id": "ask",
"type": "interrupt",
"kind": "input",
"request": [{"target": "question", "path": "input.text"}],
"resume": [{"source": "answer", "target": "state.answer"}],
}
)
],
edges=[Edge.model_validate({"from": "ask", "outcome": "submitted", "to": END})],
)
def _schema(properties: dict[str, object]) -> SchemaRef: def _schema(properties: dict[str, object]) -> SchemaRef:
return SchemaRef.model_validate({"type": "object", "properties": properties}) return SchemaRef.model_validate({"type": "object", "properties": properties})