project workflow state -> output

docs/schema_validation.md

@@ -30,6 +30,15 @@ This means schema fields are contracts for authoring, planning, documentation,
 and mapping validation. Runtime state merge behavior is separate metadata on
 declared exact state paths; undeclared paths still use `replace`.
 
+Final workflow output has two projection modes:
+
+- If `Workflow.output` contains bindings, those bindings build the public output
+  payload from graph paths such as `state.result.message`.
+- If `Workflow.output` is empty, legacy projection copies same-named top-level
+  state keys listed in `workflow.output_schema.properties`.
+
+The projected payload is then validated against `workflow.output_schema`.
+
 ## Why This Matters
 
 Node and workflow boundaries can now reject wrong primitive/container types when

docs/wf_core_architecture.md

@@ -41,7 +41,9 @@ flat modules.
 8. `runtime.ops.nodes` handles `NodeUse` input projection, handler invocation,
    output validation, and state writes.
 9. `runtime.ops.flow` records trace entries and advances frames.
-10. Completion projects workflow output from state and validates it.
+10. Completion projects workflow output and validates it. Explicit
+    `Workflow.output` bindings are used when present; older workflows fall back
+    to same-name top-level state projection from `output_schema.properties`.
 
 Async execution shares the same runtime model. The async seam is handler
 invocation; control-flow steps are still synchronous state transitions.

src/wf_artifacts/drafts/adapter.py

