interrupt

wf_core/__init__.py

@@ -2,6 +2,7 @@ from .model import (
     ConditionNode,
     Edge,
     ForeachNode,
+    InterruptNode,
     JoinNode,
     NodeDef,
     NodeResult,
@@ -18,7 +19,7 @@ from .runtime import (
     resume_workflow,
     step_workflow,
 )
-from .run_state import RunState, RunStatus, RuntimeContext, TraceEntry
+from .run_state import InterruptRequest, RunState, RunStatus, RuntimeContext, TraceEntry
 from .tokens import END, START
 from .validate import (
     ValidationIssue,
@@ -31,6 +32,7 @@ __all__ = [
     "ConditionNode",
     "Edge",
     "ForeachNode",
+    "InterruptNode",
     "JoinNode",
     "NodeDef",
     "NodeResult",
@@ -42,6 +44,7 @@ __all__ = [
     "RunStatus",
     "RuntimeContext",
     "TraceEntry",
+    "InterruptRequest",
     "START",
     "END",
     "ValidationIssue",

wf_core/model.py

@@ -104,8 +104,17 @@ class JoinNode(BaseModel):
     type: Literal["join"]
 
 
+class InterruptNode(BaseModel):
+    id: str
+    type: Literal["interrupt"]
+    kind: str
+    request_map: dict[str, str] = Field(default_factory=dict)
+    out_map: dict[str, str] = Field(default_factory=dict)
+    outcomes: list[str] = Field(default_factory=lambda: ["submitted"])
+
+
 Step = Annotated[
-    NodeUse | ConditionNode | ForeachNode | JoinNode,
+    NodeUse | ConditionNode | ForeachNode | JoinNode | InterruptNode,
     Field(discriminator="type"),
 ]
 

wf_core/run_state.py

@@ -32,6 +32,15 @@ class TraceEntry:
     state_changes: dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass(slots=True)
+class InterruptRequest:
+    id: str
+    node_id: str
+    kind: str
+    payload: dict[str, Any] = field(default_factory=dict)
+    resumable: bool = True
+
+
 @dataclass(slots=True)
 class RunState:
     workflow_name: str
@@ -44,6 +53,7 @@ class RunState:
     prior_outcome: str | None = None
     activated_incoming_edge: str | None = None
     error: str | None = None
+    interrupt: InterruptRequest | None = None
 
     def to_dict(self) -> dict[str, Any]:
         return asdict(self)

wf_core/runtime.py

@@ -5,10 +5,19 @@ from typing import Any
 
 from .conditions import eval_condition, safe_resolve_path
 from .errors import WorkflowExecutionError
-from .model import ConditionNode, ForeachNode, JoinNode, NodeDef, NodeResult, NodeUse, Workflow
-from .run_state import RunState, RunStatus, RuntimeContext, TraceEntry
+from .model import (
+    ConditionNode,
+    ForeachNode,
+    InterruptNode,
+    JoinNode,
+    NodeDef,
+    NodeResult,
+    NodeUse,
+    Workflow,
+)
+from .run_state import InterruptRequest, RunState, RunStatus, RuntimeContext, TraceEntry
 from .schema_tools import validate_payload_against_schema
-from .state_ops import apply_output_map, project_output
+from .state_ops import apply_mapped_state, apply_output_map, project_output
 from .tokens import END
 
 
@@ -44,6 +53,9 @@ def resume_workflow(
     workflow: Workflow,
     run: RunState,
     registry: dict[str, NodeHandler],
+    *,
+    resume_payload: dict[str, Any] | None = None,
+    resume_outcome: str = "submitted",
 ) -> RunState:
     if run.workflow_name != workflow.name:
         raise WorkflowExecutionError(
@@ -56,14 +68,43 @@ def resume_workflow(
     if run.status == RunStatus.COMPLETED:
         return run
 
-    run.status = RunStatus.RUNNING
-    run.error = None
     node_defs = {node_def.name: node_def for node_def in workflow.node_defs}
     nodes_by_id = {node.id: node for node in workflow.nodes}
     edge_map = {(edge.from_, edge.outcome): edge.to for edge in workflow.edges}
 
+    if run.status == RunStatus.INTERRUPTED:
+        if resume_payload is None:
+            return run
+        _resume_interrupt(
+            workflow,
+            run,
+            nodes_by_id=nodes_by_id,
+            edge_map=edge_map,
+            resume_payload=resume_payload,
+            resume_outcome=resume_outcome,
+        )
+        if run.current_node_id == END:
+            run.output = project_output(workflow, run.state)
+            validate_payload_against_schema(
+                workflow.output_schema, run.output, "workflow output"
+            )
+            run.status = RunStatus.COMPLETED
+            return run
+
+    run.status = RunStatus.RUNNING
+    run.error = None
+
     while run.current_node_id != END:
-        step_workflow(workflow, run, registry, node_defs=node_defs, nodes_by_id=nodes_by_id, edge_map=edge_map)
+        step_workflow(
+            workflow,
+            run,
+            registry,
+            node_defs=node_defs,
+            nodes_by_id=nodes_by_id,
+            edge_map=edge_map,
+        )
+        if run.status == RunStatus.INTERRUPTED:
+            return run
 
     run.output = project_output(workflow, run.state)
     validate_payload_against_schema(
@@ -85,14 +126,20 @@ def step_workflow(
 ) -> RunState:
     if run.current_node_id is None or run.current_node_id == END:
         return run
+    if run.status == RunStatus.INTERRUPTED:
+        return run
 
     if run.status == RunStatus.PENDING:
         run.status = RunStatus.RUNNING
     run.error = None
 
-    node_defs = node_defs or {node_def.name: node_def for node_def in workflow.node_defs}
+    node_defs = node_defs or {
+        node_def.name: node_def for node_def in workflow.node_defs
+    }
     nodes_by_id = nodes_by_id or {node.id: node for node in workflow.nodes}
-    edge_map = edge_map or {(edge.from_, edge.outcome): edge.to for edge in workflow.edges}
+    edge_map = edge_map or {
+        (edge.from_, edge.outcome): edge.to for edge in workflow.edges
+    }
 
     step = nodes_by_id[run.current_node_id]
 
@@ -117,6 +164,26 @@ def step_workflow(
             "output": {},
             "state_changes": {},
         }
+    elif isinstance(step, InterruptNode):
+        interrupt_request = _build_interrupt_request(
+            step,
+            run.state,
+            run.workflow_input,
+        )
+        run.interrupt = interrupt_request
+        run.status = RunStatus.INTERRUPTED
+        run.trace.append(
+            TraceEntry(
+                node_id=run.current_node_id,
+                step_type=step.type,
+                resolved_input=interrupt_request.payload,
+                outcome="interrupt",
+                next_node_id=run.current_node_id,
+                output=interrupt_request.payload,
+                state_changes={},
+            )
+        )
+        return run
     elif isinstance(step, ForeachNode):
         raise WorkflowExecutionError("foreach execution is not implemented yet")
     else:
@@ -203,3 +270,79 @@ def coerce_node_result(raw_result: NodeResult | dict[str, Any]) -> NodeResult:
     if "outcome" in raw_result and "output" in raw_result:
         return NodeResult.model_validate(raw_result)
     return NodeResult(outcome="ok", output=raw_result)
+
+
+def _build_interrupt_request(
+    node: InterruptNode,
+    state: dict[str, Any],
+    workflow_input: dict[str, Any],
+) -> InterruptRequest:
+    payload = {
+        payload_field: safe_resolve_path(
+            source_path,
+            state=state,
+            workflow_input=workflow_input,
+            context={},
+        )
+        for source_path, payload_field in node.request_map.items()
+    }
+    return InterruptRequest(
+        id=f"interrupt:{node.id}",
+        node_id=node.id,
+        kind=node.kind,
+        payload=payload,
+    )
+
+
+def _resume_interrupt(
+    workflow: Workflow,
+    run: RunState,
+    *,
+    nodes_by_id: dict[str, Any],
+    edge_map: dict[tuple[str, str], str],
+    resume_payload: dict[str, Any],
+    resume_outcome: str,
+) -> None:
+    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]
+    if not isinstance(step, InterruptNode):
+        raise WorkflowExecutionError(
+            f"interrupted run expected interrupt node, got {step.type!r}"
+        )
+    if resume_outcome not in step.outcomes:
+        raise WorkflowExecutionError(
+            f"interrupt node {step.id!r} does not declare resume outcome {resume_outcome!r}"
+        )
+
+    state_changes = apply_mapped_state(
+        workflow,
+        resume_payload,
+        step.out_map,
+        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))
+    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}"
+        )
+
+    run.trace.append(
+        TraceEntry(
+            node_id=run.current_node_id,
+            step_type=step.type,
+            resolved_input=resume_payload,
+            outcome=resume_outcome,
+            next_node_id=next_node_id,
+            output=resume_payload,
+            state_changes=state_changes,
+        )
+    )
+    run.prior_outcome = resume_outcome
+    run.activated_incoming_edge = run.current_node_id
+    run.current_node_id = next_node_id
+    run.interrupt = None

wf_core/state_ops.py

@@ -4,7 +4,12 @@ from typing import Any
 
 from .errors import WorkflowExecutionError
 from .model import NodeUse, Workflow
-from .paths import PathResolutionError, get_nested_value, set_nested_value, split_graph_path
+from .paths import (
+    PathResolutionError,
+    get_nested_value,
+    set_nested_value,
+    split_graph_path,
+)
 
 
 def apply_output_map(
@@ -12,14 +17,31 @@ def apply_output_map(
     node: NodeUse,
     node_output: dict[str, Any],
     state: dict[str, Any],
+) -> dict[str, Any]:
+    return apply_mapped_state(
+        workflow,
+        node_output,
+        node.out_map,
+        state,
+        missing_field_message=f"node {node.id!r} did not return required mapped field {{field}}",
+    )
+
+
+def apply_mapped_state(
+    workflow: Workflow,
+    source_data: dict[str, Any],
+    mapping: dict[str, str],
+    state: dict[str, Any],
+    *,
+    missing_field_message: str,
 ) -> dict[str, Any]:
     state_changes: dict[str, Any] = {}
-    for source_field, destination_path in node.out_map.items():
-        if source_field not in node_output:
+    for source_field, destination_path in mapping.items():
+        if source_field not in source_data:
             raise WorkflowExecutionError(
-                f"node {node.id!r} did not return required mapped field {source_field!r}"
+                missing_field_message.format(field=repr(source_field))
             )
-        value = node_output[source_field]
+        value = source_data[source_field]
         write_state_value(workflow, state, destination_path, value)
         state_changes[destination_path] = value
     return state_changes

wf_core/validate.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 from enum import StrEnum
-from typing import Any
 
 from .model import (
     BinaryCondition,
@@ -11,6 +10,7 @@ from .model import (
     Edge,
     ExistsCondition,
     ForeachNode,
+    InterruptNode,
     LiteralOperand,
     NodeDef,
     NodeUse,
@@ -41,6 +41,8 @@ class ValidationIssueCode(StrEnum):
     EMPTY_CONDITION_ARGS = "empty_condition_args"
     INVALID_CONDITION_PATH = "invalid_condition_path"
     INVALID_FOREACH_SOURCE = "invalid_foreach_source"
+    INVALID_INTERRUPT_SOURCE = "invalid_interrupt_source"
+    INVALID_INTERRUPT_DESTINATION = "invalid_interrupt_destination"
 
 
 @dataclass(slots=True)
@@ -108,6 +110,10 @@ def validate_workflow(workflow: Workflow) -> ValidationReport:
             _validate_foreach_node(
                 node, index, report, state_root_fields, input_root_fields
             )
+        elif isinstance(node, InterruptNode):
+            _validate_interrupt_node(
+                node, index, report, state_root_fields, input_root_fields
+            )
 
     if workflow.start not in nodes_by_id:
         report.add(
@@ -258,6 +264,42 @@ def _validate_foreach_node(
         )
 
 
+def _validate_interrupt_node(
+    node: InterruptNode,
+    index: int,
+    report: ValidationReport,
+    state_root_fields: set[str],
+    input_root_fields: set[str],
+) -> None:
+    for source_path, payload_field in node.request_map.items():
+        if not payload_field:
+            report.add(
+                ValidationIssueCode.INVALID_INTERRUPT_SOURCE,
+                f"nodes[{index}].request_map[{source_path!r}]",
+                "interrupt request payload field must not be empty",
+            )
+        if not is_valid_source_path(source_path, state_root_fields, input_root_fields):
+            report.add(
+                ValidationIssueCode.INVALID_INTERRUPT_SOURCE,
+                f"nodes[{index}].request_map[{source_path!r}]",
+                "interrupt request source must start with input. or state. and reference a declared root field",
+            )
+
+    for resume_field, destination_path in node.out_map.items():
+        if not resume_field:
+            report.add(
+                ValidationIssueCode.INVALID_INTERRUPT_DESTINATION,
+                f"nodes[{index}].out_map[{resume_field!r}]",
+                "interrupt resume field must not be empty",
+            )
+        if not is_valid_destination_path(destination_path):
+            report.add(
+                ValidationIssueCode.INVALID_INTERRUPT_DESTINATION,
+                f"nodes[{index}].out_map[{resume_field!r}]",
+                "interrupt resume destination must start with state.",
+            )
+
+
 def _validate_condition_expr(
     condition: Condition,
     path: str,
@@ -346,6 +388,8 @@ def _declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> se
         return {"done"}
     if step.type == "join":
         return {"done"}
+    if step.type == "interrupt":
+        return set(step.outcomes)
     return set()