i can now Use our system programmatically

authoring_sketch.md

@@ -0,0 +1,135 @@
+# Authoring Layer Sketch
+
+This document sketches the next layer on top of `wf_core`.
+
+The goal is to make workflow authoring pleasant for:
+
+- Python developers using `@node`
+- builder-style graph construction
+- client LLMs that consume a node catalog and emit graph structure
+
+This layer should compile down to the existing core model without changing
+runtime semantics.
+
+## Principles
+
+1. `wf_core` remains the execution model.
+2. Authoring APIs compile to `wf_core.Workflow`.
+3. `NodeDef` should usually be derived, not written manually.
+4. The LLM should usually choose from a node catalog, not invent raw node defs.
+5. Generics belong at the authoring boundary, not in the runtime core.
+
+## Main Objects
+
+### `NodeSpec[InputT, OutputT]`
+
+The durable product of `@node`.
+
+Responsibilities:
+
+- hold typed Python callable metadata
+- expose input and output model types
+- expose node outcomes
+- generate a core `NodeDef`
+- register a runtime handler
+- remain directly callable in Python
+
+This should be the single central wrapper object. Avoid creating multiple
+parallel wrapper types with overlapping meanings.
+
+### `WorkflowBuilder`
+
+Builder API for human Python authors.
+
+Responsibilities:
+
+- collect node uses
+- collect control-flow nodes
+- collect edges
+- derive unique `NodeDef`s from referenced `NodeSpec`s
+- compile to a core `Workflow`
+
+Typical entry points:
+
+- `use(node_spec, id=..., in_map=..., out_map=...)`
+- `condition(...)`
+- `foreach(...)`
+- `interrupt(...)`
+- `connect(...)`
+- `compile()`
+
+### `NodeCatalog`
+
+LLM-facing registry of available nodes.
+
+Responsibilities:
+
+- expose name, docs, schemas, and outcomes
+- provide a normalized machine-readable view for MCP consumers
+- allow the client LLM to build `NodeUse`s against known nodes
+
+The client LLM should usually receive a node catalog and emit graph structure
+that references those known nodes. It should not usually generate new raw
+`NodeDef`s.
+
+## Flow
+
+### Python authoring flow
+
+1. declare `InputModel` and `OutputModel`
+2. decorate a function with `@node(...)`
+3. receive a `NodeSpec`
+4. add `NodeSpec`s to a `WorkflowBuilder`
+5. compile builder to core `Workflow`
+6. build a registry from the same `NodeSpec`s
+7. run with existing runtime
+
+### LLM graph authoring flow
+
+1. MCP exposes a `NodeCatalog`
+2. client LLM selects nodes from the catalog
+3. client LLM emits graph structure:
+   - node uses
+   - mappings
+   - conditions
+   - foreach nodes
+   - interrupt nodes
+   - edges
+4. server compiles or validates that structure into core `Workflow`
+5. runtime executes core `Workflow`
+
+## Docs and schema descriptions
+
+Input and output models should prefer `pydantic.BaseModel`.
+
+Recommended sources of documentation:
+
+- class docstring: model-level description
+- `Field(description=...)`: strongest field-level description
+- attribute docstrings with `ConfigDict(use_attribute_docstrings=True)`: good authoring UX
+
+The authoring layer should normalize these into schema descriptions so MCP can
+surface them to client LLMs.
+
+## Async stance
+
+Do not hide async behind `.result()`.
+
+Preferred design:
+
+- `NodeSpec` knows whether a callable is sync or async
+- sync runtime accepts sync handlers
+- future async runtime accepts async handlers
+
+If sync runtime encounters an async node, fail clearly rather than faking a sync
+bridge.
+
+## Future extension
+
+### `Workflow -> NodeSpec`
+
+In the future, a compiled workflow or subgraph can be wrapped as a reusable
+`NodeSpec`, likely by treating workflow input and output schemas as the node's
+input and output schemas.
+
+That should be an authoring-layer transformation, not a core runtime rewrite.

main.py

@@ -1,11 +1,11 @@
 import json
 import sys
 
-from wf_core import RuntimeContext, execute_workflow, resume_workflow
+from wf_core import RunState, execute_workflow, resume_workflow
 from wf_core.demo_workflow import build_demo_registry, build_demo_workflow
 
 