@@ -21,9 +21,8 @@ from .models import (
 def build_workflow_from_draft(draft: WorkflowDraft) -> Workflow:
     """Adapt one typed draft through `WorkflowBuilder` into a core workflow.
 
-    Draft step `output` bindings become node-output-to-state writes. Final
-    workflow output projection stays in core runtime and uses output schema
-    property names as state keys.
+    Draft step `output` bindings become node-output-to-state writes. Draft
+    top-level `output` bindings become core root workflow output projection.
     """
     builder = WorkflowBuilder(
         name=draft.name,
@@ -39,7 +38,8 @@ def build_workflow_from_draft(draft: WorkflowDraft) -> Workflow:
     for source_id, routes in draft.routes.items():
         for outcome, target in routes.items():
             builder.connect(step_refs[source_id], outcome, target)
-    return builder.compile()
+    workflow = builder.compile()
+    return workflow.model_copy(update={"output": list(draft.output)})
 
 
 def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):

src/wf_artifacts/drafts/models.py

@@ -30,9 +30,8 @@ STEP_KIND_KEYS = frozenset(
 class DraftUseStep(BaseModel):
     """Draft step that calls one externally resolvable workflow capability.
 
-    `output` writes node-local output fields into workflow state. It does not
-    define final workflow output; core currently projects final output from
-    state keys whose names match `WorkflowDraft.output_schema.properties`.
+    `output` writes node-local output fields into workflow state. Root
+    workflow output bindings live on `WorkflowDraft.output`.
     """
 
     model_config = ConfigDict(extra="forbid", populate_by_name=True)
@@ -277,16 +276,16 @@ DraftStep = (
 class WorkflowDraft(BaseModel):
     """Patch-friendly JSON authoring document for one workflow graph.
 
-    There is intentionally no top-level output map in this draft shape. Final
-    workflow output is projected from state by matching output-schema property
-    names, so terminal steps should write required output fields to same-named
-    state paths.
+    Step `output` writes node results into state. Top-level `output` projects
+    final workflow output from graph paths. If top-level `output` is empty, core
+    falls back to same-name top-level state projection.
     """
 
     name: str
     input_schema: JsonObject
     state_schema: JsonObject
     output_schema: JsonObject
+    output: list[InputBinding] = Field(default_factory=list)
     start: str
     steps: dict[str, DraftStep]
     routes: dict[str, dict[str, str]] = Field(default_factory=dict)

src/wf_core/models/workflow.py

@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, cast
 from pydantic import BaseModel, Field
 
 from wf_core.models.schemas import NodeDef, SchemaRef, StateSchema
-from wf_core.models.steps import Step
+from wf_core.models.steps import InputBinding, Step
 
 if TYPE_CHECKING:
     from wf_core.validation.issues import ValidationReport
@@ -27,6 +27,15 @@ class Workflow(BaseModel):
     input_schema: SchemaRef
     state_schema: StateSchema
     output_schema: SchemaRef
+    output: list[InputBinding] = Field(
+        default_factory=list,
+        description=(
+            "Optional final output projection bindings. Sources read from graph "
+            "paths such as state.result, and targets write into the workflow "
+            "output payload. When omitted, legacy same-name top-level state "
+            "projection is used."
+        ),
+    )
     node_defs: list[NodeDef] = Field(default_factory=list)
     outcomes: list[str] = Field(default_factory=lambda: ["ok"], min_length=1)
     start: str

src/wf_core/runtime/ops/flow.py

@@ -96,7 +96,11 @@ 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)
+    run.output = project_output(
+        workflow,
+        run.state,
+        workflow_input=run.workflow_input,
+    )
     validate_payload_against_schema(
         workflow.output_schema, run.output, "workflow output"
     )

src/wf_core/runtime/ops/state.py

@@ -5,11 +5,22 @@ from copy import deepcopy
 from dataclasses import dataclass, field as dataclass_field
 from typing import Any
 
+from wf_core.conditions import safe_resolve_path
 from wf_core.errors import WorkflowExecutionError
-from wf_core.local_paths import LocalPathError, get_local_value, has_overlapping_paths
+from wf_core.local_paths import (
+    LocalPathError,
+    get_local_value,
+    has_overlapping_paths,
+    set_local_value,
+)
 from wf_core.models.reducers import ReducerRef
 from wf_core.models.schemas import StateFieldDecl
-from wf_core.models.steps import NodeUse, OutputBinding
+from wf_core.models.steps import (
+    InputPathBinding,
+    InputValueBinding,
+    NodeUse,
+    OutputBinding,
+)
 from wf_core.models.workflow import Workflow
 from wf_core.paths import (
     PathResolutionError,
@@ -437,12 +448,38 @@ def reducer_for_state_path(
     )
 
 
-def project_output(workflow: Workflow, state: dict[str, Any]) -> dict[str, Any]:
-    """Project final workflow output from same-named state fields.
+def project_output(
+    workflow: Workflow,
+    state: dict[str, Any],
+    *,
+    workflow_input: Mapping[str, Any] | None = None,
+    context: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Project final workflow output from explicit bindings or state fields.
 
-    Node output bindings write into state during execution. At END, the runtime
-    exposes only state keys declared by `workflow.output_schema.properties`.
+    `workflow.output` is the canonical root-output mapping for workflows whose
+    public output shape does not mirror top-level state keys. Older workflows
+    without explicit output bindings keep the same-name state projection.
     """
+    if workflow.output:
+        output: dict[str, Any] = {}
+        for binding in workflow.output:
+            if isinstance(binding, InputValueBinding):
+                value = binding.value
+            elif isinstance(binding, InputPathBinding):
+                value = safe_resolve_path(
+                    str(binding.path),
+                    state=state,
+                    workflow_input=workflow_input or {},
+                    context=context or {},
+                )
+            else:
+                raise WorkflowExecutionError("unsupported workflow output binding")
+            try:
+                set_local_value(output, binding.target, value)
+            except LocalPathError as exc:
+                raise WorkflowExecutionError(str(exc)) from exc
+        return output
     return {
         key: state[key] for key in workflow.output_schema.properties if key in state
     }

src/wf_core/runtime/subgraphs.py

@@ -249,7 +249,11 @@ def _finish_subgraph(
         raise WorkflowExecutionError(
             f"subgraph step {step.id!r} child completed without a workflow outcome"
         )
-    child_output = project_output(prepared.workflow, child_scope.committed_state)
+    child_output = project_output(
+        prepared.workflow,
+        child_scope.committed_state,
+        workflow_input=child_scope.workflow_input,
+    )
     validate_payload_against_schema(
         prepared.workflow.output_schema,
         child_output,

src/wf_core/validation/core.py

@@ -20,6 +20,7 @@ from wf_core.validation.steps import (
     validate_interrupt_node,
     validate_node_use,
     validate_subgraph_node,
+    validate_workflow_output_bindings,
 )
 
 
@@ -27,6 +28,7 @@ def validate_workflow(workflow: Workflow) -> ValidationReport:
     report = ValidationReport()
 
     node_defs = _collect_node_defs(workflow, report)
+    validate_workflow_output_bindings(workflow.output, workflow, report)
     nodes_by_id = _validate_nodes(workflow, node_defs, report)
     _validate_start(workflow, nodes_by_id, report)
     outgoing = _validate_edges(workflow, nodes_by_id, node_defs, report)

src/wf_core/validation/issues.py

@@ -18,6 +18,7 @@ class ValidationIssueCode(StrEnum):
     INVALID_NODE_INPUT_FIELD = "invalid_node_input_field"
     INVALID_SOURCE_PATH = "invalid_source_path"
     INVALID_NODE_OUTPUT_FIELD = "invalid_node_output_field"
+    INVALID_WORKFLOW_OUTPUT_FIELD = "invalid_workflow_output_field"
     INVALID_DESTINATION_PATH = "invalid_destination_path"
     EMPTY_CONDITION_ARGS = "empty_condition_args"
     INVALID_CONDITION_PATH = "invalid_condition_path"

src/wf_core/validation/steps.py

@@ -16,6 +16,7 @@ from wf_core.models.steps import (
     ForeachNode,
     InputBinding,
     InputPathBinding,
+    InputValueBinding,
     InterruptNode,
     NodeUse,
     OutputBinding,
@@ -153,6 +154,56 @@ def _validate_boundary_bindings(
         )
 
 
+def validate_workflow_output_bindings(
+    bindings: list[InputBinding],
+    workflow: Workflow,
+    report: ValidationReport,
+) -> None:
+    """Validate final workflow output projection bindings.
+
+    Root output bindings reuse input-binding shape: graph paths flow into a
+    local output payload that is later validated against `output_schema`.
+    """
+    output_fields = set(workflow.output_schema.properties)
+    state_root_fields = workflow.state_schema.root_fields()
+    input_root_fields = set(workflow.input_schema.properties)
+    output_targets = []
+    for output_index, binding in enumerate(bindings):
+        output_targets.append(binding.target)
+        destination_root = _local_root(binding.target)
+        if destination_root is None or (
+            destination_root != "." and destination_root not in output_fields
+        ):
+            report.add(
+                ValidationIssueCode.INVALID_WORKFLOW_OUTPUT_FIELD,
+                f"output[{output_index}].target",
+                "destination field is not declared in workflow output schema",
+            )
+        if isinstance(binding, InputPathBinding) and not is_valid_source_path(
+            binding.path,
+            state_root_fields,
+            input_root_fields,
+            allow_context=True,
+        ):
+            report.add(
+                ValidationIssueCode.INVALID_SOURCE_PATH,
+                f"output[{output_index}].path",
+                "source path must start with input., state., or context. and reference a declared root field when applicable",
+            )
+        elif not isinstance(binding, (InputPathBinding, InputValueBinding)):
+            report.add(
+                ValidationIssueCode.INVALID_WORKFLOW_OUTPUT_FIELD,
+                f"output[{output_index}]",
+                "unsupported workflow output binding",
+            )
+    if has_overlapping_paths(output_targets):
+        report.add(
+            ValidationIssueCode.INVALID_WORKFLOW_OUTPUT_FIELD,
+            "output",
+            "workflow output has overlapping output payload paths",
+        )
+
+
 def _local_root(path: str | LocalPath) -> str | None:
     try:
         parts = split_local_path(path)

src/wf_mcp/broker/service/core.py

@@ -663,6 +663,7 @@ class WfMcpService:
             "input_schema": plan.input_schema,
             "state_schema": plan.state_schema,
             "output_schema": plan.output_schema,
+            "output": [binding.model_dump(mode="json") for binding in plan.output],
             "start": plan.start,
             "node_defs": [node.model_dump() for node in node_defs.values()],
             "nodes": nodes,

src/wf_mcp/models.py

@@ -4,9 +4,9 @@ from dataclasses import asdict, dataclass, field
 from pathlib import Path
 from typing import Any
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from wf_core import Edge
-from wf_core.models.steps import Step
+from wf_core.models.steps import InputBinding, Step
 
 from .capabilities import CatalogNodeEntry, CatalogPromptEntry, CatalogResourceEntry
 
@@ -49,6 +49,13 @@ class RawWorkflowPlan(BaseModel):
     input_schema: dict[str, Any]
     state_schema: dict[str, Any]
     output_schema: dict[str, Any]
+    output: list[InputBinding] = Field(
+        default_factory=list,
+        description=(
+            "Optional root workflow output bindings. Sources read graph paths "
+            "such as state.result and targets write the public output payload."
+        ),
+    )
     start: str
     nodes: list[Step]
     edges: list[Edge]

tests/artifacts/test_draft_adapter.py

@@ -75,6 +75,38 @@ def test_adapter_lowers_use_steps_to_canonical_bindings() -> None:
     assert dumped["output"][0]["target"] == {"root": "state", "parts": ["echoed"]}
 
 
+def test_adapter_lowers_root_workflow_output_bindings() -> None:
+    draft = WorkflowDraft.model_validate(
+        {
+            "name": "echo",
+            "input_schema": {},
+            "state_schema": {
+                "type": "object",
+                "properties": {
+                    "raw": {
+                        "type": "object",
+                        "properties": {"echoed": {"type": "string"}},
+                    }
+                },
+            },
+            "output_schema": {
+                "type": "object",
+                "properties": {"message": {"type": "string"}},
+            },
+            "output": [{"target": "message", "path": "state.raw.echoed"}],
+            "start": "echo",
+            "steps": {"echo": {"use": "demo.echo"}},
+            "routes": {"echo": {"ok": "__end__"}},
+        }
+    )
+
+    workflow = build_workflow_from_draft(draft)
+    dumped = workflow.model_dump(mode="json")
+
+    assert dumped["output"][0]["target"] == {"root": "local", "parts": ["message"]}
+    assert dumped["output"][0]["path"] == {"root": "state", "parts": ["raw", "echoed"]}
+
+
 def test_adapter_lowers_static_inputs_for_constant_like_steps() -> None:
     draft = WorkflowDraft.model_validate(
         {

tests/core/test_workflow_outcomes.py

@@ -33,6 +33,81 @@ def test_explicit_end_node_sets_workflow_outcome() -> None:
     assert run.outcome == "error"
 
 
+def test_root_workflow_output_bindings_project_final_output() -> None:
+    workflow = Workflow.model_validate(
+        {
+            "name": "root_output_bindings",
+            "input_schema": {
+                "type": "object",
+                "properties": {"text": {"type": "string"}},
+                "required": ["text"],
+            },
+            "state_schema": {
+                "type": "object",
+                "properties": {
+                    "result": {
+                        "type": "object",
+                        "properties": {"echoed": {"type": "string"}},
+                    }
+                },
+            },
+            "output_schema": {
+                "type": "object",
+                "properties": {"echoed": {"type": "string"}},
+                "required": ["echoed"],
+            },
+            "output": [{"target": "echoed", "path": "state.result.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"],
+                }
+            ],
+            "start": "finish",
+            "nodes": [
+                {
+                    "id": "finish",
+                    "type": "node",
+                    "node": "finish",
+                    "input": [{"target": "text", "path": "input.text"}],
+                    "output": [{"source": "echoed", "target": "state.result.echoed"}],
+                }
+            ],
+            "edges": [{"from": "finish", "outcome": "done", "to": END}],
+        }
+    )
+
+    run = execute_workflow(workflow, {"text": "hello"}, {"finish": _finish})
+
+    assert run.output["echoed"] == "hello"
+    assert run.state["result"]["echoed"] == "hello"
+
+
+def test_validation_rejects_root_workflow_output_target_not_declared() -> None:
+    workflow = _workflow(
+        edges=[{"from": "finish", "outcome": "done", "to": END}],
+        output=[{"target": "missing", "path": "state.echoed"}],
+    )
+
+    report = workflow.validate_structure()
+
+    assert any(
+        issue.code == ValidationIssueCode.INVALID_WORKFLOW_OUTPUT_FIELD
+        and issue.path == "output[0].target"
+        for issue in report.errors
+    )
+
+
 def test_validation_rejects_end_node_outcome_not_declared_by_workflow() -> None:
     workflow = _workflow(
         nodes=[
@@ -80,6 +155,7 @@ def _workflow(
     outcomes: list[str] | None = None,
     nodes: list[dict[str, object]] | None = None,
     edges: list[dict[str, object]],
+    output: list[dict[str, object]] | None = None,
 ) -> Workflow:
     return Workflow.model_validate(
         {
@@ -115,6 +191,7 @@ def _workflow(
                 }
             ],
             "outcomes": outcomes or ["ok"],
+            "output": [] if output is None else output,
             "start": "finish",
             "nodes": [_finish_node_data()] if nodes is None else nodes,
             "edges": edges,

tests/wf_mcp/test_service.py

@@ -388,6 +388,55 @@ def test_service_compiles_and_runs_raw_plan() -> None:
     assert "workflow_run_completed" in event_kinds
 
 
+def test_service_preserves_raw_plan_root_output_bindings() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "root_output_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+
+    plan = _raw_plan(
+        name="demo_root_output_plan",
+        input_schema={
+            "type": "object",
+            "properties": {"text": {"type": "string"}},
+            "required": ["text"],
+        },
+        state_schema={
+            "type": "object",
+            "properties": {
+                "raw": {
+                    "type": "object",
+                    "properties": {"echoed": {"type": "string"}},
+                }
+            },
+        },
+        output_schema={
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+            "required": ["message"],
+        },
+        output=[{"target": "message", "path": "state.raw.echoed"}],
+        start="echo",
+        nodes=[
+            {
+                "id": "echo",
+                "type": "node",
+                "node": "demo.personal.echo_tool",
+                "input": [input_binding("input.text", "text")],
+                "output": [output_binding("echoed", "state.raw.echoed")],
+            }
+        ],
+        edges=[{"from": "echo", "outcome": "ok", "to": END}],
+    )
+
+    run = asyncio.run(service.run_workflow_from_plan(plan, {"text": "hello"}))
+
+    assert run.status == RunStatus.COMPLETED
+    assert run.output["message"] == "hello"
+    assert run.state["raw"]["echoed"] == "hello"
+
+
 def test_service_resolves_registered_spec_with_dotted_local_name() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "dotted_spec_store"))
     service.register_connection(