end node is real

docs/current_roadmap.md

@@ -47,10 +47,12 @@ implementation state.
   [2026-05-24 native subgraphs](./superpowers/specs/2026-05-24-native-subgraphs-design.md).
 - **Native subgraphs / graph-as-node**: next major runtime feature. A
   first-class `SubgraphNode` placeholder exists and validates parent-side
-  bindings/outcomes, but runtime execution still needs child run/frame identity,
-  child trace preservation, interrupt bubbling, and resume back into the child
-  workflow. Wrapper helpers currently run child workflows as ordinary nodes;
-  true graph-as-node behavior belongs here.
+  bindings/outcomes. Core workflows now also declare workflow-level outcomes
+  and can terminate through explicit `EndNode` steps. Runtime subgraph execution
+  still needs child run/frame identity, child trace preservation, interrupt
+  bubbling, and resume back into the child workflow. Wrapper helpers currently
+  run child workflows as ordinary nodes; true graph-as-node behavior belongs
+  here.
 - **Concurrent foreach**: implemented in core with explicit scheduling,
   reducer/merge semantics, item error policy, async handler batching, and
   quiescent interrupt behavior. Remaining work is polish and future reuse of

docs/superpowers/specs/2026-05-24-native-subgraphs-design.md

@@ -261,13 +261,13 @@ When the child workflow completes:
 2. Apply the subgraph step `output` bindings from child output into parent
    state.
 3. Record a parent `subgraph` trace entry with committed parent state changes.
-4. Advance the parent subgraph frame through outcome `ok`.
+4. Advance the parent subgraph frame through the child workflow outcome.
 
-For v1, a child workflow completion maps to one parent outcome: `ok`.
-Later, saved workflow artifacts may declare multiple outcomes, but core
-`Workflow.output_schema` is currently one output shape. Outcome-per-child-graph
-needs a separate design if we want a subgraph to behave exactly like a
-multi-outcome node.
+Core workflows now declare `Workflow.outcomes`, and explicit `EndNode` steps
+set `RunState.outcome`. The legacy `__end__` token remains compatibility
+shorthand for workflow outcome `ok`. Native subgraph execution should use that
+workflow-level outcome as the parent-visible subgraph outcome, instead of
+guessing from the child node that happened to route to a terminal.
 
 ## Failure Semantics
 

docs/wf_core_architecture.md

@@ -95,6 +95,7 @@ See `examples/raw_concurrent_foreach.py` for the canonical raw workflow shape an
 - validate start node existence
 - validate edge sources, destinations, duplicate outcomes, and declared outcomes
 - validate reachable nodes have all required outcome edges
+- validate explicit `EndNode` outcomes against `Workflow.outcomes`
 
 Validation reports multiple issues through `ValidationReport` instead of
 raising at the first failure.

readme.md

@@ -207,7 +207,10 @@ Edge
 
 - source nodes declare which outcomes are possible
 - edges map those outcomes to next nodes
-- terminal routing can go to builtin `__end__`
+- terminal routing can go to builtin `__end__`, which is compatibility
+  shorthand for workflow outcome `ok`
+- explicit `EndNode` steps set non-`ok` workflow outcomes such as `error` or
+  `needs_input`
 
 Reaching an undeclared or unwired outcome is runtime failure.
 
@@ -302,7 +305,7 @@ Executor steps:
 7. Validate typed node output
 8. Commit mapped output into state
 9. Route by returned outcome
-10. Stop when routing reaches `__end__`
+10. Stop when routing reaches `__end__` or an explicit `EndNode`
 11. Derive and validate final output from state
 
 Commit rules:

src/wf_core/__init__.py