-def print_run(label: str, run: object) -> None:
+def print_run(label: str, run: RunState) -> None:
     print(f"{label}:")
     print(json.dumps(run.to_dict(), indent=2))
 

pyproject.toml

@@ -13,3 +13,6 @@ dependencies = [
 dev = [
     "pytest>=8.4.0",
 ]
+
+[tool.pytest.ini_options]
+addopts = "-p no:cacheprovider"

tests/test_demo_workflow.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+from pydantic import BaseModel
+
 from wf_core import (
     END,
     FrameStatus,
+    RuntimeContext,
     RunStatus,
     execute_workflow,
     resume_workflow,
@@ -10,6 +13,224 @@ from wf_core import (
 )
 from wf_core.demo_workflow import build_demo_registry, build_demo_workflow
 from wf_core.run_factory import create_run_state
+from wf_authoring import WorkflowBuilder, node
+
+
+class DriveListFilesInput(BaseModel):
+    folder_id: str
+
+
+class DriveListFilesOutput(BaseModel):
+    documents: list[str]
+
+
+class SummarizeDocumentInput(BaseModel):
+    document: str
+
+
+class SummarizeDocumentOutput(BaseModel):
+    item_summary: str
+
+
+class CombineSummariesInput(BaseModel):
+    item_summaries: list[str]
+
+
+class CombineSummariesOutput(BaseModel):
+    summary: str
+
+
+class SendEmailInput(BaseModel):
+    summary: str
+
+
+class SendEmailOutput(BaseModel):
+    email_status: str
+
+
+class MarkEmailSkippedInput(BaseModel):
+    pass
+
+
+class MarkEmailSkippedOutput(BaseModel):
+    email_status: str
+
+
+@node(
+    name="drive_list_files",
+    input_model=DriveListFilesInput,
+    output_model=DriveListFilesOutput,
+)
+def drive_list_files_spec(
+    payload: DriveListFilesInput,
+    ctx: RuntimeContext,
+) -> DriveListFilesOutput:
+    return DriveListFilesOutput(
+        documents=[
+            f"{payload.folder_id}/meeting-notes.md",
+            f"{payload.folder_id}/weekly-report.md",
+        ]
+    )
+
+
+@node(
+    name="summarize_document",
+    input_model=SummarizeDocumentInput,
+    output_model=SummarizeDocumentOutput,
+)
+def summarize_document_spec(
+    payload: SummarizeDocumentInput,
+    ctx: RuntimeContext,
+) -> SummarizeDocumentOutput:
+    return SummarizeDocumentOutput(item_summary=f"Summary of {payload.document}")
+
+
+@node(
+    name="combine_summaries",
+    input_model=CombineSummariesInput,
+    output_model=CombineSummariesOutput,
+)
+def combine_summaries_spec(
+    payload: CombineSummariesInput,
+    ctx: RuntimeContext,
+) -> CombineSummariesOutput:
+    return CombineSummariesOutput(summary=" | ".join(payload.item_summaries))
+
+
+@node(
+    name="send_email",
+    input_model=SendEmailInput,
+    output_model=SendEmailOutput,
+    outcomes=("sent",),
+)
+def send_email_spec(
+    payload: SendEmailInput,
+    ctx: RuntimeContext,
+) -> dict[str, object]:
+    return {
+        "outcome": "sent",
+        "output": {"email_status": f"sent: {payload.summary}"},
+    }
+
+
+@node(
+    name="mark_email_skipped",
+    input_model=MarkEmailSkippedInput,
+    output_model=MarkEmailSkippedOutput,
+)
+def mark_email_skipped_spec(
+    payload: MarkEmailSkippedInput,
+    ctx: RuntimeContext,
+) -> MarkEmailSkippedOutput:
+    return MarkEmailSkippedOutput(email_status="skipped")
+
+
+def build_authoring_demo_workflow():
+    declared = build_demo_workflow()
+    builder = WorkflowBuilder(
+        name=declared.name,
+        input_schema=declared.input_schema,
+        state_schema=declared.state_schema,
+        output_schema=declared.output_schema,
+        start="list_files",
+    )
+
+    builder.use(
+        drive_list_files_spec,
+        id="list_files",
+        in_map={"input.folder_id": "folder_id"},
+        out_map={"documents": "state.documents"},
+        desc="List files from a Google Drive folder",
+    )
+    builder.foreach(
+        id="summarize_each",
+        over="state.documents",
+        as_="document",
+        mode="serial",
+        on_item_error="fail",
+    )
+    builder.use(
+        summarize_document_spec,
+        id="summarize_one",
+        in_map={"context.document": "document"},
+        out_map={"item_summary": "state.item_summaries"},
+        desc="Summarize one document",
+    )
+    builder.use(
+        combine_summaries_spec,
+        id="combine_summaries",
+        in_map={"state.item_summaries": "item_summaries"},
+        out_map={"summary": "state.summary"},
+        desc="Combine item summaries into one final summary",
+    )
+    builder.condition(
+        id="should_email",
+        check={
+            "op": "eq",
+            "left": {"path": "state.should_email"},
+            "right": {"value": True},
+        },
+    )
+    builder.use(
+        send_email_spec,
+        id="send_email",
+        in_map={"state.summary": "summary"},
+        out_map={"email_status": "state.email_status"},
+        desc="Send the summary by email",
+    )
+    builder.interrupt(
+        id="approve_email",
+        kind="approval",
+        request_map={
+            "state.summary": "summary",
+            "input.folder_id": "folder_id",
+        },
+        out_map={
+            "approved": "state.approved",
+            "comment": "state.approval_comment",
+        },
+        outcomes=["submitted", "cancelled"],
+    )
+    builder.use(
+        mark_email_skipped_spec,
+        id="skip_email",
+        out_map={"email_status": "state.email_status"},
+        desc="Record that email delivery was skipped",
+    )
+
+    builder.connect("list_files", "ok", "summarize_each")
+    builder.connect("summarize_each", "loop", "summarize_one")
+    builder.connect("summarize_each", "done", "combine_summaries")
+    builder.connect("summarize_one", "ok", END)
+    builder.connect("combine_summaries", "ok", "should_email")
+    builder.connect("should_email", "true", "approve_email")
+    builder.connect("should_email", "false", "skip_email")
+    builder.connect("approve_email", "submitted", "send_email")
+    builder.connect("approve_email", "cancelled", "skip_email")
+    builder.connect("send_email", "sent", END)
+    builder.connect("skip_email", "ok", END)
+
+    registry = {
+        drive_list_files_spec.name: drive_list_files_spec.to_registry_handler(),
+        summarize_document_spec.name: summarize_document_spec.to_registry_handler(),
+        combine_summaries_spec.name: combine_summaries_spec.to_registry_handler(),
+        send_email_spec.name: send_email_spec.to_registry_handler(),
+        mark_email_skipped_spec.name: mark_email_skipped_spec.to_registry_handler(),
+    }
+    return builder.compile(), registry
+
+
+def _strip_schema_titles(value: object) -> object:
+    if isinstance(value, dict):
+        normalized = {
+            key: _strip_schema_titles(inner)
+            for key, inner in value.items()
+            if key not in {"title", "items"}
+        }
+        return normalized
+    if isinstance(value, list):
+        return [_strip_schema_titles(item) for item in value]
+    return value
 
 
 def test_interrupt_then_resume_to_send_email() -> None:
@@ -116,3 +337,34 @@ def test_foreach_stress_with_many_documents() -> None:
     assert len([entry for entry in run.trace if entry.step_type == "foreach"]) == (
         document_count + 1
     )
+
+
+def test_builder_compiles_same_workflow_as_declared_demo() -> None:
+    declared = build_demo_workflow()
+    built, _registry = build_authoring_demo_workflow()
+
+    assert _strip_schema_titles(built.model_dump(by_alias=True)) == _strip_schema_titles(
+        declared.model_dump(by_alias=True)
+    )
+
+
+def test_builder_compiled_workflow_executes_like_declared_demo() -> None:
+    declared = build_demo_workflow()
+    declared_registry = build_demo_registry()
+    built, built_registry = build_authoring_demo_workflow()
+
+    declared_run = execute_workflow(
+        declared,
+        {"folder_id": "demo-folder", "should_email": False},
+        declared_registry,
+    )
+    built_run = execute_workflow(
+        built,
+        {"folder_id": "demo-folder", "should_email": False},
+        built_registry,
+    )
+
+    assert built_run.status == declared_run.status
+    assert built_run.output == declared_run.output
+    assert built_run.state == declared_run.state
+    assert built_run.current_node_id == declared_run.current_node_id

wf_authoring/__init__.py

@@ -0,0 +1,11 @@
+from .builder import WorkflowBuilder
+from .catalog import NodeCatalog, NodeCatalogEntry
+from .spec import NodeSpec, node
+
+__all__ = [
+    "NodeCatalog",
+    "NodeCatalogEntry",
+    "NodeSpec",
+    "WorkflowBuilder",
+    "node",
+]

wf_authoring/builder.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from wf_core import (
+    ConditionNode,
+    Edge,
+    ForeachNode,
+    InterruptNode,
+    NodeUse,
+    SchemaRef,
+    StateSchema,
+    Workflow,
+)
+
+from .spec import NodeSpec
+
+
+@dataclass(slots=True)
+class WorkflowBuilder:
+    name: str
+    input_schema: SchemaRef
+    state_schema: StateSchema
+    output_schema: SchemaRef
+    start: str
+    node_specs: dict[str, NodeSpec[Any, Any]] = field(default_factory=dict)
+    nodes: list[Any] = field(default_factory=list)
+    edges: list[Edge] = field(default_factory=list)
+
+    def use(
+        self,
+        spec: NodeSpec[Any, Any],
+        *,
+        id: str,
+        in_map: dict[str, str] | None = None,
+        out_map: dict[str, str] | None = None,
+        desc: str | None = None,
+    ) -> NodeUse:
+        self.node_specs[spec.name] = spec
+        node = NodeUse(
+            id=id,
+            type="node",
+            node=spec.name,
+            desc=desc or spec.description,
+            in_map=in_map or {},
+            out_map=out_map or {},
+        )
+        self.nodes.append(node)
+        return node
+
+    def condition(self, *, id: str, check: Any) -> ConditionNode:
+        node = ConditionNode(id=id, type="condition", check=check)
+        self.nodes.append(node)
+        return node
+
+    def foreach(
+        self,
+        *,
+        id: str,
+        over: str,
+        as_: str,
+        mode: Literal["serial", "parallel"] = "serial",
+        on_item_error: Literal["fail", "collect", "skip"] = "fail",
+    ) -> ForeachNode:
+        node = ForeachNode.model_validate(
+            {
+                "id": id,
+                "type": "foreach",
+                "over": over,
+                "as": as_,
+                "mode": mode,
+                "on_item_error": on_item_error,
+            }
+        )
+        self.nodes.append(node)
+        return node
+
+    def interrupt(
+        self,
+        *,
+        id: str,
+        kind: str,
+        request_map: dict[str, str] | None = None,
+        out_map: dict[str, str] | None = None,
+        outcomes: list[str] | None = None,
+    ) -> InterruptNode:
+        node = InterruptNode(
+            id=id,
+            type="interrupt",
+            kind=kind,
+            request_map=request_map or {},
+            out_map=out_map or {},
+            outcomes=outcomes or ["submitted"],
+        )
+        self.nodes.append(node)
+        return node
+
+    def connect(self, from_: str, outcome: str, to: str) -> None:
+        self.edges.append(Edge.model_validate({"from": from_, "outcome": outcome, "to": to}))
+
+    def compile(self) -> Workflow:
+        node_defs = [spec.to_node_def() for spec in self.node_specs.values()]
+        return Workflow(
+            name=self.name,
+            input_schema=self.input_schema,
+            state_schema=self.state_schema,
+            output_schema=self.output_schema,
+            node_defs=node_defs,
+            start=self.start,
+            nodes=self.nodes,
+            edges=self.edges,
+        )

wf_authoring/catalog.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from .spec import NodeSpec
+
+
+@dataclass(slots=True)
+class NodeCatalogEntry:
+    name: str
+    description: str | None
+    outcomes: tuple[str, ...]
+    input_schema: dict[str, Any]
+    output_schema: dict[str, Any]
+
+    @classmethod
+    def from_spec(cls, spec: NodeSpec[Any, Any]) -> "NodeCatalogEntry":
+        return cls(
+            name=spec.name,
+            description=spec.description,
+            outcomes=spec.outcomes,
+            input_schema=spec.input_model.model_json_schema(),
+            output_schema=spec.output_model.model_json_schema(),
+        )
+
+
+@dataclass(slots=True)
+class NodeCatalog:
+    specs: dict[str, NodeSpec[Any, Any]]
+
+    @classmethod
+    def from_specs(cls, *specs: NodeSpec[Any, Any]) -> "NodeCatalog":
+        return cls(specs={spec.name: spec for spec in specs})
+
+    def entries(self) -> list[NodeCatalogEntry]:
+        return [NodeCatalogEntry.from_spec(spec) for spec in self.specs.values()]
+
+    def as_mcp_payload(self) -> dict[str, Any]:
+        return {
+            "nodes": [
+                {
+                    "name": entry.name,
+                    "description": entry.description,
+                    "outcomes": list(entry.outcomes),
+                    "input_schema": entry.input_schema,
+                    "output_schema": entry.output_schema,
+                }
+                for entry in self.entries()
+            ]
+        }

wf_authoring/spec.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any, Generic, TypeVar
+
+from pydantic import BaseModel
+
+from wf_core import NodeDef, RuntimeContext, SchemaRef
+
+InputT = TypeVar("InputT", bound=BaseModel)
+OutputT = TypeVar("OutputT", bound=BaseModel)
+
+
+def _schema_ref_for(model_type: type[BaseModel]) -> SchemaRef:
+    return SchemaRef.model_validate(model_type.model_json_schema())
+
+
+@dataclass(slots=True)
+class NodeSpec(Generic[InputT, OutputT]):
+    name: str
+    input_model: type[InputT]
+    output_model: type[OutputT]
+    outcomes: tuple[str, ...]
+    fn: Callable[[InputT, RuntimeContext], OutputT | dict[str, Any]]
+    description: str | None = None
+    is_async: bool = False
+
+    def __call__(
+        self,
+        payload: InputT,
+        ctx: RuntimeContext,
+    ) -> OutputT | dict[str, Any]:
+        return self.fn(payload, ctx)
+
+    def to_node_def(self) -> NodeDef:
+        return NodeDef(
+            name=self.name,
+            input_schema=_schema_ref_for(self.input_model),
+            output_schema=_schema_ref_for(self.output_model),
+            outcomes=list(self.outcomes),
+        )
+
+    def to_registry_handler(self) -> Callable[[dict[str, Any], RuntimeContext], dict[str, Any]]:
+        def handler(payload: dict[str, Any], ctx: RuntimeContext) -> dict[str, Any]:
+            parsed = self.input_model.model_validate(payload)
+            raw = self.fn(parsed, ctx)
+            if isinstance(raw, self.output_model):
+                return {"outcome": self.outcomes[0], "output": raw.model_dump()}
+            if isinstance(raw, dict):
+                return raw
+            raise TypeError(
+                f"node {self.name!r} returned unsupported value {type(raw)!r}"
+            )
+
+        return handler
+
+
+def node(
+    *,
+    name: str | None = None,
+    input_model: type[InputT],
+    output_model: type[OutputT],
+    outcomes: tuple[str, ...] = ("ok",),
+    description: str | None = None,
+) -> Callable[
+    [Callable[[InputT, RuntimeContext], OutputT | dict[str, Any]]],
+    NodeSpec[InputT, OutputT],
+]:
+    def decorator(
+        fn: Callable[[InputT, RuntimeContext], OutputT | dict[str, Any]],
+    ) -> NodeSpec[InputT, OutputT]:
+        resolved_name = name or getattr(fn, "__name__", "node")
+        return NodeSpec(
+            name=resolved_name,
+            input_model=input_model,
+            output_model=output_model,
+            outcomes=outcomes,
+            fn=fn,
+            description=description or fn.__doc__,
+            is_async=False,
+        )
+
+    return decorator

wf_core/__init__.py

@@ -7,6 +7,7 @@ from .model import (
     NodeDef,
     NodeResult,
     NodeUse,
+    SchemaRef,
     StateField,
     StateSchema,
     Workflow,
@@ -46,6 +47,7 @@ __all__ = [
     "NodeDef",
     "NodeResult",
     "NodeUse",
+    "SchemaRef",
     "StateField",
     "StateSchema",
     "NodeHandler",

wf_core/demo_workflow.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
+from typing import cast
 
 from .model import Workflow
 from .run_state import RuntimeContext
@@ -233,7 +234,8 @@ def summarize_documents(
 def combine_summaries(
     payload: dict[str, object], ctx: RuntimeContext
 ) -> dict[str, object]:
-    item_summaries = payload["item_summaries"]
+    raw_item_summaries = cast(list[object], payload["item_summaries"])
+    item_summaries = [str(item) for item in raw_item_summaries]
     return {
         "outcome": "ok",
         "output": {"summary": " | ".join(item_summaries)},

wf_core/model.py

@@ -31,8 +31,8 @@ class NodeDef(BaseModel):
     input_schema: SchemaRef
     output_schema: SchemaRef
     outcomes: list[str] = Field(min_length=1)
-    retry: int | None = Field(None, ge=0)
+    retry: int | None = Field(default=None, ge=0)
-    timeout_seconds: int | None = Field(None, gt=0)
+    timeout_seconds: int | None = Field(default=None, gt=0)
 
 
 class NodeUse(BaseModel):
@@ -42,8 +42,8 @@ class NodeUse(BaseModel):
     desc: str | None = None
     in_map: dict[str, str] = Field(default_factory=dict)
     out_map: dict[str, str] = Field(default_factory=dict)
-    retry: int | None = Field(None, ge=0)
+    retry: int | None = Field(default=None, ge=0)
-    timeout_seconds: int | None = Field(None, gt=0)
+    timeout_seconds: int | None = Field(default=None, gt=0)
 
 
 class PathOperand(BaseModel):
@@ -91,6 +91,8 @@ class ConditionNode(BaseModel):
 
 
 class ForeachNode(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
     id: str
     type: Literal["foreach"]
     over: str

wf_core/node_exec.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from typing import Any
 
 from .conditions import safe_resolve_path
@@ -19,7 +19,7 @@ def execute_node_use(
     run: RunState,
     node: NodeUse,
     node_def: NodeDef,
-    registry: dict[str, NodeHandler],
+    registry: Mapping[str, NodeHandler],
 ) -> StepExecutionResult:
     handler = registry.get(node.node)
     if handler is None:

wf_core/runtime.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any
 
 from .errors import WorkflowExecutionError
@@ -43,7 +44,7 @@ __all__ = [
 def execute_workflow(
     workflow: Workflow,
     workflow_input: dict[str, Any],
-    registry: dict[str, NodeHandler],
+    registry: Mapping[str, NodeHandler],
 ) -> RunState:
     run = create_run_state(workflow, workflow_input)
 
@@ -62,7 +63,7 @@ def execute_workflow(
 def resume_workflow(
     workflow: Workflow,
     run: RunState,
-    registry: dict[str, NodeHandler],
+    registry: Mapping[str, NodeHandler],
     *,
     resume_payload: dict[str, Any] | None = None,
     resume_outcome: str = "submitted",
@@ -121,7 +122,7 @@ def resume_workflow(
 def step_workflow(
     workflow: Workflow,
     run: RunState,
-    registry: dict[str, NodeHandler],
+    registry: Mapping[str, NodeHandler],
     *,
     index: WorkflowIndex | None = None,
 ) -> RunState:

wf_core/step_handlers.py

@@ -4,7 +4,7 @@ from .conditions import eval_condition
 from .flow_ops import append_trace
 from .frame_ops import frame_context_values
 from .interrupt_ops import build_interrupt_request
-from .model import ConditionNode, InterruptNode, JoinNode
+from .model import ConditionNode, InterruptNode
 from .run_state import FrameStatus, RunState, RunStatus, StepExecutionResult
 
 

wf_core/validate.py

@@ -390,7 +390,7 @@ def _declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> se
         return {"loop", "done"}
     if step.type == "join":
         return {"done"}
-    if step.type == "interrupt":
+    if isinstance(step, InterruptNode):
         return set(step.outcomes)
     return set()