interrupt to get the same treatment too!

This commit is contained in:
lda
2026-05-21 14:15:52 +07:00 Verified
parent 274bf34edf
commit b619e529fa
15 changed files with 317 additions and 66 deletions
+6
View File
@@ -79,6 +79,12 @@ dump only canonical `input` and `output` bindings. `wf_authoring` exposes the
same canonical binding lists and keeps `in_map`, `input_values`, and `out_map` same canonical binding lists and keeps `in_map`, `input_values`, and `out_map`
only as deprecated Python-builder sugar that compiles into canonical bindings. only as deprecated Python-builder sugar that compiles into canonical bindings.
Interrupts follow the same pattern with different field names: `request` uses
canonical input bindings to build the public interrupt payload, and `resume`
uses canonical output bindings to write resume payload fields back into state.
Legacy interrupt `request_map` and `out_map` inputs are accepted only as
parse compatibility and should not be written by new plans.
## Explicitness Rules ## Explicitness Rules
### Node-local writes must not overlap ### Node-local writes must not overlap
+8 -6
View File
@@ -150,16 +150,18 @@ metadata at runtime without becoming a declared schema field.
Current core interrupt semantics are node-level. An `InterruptNode` has: Current core interrupt semantics are node-level. An `InterruptNode` has:
- `kind` - `kind`
- `request_map`, mapping state/input/context paths to public interrupt payload - `request`, a canonical input-binding list mapping state/input/context paths
fields or literal values to public interrupt payload fields
- `out_map`, mapping resume payload fields back into state - `resume`, a canonical output-binding list mapping resume payload fields back
into workflow state
- declared resume `outcomes` - declared resume `outcomes`
That means an artifact can document interrupt boundaries by scanning its That means an artifact can document interrupt boundaries by scanning its
declarative plan for interrupt nodes and deriving their request/resume payload declarative plan for interrupt nodes and deriving their request/resume payload
schemas from the maps and workflow state/input schemas. It should not need to schemas from the bindings and workflow state/input schemas. Legacy
store unrelated child graph internals just to describe the public interrupt `request_map` and `out_map` inputs are parse-only compatibility shapes; saved
points. artifacts should write `request` and `resume`. They should not need to store
unrelated child graph internals just to describe the public interrupt points.
## Composition Rule ## Composition Rule
+18 -6
View File
@@ -296,17 +296,29 @@ Declares an interrupting step.
{ {
"interrupt": { "interrupt": {
"kind": "input", "kind": "input",
"request": { "request": [
"state.question": "question" {
}, "target": {"root": "local", "parts": ["question"]},
"resume": { "path": {"root": "state", "parts": ["question"]}
"answer": "state.answer" }
}, ],
"resume": [
{
"source": {"root": "local", "parts": ["answer"]},
"target": {"root": "state", "parts": ["answer"]}
}
],
"outcomes": ["resumed", "cancelled"] "outcomes": ["resumed", "cancelled"]
} }
} }
``` ```
Draft interrupts use the same binding shapes as core interrupt nodes:
`request` builds the public interrupt payload, while `resume` maps the payload
provided on resume back into workflow state. Older map-shaped `request` and
`resume` values are accepted only as parse compatibility and dump back to the
canonical list shape.
Saved interrupting artifacts are still limited in the current execution Saved interrupting artifacts are still limited in the current execution
surface. If a deployment reports `interrupting_artifact_unsupported`, that is a surface. If a deployment reports `interrupting_artifact_unsupported`, that is a
known platform limitation rather than a draft bug. known platform limitation rather than a draft bug.
+21 -6
View File
@@ -209,14 +209,29 @@ def build_demo_workflow() -> Workflow:
"id": "approve_email", "id": "approve_email",
"type": "interrupt", "type": "interrupt",
"kind": "approval", "kind": "approval",
"request_map": { "request": [
"state.summary": "summary", {
"input.folder_id": "folder_id", "target": {"root": "local", "parts": ["summary"]},
"path": {"root": "state", "parts": ["summary"]},
}, },
"out_map": { {
"approved": "state.approved", "target": {"root": "local", "parts": ["folder_id"]},
"comment": "state.approval_comment", "path": {"root": "input", "parts": ["folder_id"]},
}, },
],
"resume": [
{
"source": {"root": "local", "parts": ["approved"]},
"target": {"root": "state", "parts": ["approved"]},
},
{
"source": {"root": "local", "parts": ["comment"]},
"target": {
"root": "state",
"parts": ["approval_comment"],
},
},
],
"outcomes": ["submitted", "cancelled"], "outcomes": ["submitted", "cancelled"],
}, },
{ {
+2 -2
View File
@@ -58,8 +58,8 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
return builder.interrupt( return builder.interrupt(
id=step_id, id=step_id,
kind=step.interrupt.kind, kind=step.interrupt.kind,
request_map=step.interrupt.request, request=step.interrupt.request,
out_map=step.interrupt.resume, resume=step.interrupt.resume,
outcomes=step.interrupt.outcomes, outcomes=step.interrupt.outcomes,
) )
if isinstance(step, DraftJoinStep): if isinstance(step, DraftJoinStep):
+22 -2
View File
@@ -114,10 +114,30 @@ class DraftInterruptPayload(BaseModel):
model_config = ConfigDict(extra="forbid") model_config = ConfigDict(extra="forbid")
kind: str kind: str
request: dict[str, str] = Field(default_factory=dict) request: list[InputBinding] = Field(default_factory=list)
resume: dict[str, str] = Field(default_factory=dict) resume: list[OutputBinding] = Field(default_factory=list)
outcomes: list[str] = Field(default_factory=lambda: ["submitted"]) outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
@model_validator(mode="before")
@classmethod
def _coerce_legacy_maps(cls, data: object) -> object:
"""Accept old draft interrupt maps but save canonical binding lists."""
if not isinstance(data, dict):
return data
data = dict(data)
request = data.get("request", [])
resume = data.get("resume", [])
if isinstance(request, dict):
data["request"] = [
{"target": target, "path": path} for path, target in request.items()
]
if isinstance(resume, dict):
data["resume"] = [
{"source": source, "target": target}
for source, target in resume.items()
]
return data
class DraftInterruptStep(BaseModel): class DraftInterruptStep(BaseModel):
"""Draft step that pauses execution and waits for resume input.""" """Draft step that pauses execution and waits for resume input."""
+28 -3
View File
@@ -49,7 +49,6 @@ from .mapping import (
normalize_input_bindings, normalize_input_bindings,
normalize_input_mapping, normalize_input_mapping,
normalize_input_values, normalize_input_values,
normalize_mapping,
normalize_output_bindings, normalize_output_bindings,
normalize_output_mapping, normalize_output_mapping,
) )
@@ -370,16 +369,42 @@ class WorkflowBuilder:
*, *,
id: str | None = None, id: str | None = None,
kind: str, kind: str,
request: Sequence[InputBindingArg] | None = None,
resume: Sequence[OutputBindingArg] | None = None,
request_map: MapArg | None = None, request_map: MapArg | None = None,
out_map: MapArg | None = None, out_map: MapArg | None = None,
outcomes: list[str] | None = None, outcomes: list[str] | None = None,
) -> InterruptNode: ) -> InterruptNode:
if request is not None and request_map is not None:
raise TypeError("cannot mix canonical request with deprecated request_map")
if resume is not None and out_map is not None:
raise TypeError("cannot mix canonical resume with deprecated out_map")
if request_map is not None or out_map is not None:
warnings.warn(
"request_map/out_map are deprecated interrupt sugar; use canonical "
"request/resume binding lists instead",
DeprecationWarning,
stacklevel=2,
)
request_bindings = (
normalize_input_bindings(request)
if request is not None
else _canonical_input_bindings(
normalize_input_mapping(request_map),
{},
)
)
resume_bindings = (
normalize_output_bindings(resume)
if resume is not None
else _canonical_output_bindings(normalize_output_mapping(out_map))
)
node = InterruptNode( node = InterruptNode(
id=id or self._next_step_id(f"interrupt_{slug_id(kind)}"), id=id or self._next_step_id(f"interrupt_{slug_id(kind)}"),
type="interrupt", type="interrupt",
kind=kind, kind=kind,
request_map=normalize_mapping(request_map), request=request_bindings,
out_map=normalize_mapping(out_map), resume=resume_bindings,
outcomes=outcomes or ["submitted"], outcomes=outcomes or ["submitted"],
) )
self.nodes.append(node) self.nodes.append(node)
+40 -2
View File
@@ -143,10 +143,48 @@ class InterruptNode(BaseModel):
id: str id: str
type: Literal["interrupt"] type: Literal["interrupt"]
kind: str kind: str
request_map: dict[str, str] = Field(default_factory=dict) request: list[InputBinding] = Field(default_factory=list)
out_map: dict[str, str] = Field(default_factory=dict) resume: list[OutputBinding] = Field(default_factory=list)
outcomes: list[str] = Field(default_factory=lambda: ["submitted"]) outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
@model_validator(mode="before")
@classmethod
def _coerce_deprecated_maps(cls, data: object) -> object:
"""Normalize legacy interrupt maps into canonical parse-only bindings."""
if not isinstance(data, Mapping):
return data
old_fields = ("request_map", "out_map")
has_canonical = "request" in data or "resume" in data
present_old_fields = [field for field in old_fields if field in data]
if has_canonical and present_old_fields:
old_names = ", ".join(present_old_fields)
raise ValueError(
f"cannot mix canonical request/resume with deprecated fields: {old_names}"
)
normalized = dict(data)
request_bindings = list(normalized.pop("request", []))
resume_bindings = list(normalized.pop("resume", []))
request_map = NodeUse._deprecated_mapping(
normalized.pop("request_map", {}), field_name="request_map"
)
out_map = NodeUse._deprecated_mapping(
normalized.pop("out_map", {}), field_name="out_map"
)
request_bindings.extend(
{"target": target, "path": path} for path, target in request_map.items()
)
resume_bindings.extend(
{"source": source, "target": target} for source, target in out_map.items()
)
normalized["request"] = request_bindings
normalized["resume"] = resume_bindings
return normalized
Step = Annotated[ Step = Annotated[
NodeUse | ConditionNode | ForeachNode | JoinNode | InterruptNode, NodeUse | ConditionNode | ForeachNode | JoinNode | InterruptNode,
+20 -9
View File
@@ -5,13 +5,14 @@ from typing import Any
from wf_core.conditions import safe_resolve_path from wf_core.conditions import safe_resolve_path
from wf_core.errors import WorkflowExecutionError from wf_core.errors import WorkflowExecutionError
from wf_core.models.steps import InterruptNode from wf_core.local_paths import LocalPathError, set_local_value
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 InterruptRequest, RunState, StepExecutionResult
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_mapped_state from wf_core.runtime.ops.state import apply_output_bindings
def build_interrupt_request( def build_interrupt_request(
@@ -22,15 +23,25 @@ def build_interrupt_request(
workflow_input: dict[str, Any], workflow_input: dict[str, Any],
context: dict[str, Any], context: dict[str, Any],
) -> InterruptRequest: ) -> InterruptRequest:
payload = { payload: dict[str, Any] = {}
payload_field: safe_resolve_path( for binding in node.request:
source_path, if isinstance(binding, InputValueBinding):
value = binding.value
elif isinstance(binding, InputPathBinding):
value = safe_resolve_path(
str(binding.path),
state=state, state=state,
workflow_input=workflow_input, workflow_input=workflow_input,
context=context, context=context,
) )
for source_path, payload_field in node.request_map.items() else:
} raise WorkflowExecutionError(
f"unsupported request binding for interrupt {node.id!r}"
)
try:
set_local_value(payload, binding.target, value)
except LocalPathError as exc:
raise WorkflowExecutionError(str(exc)) from exc
return InterruptRequest( return InterruptRequest(
id=f"interrupt:{node.id}", id=f"interrupt:{node.id}",
frame_id=frame_id, frame_id=frame_id,
@@ -67,10 +78,10 @@ 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_mapped_state( state_changes = apply_output_bindings(
workflow, workflow,
step.resume,
resume_payload, resume_payload,
step.out_map,
run.state, run.state,
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}",
+18 -11
View File
@@ -165,31 +165,38 @@ def validate_interrupt_node(
state_root_fields: set[str], state_root_fields: set[str],
input_root_fields: set[str], input_root_fields: set[str],
) -> None: ) -> None:
for source_path, payload_field in node.request_map.items(): for binding_index, binding in enumerate(node.request):
if not payload_field: field_path = f"nodes[{index}].request[{binding_index}]"
if not str(binding.target):
report.add( report.add(
ValidationIssueCode.INVALID_INTERRUPT_SOURCE, ValidationIssueCode.INVALID_INTERRUPT_SOURCE,
f"nodes[{index}].request_map[{source_path!r}]", field_path,
"interrupt request payload field must not be empty", "interrupt request payload field must not be empty",
) )
if not is_valid_source_path(source_path, state_root_fields, input_root_fields): if isinstance(binding, InputPathBinding) and not is_valid_source_path(
binding.path,
state_root_fields,
input_root_fields,
allow_context=True,
):
report.add( report.add(
ValidationIssueCode.INVALID_INTERRUPT_SOURCE, ValidationIssueCode.INVALID_INTERRUPT_SOURCE,
f"nodes[{index}].request_map[{source_path!r}]", field_path,
"interrupt request source must start with input. or state. and reference a declared root field", "interrupt request source must start with input., state., or context. and reference a declared root field when applicable",
) )
for resume_field, destination_path in node.out_map.items(): for binding_index, binding in enumerate(node.resume):
if not resume_field: field_path = f"nodes[{index}].resume[{binding_index}]"
if not str(binding.source):
report.add( report.add(
ValidationIssueCode.INVALID_INTERRUPT_DESTINATION, ValidationIssueCode.INVALID_INTERRUPT_DESTINATION,
f"nodes[{index}].out_map[{resume_field!r}]", field_path,
"interrupt resume field must not be empty", "interrupt resume field must not be empty",
) )
if not is_valid_destination_path(destination_path): if not is_valid_destination_path(binding.target):
report.add( report.add(
ValidationIssueCode.INVALID_INTERRUPT_DESTINATION, ValidationIssueCode.INVALID_INTERRUPT_DESTINATION,
f"nodes[{index}].out_map[{resume_field!r}]", field_path,
"interrupt resume destination must start with state.", "interrupt resume destination must start with state.",
) )
+40
View File
@@ -55,6 +55,46 @@ def test_workflow_draft_accepts_legacy_use_maps_but_dumps_canonical_bindings() -
} }
def test_workflow_draft_accepts_legacy_interrupt_maps_but_dumps_canonical_bindings() -> (
None
):
draft = WorkflowDraft.model_validate(
{
**_keyed_echo_draft(),
"start": "approval",
"steps": {
"approval": {
"interrupt": {
"kind": "approval",
"request": {"input.text": "message"},
"resume": {"approved": "state.approved"},
}
},
},
"routes": {"approval": {"submitted": "__end__"}},
}
)
dumped = draft.model_dump(mode="json")
assert dumped["steps"]["approval"]["interrupt"]["request"][0]["path"] == {
"root": "input",
"parts": ["text"],
}
assert dumped["steps"]["approval"]["interrupt"]["request"][0]["target"] == {
"root": "local",
"parts": ["message"],
}
assert dumped["steps"]["approval"]["interrupt"]["resume"][0]["source"] == {
"root": "local",
"parts": ["approved"],
}
assert dumped["steps"]["approval"]["interrupt"]["resume"][0]["target"] == {
"root": "state",
"parts": ["approved"],
}
def test_draft_step_requires_exactly_one_kind_key() -> None: def test_draft_step_requires_exactly_one_kind_key() -> None:
draft = _keyed_echo_draft() draft = _keyed_echo_draft()
steps = draft["steps"] steps = draft["steps"]
+21
View File
@@ -277,6 +277,27 @@ def test_builder_can_auto_id_condition_foreach_and_interrupt() -> None:
assert interrupt.id == "interrupt_approval" assert interrupt.id == "interrupt_approval"
def test_builder_interrupt_accepts_canonical_request_and_resume_bindings() -> None:
builder = WorkflowBuilder(
name="interrupt_bindings",
input_schema=AutoBindInput,
state_schema=AutoBindState,
output_schema=AutoBindOutput,
)
interrupt = builder.interrupt(
kind="approval",
request=[input_from(input_path("text"), "message")],
resume=[output_to("text", state_path("text"))],
)
assert isinstance(interrupt.request[0], InputPathBinding)
assert interrupt.request[0].path == GraphSourcePath.input("text")
assert interrupt.request[0].target == LocalPath.of("message")
assert interrupt.resume[0].source == LocalPath.of("text")
assert interrupt.resume[0].target == StatePath.of("text")
def test_builder_connect_can_use_node_specs_and_returns_resolved_refs() -> None: def test_builder_connect_can_use_node_specs_and_returns_resolved_refs() -> None:
builder = WorkflowBuilder( builder = WorkflowBuilder(
name="connect_specs_demo", name="connect_specs_demo",
+8 -10
View File
@@ -18,8 +18,6 @@ from examples.demo_workflow import build_demo_registry, build_demo_workflow
from wf_authoring import ( from wf_authoring import (
NodeReturn, NodeReturn,
WorkflowBuilder, WorkflowBuilder,
bind_fields,
bind_state,
build_registry, build_registry,
input_from, input_from,
state, state,
@@ -192,14 +190,14 @@ def build_authoring_demo_workflow():
approve_email = builder.interrupt( approve_email = builder.interrupt(
id="approve_email", id="approve_email",
kind="approval", kind="approval",
request_map=bind_fields( request=[
summary=state_path("summary"), input_from(state_path("summary"), "summary"),
folder_id=input_path("folder_id"), input_from(input_path("folder_id"), "folder_id"),
), ],
out_map=bind_state( resume=[
approved=state_path("approved"), output_to("approved", state_path("approved")),
comment=state_path("approval_comment"), output_to("comment", state_path("approval_comment")),
), ],
outcomes=["submitted", "cancelled"], outcomes=["submitted", "cancelled"],
) )
skip_email = builder.use( skip_email = builder.use(
+57 -1
View File
@@ -1,7 +1,12 @@
import pytest import pytest
from pydantic import ValidationError from pydantic import ValidationError
from wf_core.models.steps import InputPathBinding, InputValueBinding, NodeUse from wf_core.models.steps import (
InputPathBinding,
InputValueBinding,
InterruptNode,
NodeUse,
)
from wf_core.paths import GraphSourcePath, LocalPath, StatePath from wf_core.paths import GraphSourcePath, LocalPath, StatePath
@@ -220,3 +225,54 @@ def test_deprecated_input_value_preserves_explicit_null():
dumped_input = node.model_dump(mode="json")["input"] dumped_input = node.model_dump(mode="json")["input"]
assert dumped_input[0]["target"] == {"root": "local", "parts": ["maybe"]} assert dumped_input[0]["target"] == {"root": "local", "parts": ["maybe"]}
assert dumped_input[0]["value"] is None assert dumped_input[0]["value"] is None
def test_interrupt_node_accepts_canonical_request_and_resume_bindings():
node = InterruptNode.model_validate(
{
"id": "approval",
"type": "interrupt",
"kind": "approval",
"request": [{"target": "summary", "path": "state.summary"}],
"resume": [{"source": "approved", "target": "state.approved"}],
}
)
assert isinstance(node.request[0], InputPathBinding)
assert node.request[0].path == GraphSourcePath.state("summary")
assert node.request[0].target == LocalPath.of("summary")
assert node.resume[0].source == LocalPath.of("approved")
assert node.resume[0].target == StatePath.of("approved")
def test_interrupt_node_converts_old_maps_to_canonical_bindings():
node = InterruptNode.model_validate(
{
"id": "approval",
"type": "interrupt",
"kind": "approval",
"request_map": {"input.message": "message"},
"out_map": {"approved": "state.approved"},
}
)
dumped = node.model_dump(mode="json")
assert "request_map" not in dumped
assert "out_map" not in dumped
assert dumped["request"][0]["path"] == {"root": "input", "parts": ["message"]}
assert dumped["request"][0]["target"] == {"root": "local", "parts": ["message"]}
assert dumped["resume"][0]["source"] == {"root": "local", "parts": ["approved"]}
assert dumped["resume"][0]["target"] == {"root": "state", "parts": ["approved"]}
def test_interrupt_node_rejects_mixed_old_and_new_binding_styles():
with pytest.raises(ValidationError):
InterruptNode.model_validate(
{
"id": "approval",
"type": "interrupt",
"kind": "approval",
"request": [{"target": "message", "path": "input.message"}],
"request_map": {"input.other": "other"},
}
)
+2 -2
View File
@@ -564,8 +564,8 @@ def _interrupt_artifact() -> WorkflowArtifact:
"id": "approval", "id": "approval",
"type": "interrupt", "type": "interrupt",
"kind": "approval", "kind": "approval",
"request_map": {"input.message": "message"}, "request": [input_binding("input.message", "message")],
"out_map": {}, "resume": [],
"outcomes": ["submitted"], "outcomes": ["submitted"],
} }
], ],