@@ -1,5 +1,6 @@
 from .models import (
     ConditionNode,
+    EndNode,
     Edge,
     ForeachConcurrentPolicy,
     ForeachItemErrorPolicy,
@@ -51,6 +52,7 @@ from .validation import (
 __all__ = [
     "ConditionNode",
     "Edge",
+    "EndNode",
     "ForeachConcurrentPolicy",
     "ForeachItemErrorPolicy",
     "ForeachNode",

src/wf_core/models/__init__.py

@@ -13,6 +13,7 @@ from wf_core.models.reducers import ReducerRef, ReducerSpec, SiblingWritePolicy
 from wf_core.models.schemas import NodeDef, SchemaRef, StateField, StateSchema
 from wf_core.models.steps import (
     ConditionNode,
+    EndNode,
     ForeachConcurrentPolicy,
     ForeachItemErrorPolicy,
     ForeachNode,
@@ -29,6 +30,7 @@ __all__ = [
     "Condition",
     "ConditionNode",
     "Edge",
+    "EndNode",
     "ExistsCondition",
     "ForeachConcurrentPolicy",
     "ForeachItemErrorPolicy",

src/wf_core/models/steps.py

@@ -312,6 +312,20 @@ class JoinNode(BaseModel):
     type: Literal["join"]
 
 
+class EndNode(BaseModel):
+    """Explicit workflow terminal that sets the workflow-level outcome.
+
+    `__end__` remains the compatibility shorthand for outcome ``ok``. New
+    workflows that need business outcomes such as ``error`` or ``needs_input``
+    should route to explicit end nodes so the terminal contract is visible in
+    the graph.
+    """
+
+    id: str
+    type: Literal["end"]
+    outcome: str = "ok"
+
+
 class InterruptNode(BaseModel):
     """Control-flow step that pauses a run and waits for resume input."""
 
@@ -372,7 +386,13 @@ class InterruptNode(BaseModel):
 
 
 Step = Annotated[
-    NodeUse | SubgraphNode | ConditionNode | ForeachNode | JoinNode | InterruptNode,
+    NodeUse
+    | SubgraphNode
+    | ConditionNode
+    | ForeachNode
+    | JoinNode
+    | EndNode
+    | InterruptNode,
     Field(discriminator="type"),
 ]
 """Discriminated union of all executable workflow graph steps."""

src/wf_core/models/workflow.py

@@ -28,6 +28,7 @@ class Workflow(BaseModel):
     state_schema: StateSchema
     output_schema: SchemaRef
     node_defs: list[NodeDef] = Field(default_factory=list)
+    outcomes: list[str] = Field(default_factory=lambda: ["ok"], min_length=1)
     start: str
     nodes: list[Step]
     edges: list[Edge]

src/wf_core/run_state.py

@@ -132,6 +132,7 @@ class RunState:
     status: RunStatus
     workflow_input: dict[str, Any]
     state: dict[str, Any]
+    outcome: str | None = None
     output: dict[str, Any] = field(default_factory=dict)
     trace: list[TraceEntry] = field(default_factory=list)
     frames: dict[str, ExecutionFrame] = field(default_factory=dict)

src/wf_core/runtime/ops/flow.py

@@ -90,6 +90,8 @@ def advance_frame(
 
 
 def finalize_run(workflow: Workflow, run: RunState) -> RunState:
+    if run.outcome is None:
+        run.outcome = "ok"
     run.output = project_output(workflow, run.state)
     validate_payload_against_schema(
         workflow.output_schema, run.output, "workflow output"

src/wf_core/runtime/step.py

@@ -7,6 +7,7 @@ from typing import Any
 from wf_core.errors import WorkflowExecutionError
 from wf_core.models.steps import (
     ConditionNode,
+    EndNode,
     ForeachNode,
     InterruptNode,
     JoinNode,
@@ -37,7 +38,8 @@ from wf_core.runtime.scheduler import (
     select_next_frame,
     wake_parent_for_child_progress,
 )
-from wf_core.run_state import ExecutionFrame, FrameStatus, RunState
+from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, StepExecutionResult
+from wf_core.tokens import END
 
 from .preparation import prepare_step
 
@@ -74,6 +76,33 @@ def complete_step(
     return run
 
 
+def complete_end_step(
+    *,
+    run: RunState,
+    frame_id: str,
+    node_id: str,
+    outcome: str,
+) -> RunState:
+    """Record an explicit workflow terminal and complete the active frame."""
+    result = StepExecutionResult(outcome=outcome)
+    run.outcome = outcome
+    append_step_result_trace(
+        run,
+        frame_id=frame_id,
+        node_id=node_id,
+        step_type="end",
+        next_node_id=END,
+        result=result,
+    )
+    advance_frame(
+        run,
+        run.frames[frame_id],
+        outcome=outcome,
+        next_node_id=END,
+    )
+    return run
+
+
 def step_workflow(
     workflow: Workflow,
     run: RunState,
@@ -112,6 +141,13 @@ def step_workflow(
         step_result = handle_condition_step(run, step)
     elif isinstance(step, JoinNode):
         step_result = handle_join_step()
+    elif isinstance(step, EndNode):
+        return complete_end_step(
+            run=run,
+            frame_id=frame.id,
+            node_id=frame.node_id,
+            outcome=step.outcome,
+        )
     elif isinstance(step, InterruptNode):
         return handle_interrupt_step(run, step)
     elif isinstance(step, ForeachNode):
@@ -211,6 +247,13 @@ async def step_workflow_async(
         step_result = handle_condition_step(run, step)
     elif isinstance(step, JoinNode):
         step_result = handle_join_step()
+    elif isinstance(step, EndNode):
+        return complete_end_step(
+            run=run,
+            frame_id=frame.id,
+            node_id=frame.node_id,
+            outcome=step.outcome,
+        )
     elif isinstance(step, InterruptNode):
         return handle_interrupt_step(run, step)
     elif isinstance(step, ForeachNode):

src/wf_core/validation/core.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from wf_core.models.steps import (
     ConditionNode,
+    EndNode,
     ForeachNode,
     InterruptNode,
     NodeUse,
@@ -9,7 +10,7 @@ from wf_core.models.steps import (
     SubgraphNode,
 )
 from wf_core.models.schemas import NodeDef
-from wf_core.models.workflow import Edge, Workflow
+from wf_core.models.workflow import Workflow
 from wf_core.tokens import END
 from wf_core.validation.issues import ValidationIssueCode, ValidationReport
 from wf_core.validation.outcomes import declared_outcomes_for_step, reachable_node_ids
@@ -28,7 +29,7 @@ def validate_workflow(workflow: Workflow) -> ValidationReport:
     node_defs = _collect_node_defs(workflow, report)
     nodes_by_id = _validate_nodes(workflow, node_defs, report)
     _validate_start(workflow, nodes_by_id, report)
-    outgoing = _validate_edges(workflow.edges, nodes_by_id, node_defs, report)
+    outgoing = _validate_edges(workflow, nodes_by_id, node_defs, report)
     _validate_reachable_outcomes(workflow, nodes_by_id, node_defs, outgoing, report)
 
     return report
@@ -73,6 +74,8 @@ def _validate_nodes(
             validate_node_use(node, index, node_defs, workflow, report)
         elif isinstance(node, SubgraphNode):
             validate_subgraph_node(node, index, workflow, report)
+        elif isinstance(node, EndNode):
+            _validate_end_node(node, index, workflow, report)
         elif isinstance(node, ConditionNode):
             validate_condition_node(
                 node, index, report, state_root_fields, input_root_fields
@@ -94,6 +97,21 @@ def _validate_nodes(
     return nodes_by_id
 
 
+def _validate_end_node(
+    node: EndNode,
+    index: int,
+    workflow: Workflow,
+    report: ValidationReport,
+) -> None:
+    """Validate explicit workflow terminal outcomes."""
+    if node.outcome not in workflow.outcomes:
+        report.add(
+            ValidationIssueCode.UNDECLARED_WORKFLOW_OUTCOME,
+            f"nodes[{index}].outcome",
+            f"workflow outcome {node.outcome!r} is not declared",
+        )
+
+
 def _validate_start(
     workflow: Workflow,
     nodes_by_id: dict[str, Step],
@@ -108,7 +126,7 @@ def _validate_start(
 
 
 def _validate_edges(
-    edges: list[Edge],
+    workflow: Workflow,
     nodes_by_id: dict[str, Step],
     node_defs: dict[str, NodeDef],
     report: ValidationReport,
@@ -116,7 +134,7 @@ def _validate_edges(
     outgoing: dict[str, set[str]] = {}
     edge_keys: set[tuple[str, str]] = set()
 
-    for index, edge in enumerate(edges):
+    for index, edge in enumerate(workflow.edges):
         edge_key = (edge.from_, edge.outcome)
         if edge_key in edge_keys:
             report.add(
@@ -150,6 +168,14 @@ def _validate_edges(
                 f"edges[{index}].to",
                 f"unknown destination node {edge.to!r}",
             )
+        if edge.to == END and "ok" not in workflow.outcomes:
+            # `__end__` is the legacy implicit end node for workflow outcome
+            # "ok". Explicit non-ok outcomes should use EndNode instead.
+            report.add(
+                ValidationIssueCode.UNDECLARED_WORKFLOW_OUTCOME,
+                f"edges[{index}].to",
+                "legacy __end__ requires workflow outcome 'ok' to be declared",
+            )
 
     return outgoing
 

src/wf_core/validation/issues.py

@@ -13,6 +13,7 @@ class ValidationIssueCode(StrEnum):
     UNKNOWN_EDGE_DESTINATION = "unknown_edge_destination"
     UNDECLARED_EDGE_OUTCOME = "undeclared_edge_outcome"
     MISSING_OUTCOME_EDGE = "missing_outcome_edge"
+    UNDECLARED_WORKFLOW_OUTCOME = "undeclared_workflow_outcome"
     UNKNOWN_NODE_DEF = "unknown_node_def"
     INVALID_NODE_INPUT_FIELD = "invalid_node_input_field"
     INVALID_SOURCE_PATH = "invalid_source_path"

src/wf_core/validation/outcomes.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from wf_core.models.schemas import NodeDef
-from wf_core.models.steps import InterruptNode, NodeUse, Step, SubgraphNode
+from wf_core.models.steps import EndNode, InterruptNode, NodeUse, Step, SubgraphNode
 from wf_core.models.workflow import Edge
 from wf_core.tokens import END
 
@@ -21,6 +21,8 @@ def declared_outcomes_for_step(step: Step, node_defs: dict[str, NodeDef]) -> set
         return outcomes
     if step.type == "join":
         return {"done"}
+    if isinstance(step, EndNode):
+        return set()
     if isinstance(step, InterruptNode):
         return set(step.outcomes)
     return set()

tests/core/test_workflow_outcomes.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+from wf_core import END, Workflow
+from wf_core.runtime import execute_workflow
+from wf_core.validation.issues import ValidationIssueCode
+
+
+def test_legacy_end_token_completes_with_ok_workflow_outcome() -> None:
+    workflow = _workflow(edges=[{"from": "finish", "outcome": "done", "to": END}])
+
+    run = execute_workflow(workflow, {"text": "hello"}, {"finish": _finish})
+
+    assert run.status == "completed"
+    assert run.outcome == "ok"
+
+
+def test_explicit_end_node_sets_workflow_outcome() -> None:
+    workflow = _workflow(
+        outcomes=["ok", "error"],
+        nodes=[
+            _finish_node_data(),
+            {"id": "end_error", "type": "end", "outcome": "error"},
+        ],
+        edges=[{"from": "finish", "outcome": "done", "to": "end_error"}],
+    )
+
+    run = execute_workflow(workflow, {"text": "hello"}, {"finish": _finish})
+
+    assert run.status == "completed"
+    assert run.outcome == "error"
+
+
+def test_validation_rejects_end_node_outcome_not_declared_by_workflow() -> None:
+    workflow = _workflow(
+        nodes=[
+            _finish_node_data(),
+            {"id": "end_error", "type": "end", "outcome": "error"},
+        ],
+        edges=[{"from": "finish", "outcome": "done", "to": "end_error"}],
+    )
+
+    report = workflow.validate_structure()
+
+    assert any(
+        issue.code == ValidationIssueCode.UNDECLARED_WORKFLOW_OUTCOME
+        and issue.path == "nodes[1].outcome"
+        for issue in report.errors
+    )
+
+
+def test_validation_rejects_legacy_end_without_ok_workflow_outcome() -> None:
+    workflow = _workflow(
+        outcomes=["error"],
+        edges=[{"from": "finish", "outcome": "done", "to": END}],
+    )
+
+    report = workflow.validate_structure()
+
+    assert any(
+        issue.code == ValidationIssueCode.UNDECLARED_WORKFLOW_OUTCOME
+        and issue.path == "edges[0].to"
+        for issue in report.errors
+    )
+
+
+def _finish(payload: dict[str, object], _ctx: object) -> dict[str, object]:
+    return {"outcome": "done", "output": {"echoed": payload["text"]}}
+
+
+def _workflow(
+    *,
+    outcomes: list[str] | None = None,
+    nodes: list[dict[str, object]] | None = None,
+    edges: list[dict[str, object]],
+) -> Workflow:
+    return Workflow.model_validate(
+        {
+            "name": "workflow_outcomes",
+            "input_schema": {
+                "type": "object",
+                "properties": {"text": {"type": "string"}},
+                "required": ["text"],
+            },
+            "state_schema": {
+                "type": "object",
+                "properties": {"echoed": {"type": "string"}},
+            },
+            "output_schema": {
+                "type": "object",
+                "properties": {"echoed": {"type": "string"}},
+                "required": ["echoed"],
+            },
+            "node_defs": [
+                {
+                    "name": "finish",
+                    "input_schema": {
+                        "type": "object",
+                        "properties": {"text": {"type": "string"}},
+                        "required": ["text"],
+                    },
+                    "output_schema": {
+                        "type": "object",
+                        "properties": {"echoed": {"type": "string"}},
+                        "required": ["echoed"],
+                    },
+                    "outcomes": ["done"],
+                }
+            ],
+            "outcomes": outcomes or ["ok"],
+            "start": "finish",
+            "nodes": [_finish_node_data()] if nodes is None else nodes,
+            "edges": edges,
+        }
+    )
+
+
+def _finish_node_data() -> dict[str, object]:
+    return {
+        "id": "finish",
+        "type": "node",
+        "node": "finish",
+        "input": [{"target": "text", "path": "input.text"}],
+        "output": [{"source": "echoed", "target": "state.echoed"}],
+    }