feat: add workflow output draft command

docs/current_roadmap.md

@@ -92,6 +92,10 @@ clear operator feedback before adding more architecture.
   and step bindings so agents can recover from bad edits without raw JSON Patch.
   Implementation:
   [`draft remove commands`](historical/superpowers/plans/2026-06-28-draft-remove-commands.md).
+- Completed: `wf draft set-workflow-output` and full-stack API/RPC/CLI support
+  for editing top-level workflow output bindings. Accepts repeatable `--map`
+  and `--merge` flag. Implementation:
+  [`set-workflow-output API/RPC/CLI`](historical/superpowers/plans/2026-06-29-set-workflow-output.md).
 - Keep status read-only; do not mutate registry, auth, config, or stores.
 
 ## Priority 2: Durable Run/Resume Hardening

docs/historical/superpowers/plans/2026-06-29-set-workflow-output.md

@@ -18,7 +18,7 @@
 - Modify: `src/wf_api/surface.py`
 - Test: `tests/wf_api/test_drafts_service.py`
 
-- [ ] **Step 1: Write failing API tests**
+- [x] **Step 1: Write failing API tests**
 
 Add tests that create a draft with empty `output`, call `set_workflow_output_map`, and assert the stored draft has top-level output bindings:
 
@@ -60,7 +60,7 @@ async def test_set_workflow_output_map_merges_top_level_output(tmp_path: Path) -
     ]
 ```
 
-- [ ] **Step 2: Run tests RED**
+- [x] **Step 2: Run tests RED**
 
 Run:
 
@@ -70,7 +70,7 @@ uv run pytest tests/wf_api/test_drafts_service.py::test_set_workflow_output_map_
 
 Expected: fail because `set_workflow_output_map` does not exist.
 
-- [ ] **Step 3: Implement API method**
+- [x] **Step 3: Implement API method**
 
 In `WorkflowDraftApi`, add:
 
@@ -111,11 +111,11 @@ async def set_workflow_output_map(
 
 Add delegates to `WorkflowApi` and `WorkflowDraftSurface`.
 
-- [ ] **Step 4: Run API tests GREEN**
+- [x] **Step 4: Run API tests GREEN**
 
 Run the two tests from Step 2. Expected: pass.
 
-- [ ] **Step 5: Commit**
+- [x] **Step 5: Commit**
 
 ```powershell
 git add src/wf_api/drafts.py src/wf_api/service.py src/wf_api/surface.py tests/wf_api/test_drafts_service.py
@@ -135,7 +135,7 @@ git commit -m "feat: edit workflow output map in draft api"
 - Test: `tests/wf_cli/test_remote_target.py`
 - Test: `tests/wf_cli/test_app.py`
 
-- [ ] **Step 1: Add failing transport and CLI tests**
+- [x] **Step 1: Add failing transport and CLI tests**
 
 Add RPC app/client tests that call `workflow.draft_workspaces.set_workflow_output_map` with `{"state.echoed": "message"}`.
 
@@ -177,7 +177,7 @@ def test_wf_draft_set_workflow_output_uses_rpc_target(monkeypatch: pytest.Monkey
     ]
 ```
 
-- [ ] **Step 2: Run tests RED**
+- [x] **Step 2: Run tests RED**
 
 Run:
 
@@ -187,7 +187,7 @@ uv run pytest tests/wf_transport_rpc_http/test_app.py::test_rpc_draft_workspace_
 
 Expected: fail because DTO/method/command are missing.
 
-- [ ] **Step 3: Implement transport and CLI**
+- [x] **Step 3: Implement transport and CLI**
 
 Add `SetWorkflowOutputMapParams` with `workspace_id`, `revision`, `output_map`, `merge`.
 
@@ -228,11 +228,11 @@ def set_workflow_output(...):
 
 Use the existing `_parse_map_flags` helper. Add `--merge` with the same replace/merge wording used by `set-input` and `set-output`.
 
-- [ ] **Step 4: Run tests GREEN**
+- [x] **Step 4: Run tests GREEN**
 
 Run the tests from Step 2. Expected: pass.
 
-- [ ] **Step 5: Commit**
+- [x] **Step 5: Commit**
 
 ```powershell
 git add src/wf_transport_rpc_http src/wf_cli/commands/drafts.py tests/wf_transport_rpc_http tests/wf_cli
@@ -248,7 +248,7 @@ git commit -m "feat: expose workflow output draft command"
 - Modify: `skills/wf-workflow/references/draft-workspaces.md`
 - Modify: `docs/current_roadmap.md`
 
-- [ ] **Step 1: Document command**
+- [x] **Step 1: Document command**
 
 Add example:
 
@@ -261,7 +261,7 @@ wf draft set-workflow-output report_ws \
 
 State clearly: this edits top-level `WorkflowDraft.output`; step-level `wf draft set-output` edits one step's node-output-to-state bindings.
 
-- [ ] **Step 2: Update skills**
+- [x] **Step 2: Update skills**
 
 Add a rule:
 
@@ -270,7 +270,7 @@ Use `wf draft set-workflow-output` for final workflow output projection.
 Use `wf draft set-output` only for step output bindings.
 ```
 
-- [ ] **Step 3: Verify**
+- [x] **Step 3: Verify**
 
 Run:
 
@@ -280,7 +280,7 @@ uv run ruff check src/wf_api src/wf_cli src/wf_transport_rpc_http tests/wf_api t
 uv run basedpyright --level error src/wf_api/drafts.py src/wf_cli/commands/drafts.py src/wf_transport_rpc_http tests/wf_api/test_drafts_service.py tests/wf_cli/test_remote_target.py
 ```
 
-- [ ] **Step 4: Commit**
+- [x] **Step 4: Commit**
 
 ```powershell
 git add docs/wf_cli.md docs/workflow_drafts.md skills/wf-cli/SKILL.md skills/wf-workflow/references/draft-workspaces.md docs/current_roadmap.md

docs/superpowers/plans/2026-06-29-draft-bind-repair-hints.md

@@ -2,34 +2,48 @@
 
 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
 
-**Goal:** Make schema-aware `wf draft bind` discoverable from validation output for workflow input, state, and workflow output projection errors.
+**Goal:** Make schema-aware `wf draft bind` correctly support workflow output projection and make focused bind repairs discoverable from validation output.
 
-**Architecture:** Draft validation already enriches diagnostics through `_with_workspace_repair_hints()` in `src/wf_api/drafts.py`. Extend `_draft_repair_hint()` beyond `invalid_destination_path` so agents get concrete `wf draft bind` or `wf draft set-workflow-output` commands instead of falling to JSON Patch.
+**Architecture:** Node output bindings can only target workflow state. Therefore `local.x -> output.y` must lower atomically into `local.x -> state.y` plus top-level `state.y -> output.y`, while projecting the capability field schema into both state and output schemas. Draft validation already enriches diagnostics through `_with_workspace_repair_hints()` in `src/wf_api/drafts.py`; extend those hints after the bind behavior is correct.
 
 **Tech Stack:** Python 3.14, validation diagnostics, Typer CLI help/docs, pytest.
 
 ---
 
-### Task 1: Repair Hint For Missing Workflow Output Schema
+### Task 1: Make `bind local -> output` Produce Valid Workflow Output
 
 **Files:**
-- Modify: `src/wf_api/drafts.py`
+- Modify: `src/wf_api/draft_authoring.py`
 - Test: `tests/wf_api/test_drafts_service.py`
 - Test: `tests/wf_cli/test_remote_target.py`
 
 - [ ] **Step 1: Write failing tests**
 
-Add a draft with top-level output binding `{"path": "state.markdown", "target": "markdown"}` and empty `output_schema.properties`. Validate the workspace and assert the diagnostic has:
+Add a capability-backed `render` step whose capability output schema declares `markdown`. Call:
 
 ```python
-assert diagnostic["code"] == "invalid_workflow_output_field"
+result = await authoring.bind_draft(
-assert "wf draft bind" in diagnostic["repair_hint"] or "wf draft set-workflow-output" in diagnostic["repair_hint"]
+    workspace_id="report",
+    revision=1,
+    step_id="render",
+    source_path="local.markdown",
+    target_path="output.markdown",
+)
 ```
 
-If the source came from a capability local output in the same step, prefer a `wf draft bind ... --from local.markdown --to output.markdown` hint. If the diagnostic lacks step/local context, use:
+Assert the edit is valid and lowered through state:
 
-```text
+```python
-wf draft set-workflow-output <workspace> --revision <n> --map state.markdown=markdown
+assert result["status"] == "valid"
+workspace = await drafts.get_draft_workspace(workspace_id="report", include_draft=True)
+assert workspace["draft"]["steps"]["render"]["output"] == [
+    {"source": "markdown", "target": "state.markdown"}
+]
+assert workspace["draft"]["output"] == [
+    {"path": "state.markdown", "target": "markdown"}
+]
+assert workspace["draft"]["state_schema"]["properties"]["markdown"]["type"] == "string"
+assert workspace["draft"]["output_schema"]["properties"]["markdown"]["type"] == "string"
 ```
 
 - [ ] **Step 2: Run tests RED**
@@ -37,36 +51,35 @@ wf draft set-workflow-output <workspace> --revision <n> --map state.markdown=mar
 Run:
 
 ```powershell
-uv run pytest tests/wf_api/test_drafts_service.py::test_validate_draft_workspace_hints_workflow_output_projection -q
+uv run pytest tests/wf_api/test_drafts_service.py::test_bind_draft_local_output_to_workflow_output_lowers_through_state -q
 ```
 
-Expected: fail because no repair hint is present.
+Expected: fail because current code writes an illegal `output.*` node destination.
 
 - [ ] **Step 3: Implement hint branch**
 
-In `_draft_repair_hint`, add handling for:
+In `WorkflowDraftAuthoringApi.bind_draft`, split the existing local-output branch:
 
 ```python
-if diagnostic.get("code") == "invalid_workflow_output_field":
+if source_root == "local" and target_root == "output":
-    path = diagnostic.get("path")
+    output_parts = target_parts
-    # For output[N].target diagnostics, tell the user how to edit top-level output.
+    state_path = f"state.{'.'.join(output_parts)}"
-    return (
+    # Project the capability output field into state_schema and output_schema.
-        f"wf draft set-workflow-output {workspace_id} --revision {revision} "
+    # Merge the step local->state binding and top-level state->output binding
-        "--map <input.or.state.path>=<output_field>"
+    # in one revision-checked patch.
-    )
 ```
 
-Keep this generic if details do not include enough fields. Do not invent source paths.
+Use `project_property_to_schema_path` for both schemas so `$defs` are preserved. Do not create a node output binding with an `output.*` target; `OutputBinding.target` is `StatePath`.
 
 - [ ] **Step 4: Run tests GREEN**
 
-Run the test from Step 2. Expected: pass.
+Run the test from Step 2. Expected: pass with `status: valid`.
 
 - [ ] **Step 5: Commit**
 
 ```powershell
-git add src/wf_api/drafts.py tests/wf_api/test_drafts_service.py tests/wf_cli/test_remote_target.py
+git add src/wf_api/draft_authoring.py tests/wf_api/test_drafts_service.py tests/wf_cli/test_remote_target.py
-git commit -m "fix: hint workflow output draft repairs"
+git commit -m "fix: lower workflow output binds through state"
 ```
 
 ### Task 2: Repair Hint For Undeclared Workflow Input Used By Step Input
@@ -148,7 +161,7 @@ wf draft set-workflow-output report_ws --revision 6 --map state.markdown=markdow
 Add:
 
 ```md
-When validation gives a `repair_hint`, run that exact focused command before JSON Patch. For input/output schema errors, prefer `wf draft bind`.
+When validation gives a `repair_hint`, run that exact focused command before JSON Patch. Use `wf draft bind local.x -> output.y` when one capability output should become public workflow output; it creates the required state intermediary and schemas atomically.
 ```
 
 - [ ] **Step 3: Verify**

docs/wf_cli.md

@@ -308,9 +308,10 @@ wf draft set-name concat_ws --revision 1 --name concat_ws_v2
 wf draft set-route concat_ws --revision 2 --step call --outcome ok --to __end__
 wf draft set-input concat_ws --revision 3 --step call --map input.items=items --map input.separator=separator
 wf draft set-output concat_ws --revision 4 --step call --map value=state.value
-wf draft set-input concat_ws --revision 5 --step call --merge --map input.limit=limit
+wf draft set-workflow-output concat_ws --revision 5 --map state.value=result
-wf draft branch concat_ws --revision 6 --step call --route ok=__end__ --route error=tool_error
+wf draft set-input concat_ws --revision 6 --step call --merge --map input.limit=limit
-wf draft handle concat_ws --revision 7 --to fail --branch lookup:error --branch transform:error
+wf draft branch concat_ws --revision 7 --step call --route ok=__end__ --route error=tool_error
+wf draft handle concat_ws --revision 8 --to fail --branch lookup:error --branch transform:error
 wf draft compile concat_ws
 ```
 
@@ -320,10 +321,17 @@ wf draft compile concat_ws
 `set-output` maps node-local output fields to workflow state paths:
 `text=state.text` means `local.text -> state.text`.
 
-By default, `set-input` and `set-output` replace the whole map for that step.
+`set-workflow-output` maps graph source paths (`input.*`, `state.*`, or
-Use repeated `--map` flags in one command when you know the complete map. Use
+`context.*`) to top-level output fields: `state.value=result` means
-`--merge` when adding or updating one entry across a later revision while
+`state.value -> output.result`.
-preserving existing bindings.
+
+The output field must already be declared in `output_schema`. The command edits
+the projection map; it does not infer or add output schema fields.
+
+By default, `set-input`, `set-output`, and `set-workflow-output` replace the
+whole map for that step or output scope. Use repeated `--map` flags in one
+command when you know the complete map. Use `--merge` when adding or updating
+one entry across a later revision while preserving existing bindings.
 
 ### Bind A Step Path
 
@@ -334,8 +342,8 @@ use `input.*` or `state.*` to `local.*` for step inputs, and `local.*` to
 `state.*` or `output.*` for step outputs.
 
 ```bash
-wf draft bind concat_ws --revision 6 --step call --from local.value --to state.value
+wf draft bind concat_ws --revision 9 --step call --from local.value --to state.value
-wf draft bind concat_ws --revision 7 --step call --from input.text --to local.text
+wf draft bind concat_ws --revision 9 --step call --from input.text --to local.text
 wf draft validate concat_ws
 ```
 

docs/workflow_drafts.md

@@ -668,6 +668,8 @@ wf draft set-name <workspace_id> --revision <n> --name <name>
 wf draft set-route <workspace_id> --revision <n> --step <step_id> --outcome ok --to <target_step_or___end__>
 wf draft set-input <workspace_id> --revision <n> --step <step_id> --map input.text=text
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=state.text
+wf draft set-workflow-output <workspace_id> --revision <n> --map state.value=result
+wf draft set-workflow-output <workspace_id> --revision <n> --merge --map state.other=other
 wf draft branch <workspace_id> --revision <n> --step <step_id> --route ok=__end__ --route error=tool_error
 wf draft handle <workspace_id> --revision <n> --to fail --branch lookup:error --branch transform:error
 wf draft compile <workspace_id>

skills/wf-cli/SKILL.md

@@ -43,6 +43,8 @@ wf draft set-input <workspace_id> --revision <n> --step <step_id> --map input.te
 wf draft set-input <workspace_id> --revision <n> --step <step_id> --merge --map input.other=other
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=state.text
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map other=state.other
+wf draft set-workflow-output <workspace_id> --revision <n> --map state.value=result
+wf draft set-workflow-output <workspace_id> --revision <n> --merge --map state.other=other
 wf draft branch <workspace_id> --revision <n> --step <step_id> --route ok=__end__ --route error=fail
 wf draft handle <workspace_id> --revision <n> --to fail --branch lookup:error --branch transform:error
 wf draft compile <workspace_id>

skills/wf-workflow/references/draft-workspaces.md

@@ -78,6 +78,7 @@ Prefer focused helpers over JSON Patch for common edits:
 - `set_draft_route`
 - `set_step_input_map`
 - `set_step_output_map`
+- `set_workflow_output_map`
 - `bind_draft`
 - `add_step_from_capability`
 - `branch_draft`
@@ -93,6 +94,8 @@ wf draft set-input <workspace_id> --revision <n> --step <step_id> --map input.te
 wf draft set-input <workspace_id> --revision <n> --step <step_id> --merge --map input.other=other
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=state.text
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map other=state.other
+wf draft set-workflow-output <workspace_id> --revision <n> --map state.value=result
+wf draft set-workflow-output <workspace_id> --revision <n> --merge --map state.other=other
 wf draft branch <workspace_id> --revision <n> --step <step_id> --route ok=__end__ --route error=fail
 wf draft handle <workspace_id> --revision <n> --to fail --branch lookup:error --branch transform:error
 wf draft compile <workspace_id>
@@ -101,15 +104,21 @@ wf draft bind <workspace_id> --revision <n> --step <step_id> --from input.<field
 wf draft add-step <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --route ok=__end__ --route error=fail --input input.text=text --bind-output result=state.result
 ```
 
+`set-workflow-output` maps a graph source path (`input.*`, `state.*`, or
+`context.*`) to one public workflow output field. It edits top-level
+`WorkflowDraft.output`; `set-output` edits one step's local-to-state bindings.
+The public output field must already exist in `output_schema`.
+
 `set-input` direction: `input.text=text` means graph source `input.text` maps to
 node-local target `local.text`.
 
 `set-output` direction: `text=state.text` means node-local source `local.text`
 maps to graph target `state.text`.
 
-Without `--merge`, `set-input` and `set-output` replace the whole map for that
+Without `--merge`, `set-input`, `set-output`, and `set-workflow-output` replace
-step. Use repeated `--map` flags in one command for a complete replacement. Use
+the whole map for that step or output scope. Use repeated `--map` flags in one
-`--merge` only when adding/updating entries over multiple revisions.
+command for a complete replacement. Use `--merge` only when adding/updating
+entries over multiple revisions.
 
 - `bind_draft`
 

src/wf_api/drafts.py

@@ -299,6 +299,58 @@ class WorkflowDraftApi:
             ],
         )
 
+    async def set_workflow_output_map(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        output_map: dict[str, str],
+        merge: bool = False,
+    ) -> dict[str, Any]:
+        output_bindings: list[dict[str, Any]]
+        if merge:
+            workspace = self._draft_store().get_workspace(workspace_id)
+            remaining = dict(output_map)
+            output_bindings = []
+            output_payload = workspace.draft.get("output")
+            if isinstance(output_payload, list):
+                for binding in output_payload:
+                    if not isinstance(binding, dict):
+                        continue
+                    source = binding.get("path")
+                    target = binding.get("target")
+                    if isinstance(source, str) and isinstance(target, str):
+                        output_bindings.append(
+                            {
+                                "path": source,
+                                "target": remaining.pop(source, target),
+                            }
+                        )
+                    elif isinstance(target, str) and "value" in binding:
+                        # Literal workflow outputs cannot be represented by the
+                        # path-only CLI map, but --merge must not discard them.
+                        output_bindings.append(dict(binding))
+            output_bindings.extend(
+                {"path": source, "target": target}
+                for source, target in remaining.items()
+            )
+        else:
+            output_bindings = [
+                {"path": source, "target": target}
+                for source, target in output_map.items()
+            ]
+        return await self.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[
+                {
+                    "op": "replace",
+                    "path": "/output",
+                    "value": output_bindings,
+                }
+            ],
+        )
+
     def _step_input_maps(
         self,
         *,

src/wf_api/service.py

@@ -371,6 +371,21 @@ class WorkflowApi:
             merge=merge,
         )
 
+    async def set_workflow_output_map(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        output_map: dict[str, str],
+        merge: bool = False,
+    ) -> dict[str, Any]:
+        return await self.drafts.set_workflow_output_map(
+            workspace_id=workspace_id,
+            revision=revision,
+            output_map=output_map,
+            merge=merge,
+        )
+
     async def bind_draft(
         self,
         *,

src/wf_api/surface.py

@@ -114,6 +114,15 @@ class WorkflowDraftSurface(Protocol):
         merge: bool = False,
     ) -> dict[str, Any]: ...
 
+    async def set_workflow_output_map(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        output_map: dict[str, str],
+        merge: bool = False,
+    ) -> dict[str, Any]: ...
+
     async def bind_draft(
         self,
         *,

src/wf_cli/commands/drafts.py

@@ -330,6 +330,60 @@ def set_step_output_map(
     )
 
 
+@app.command("set-workflow-output")
+def set_workflow_output(
+    ctx: typer.Context,
+    workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
+    revision: Annotated[
+        int, typer.Option("--revision", min=1, help="Expected workspace revision.")
+    ],
+    mapping: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--map",
+            help=(
+                "One output binding GRAPH_SOURCE=OUTPUT_FIELD, for example "
+                "state.markdown=markdown. Repeat in one command."
+            ),
+        ),
+    ] = None,
+    merge: Annotated[
+        bool,
+        typer.Option(
+            "--merge",
+            help="Preserve existing workflow output bindings and add/update the passed --map entries.",
+        ),
+    ] = False,
+) -> None:
+    """Set the top-level workflow output projection without writing JSON Patch manually.
+
+    Default behavior replaces the full workflow output map. Pass all desired
+    --map entries in one command for a complete replacement. Use --merge only
+    when adding or updating entries across a later revision.
+
+    This edits WorkflowDraft.output (top-level workflow output). Use
+    wf draft set-output for step-level output bindings.
+
+    Repeat --map for multiple mappings:
+    --map state.markdown=markdown --map state.title=title
+
+    Run `wf draft validate <workspace_id>` after editing the projection.
+    """
+    output_map = _parse_map_flags(mapping)
+    context = load_cli_context(ctx)
+    emit_json(
+        run_cli_operation(
+            context,
+            context.handlers.set_workflow_output_map(
+                workspace_id=workspace_id,
+                revision=revision,
+                output_map=output_map,
+                merge=merge,
+            ),
+        )
+    )
+
+
 @app.command("bind")
 def bind_draft(
     ctx: typer.Context,

src/wf_mcp/proxy/runtime.py

@@ -50,6 +50,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     "wf.workflow.set_draft_route",
     "wf.workflow.set_step_input_map",
     "wf.workflow.set_step_output_map",
+    "wf.workflow.set_workflow_output_map",
     "wf.workflow.create_minimal_draft_workspace",
     "wf.workflow.create_artifact_from_workspace",
     "wf.workflow.create_wrapper_from_workspace",

src/wf_mcp/workflow_surface/models.py

@@ -243,6 +243,21 @@ class SetStepOutputMapRequest(BaseModel):
     )
 
 
+class SetWorkflowOutputMapRequest(BaseModel):
+    """Typed MCP request for replacing or merging workflow output projection."""
+
+    workspace_id: WorkspaceId
+    revision: int = Field(ge=1, description="Expected current workspace revision.")
+    output_map: DraftPathMap
+    merge: bool = Field(
+        default=False,
+        description=(
+            "When false, replace the full workflow output map. When true, "
+            "preserve existing bindings and add/update only output_map entries."
+        ),
+    )
+
+
 class BindDraftRequest(BaseModel):
     """Typed MCP request for binding one draft step path with schema projection."""
 

src/wf_mcp/workflow_surface/tools.py

@@ -37,6 +37,7 @@ from .models import (
     SetDraftRouteRequest,
     SetStepInputMapRequest,
     SetStepOutputMapRequest,
+    SetWorkflowOutputMapRequest,
     TraceRange,
     ValidateDeploymentResult,
     ValidateDraftWorkspaceRequest,
@@ -463,6 +464,27 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             )
         )
 
+    @server.tool(
+        name="wf.workflow.set_workflow_output_map",
+        title="Set Workflow Output Map",
+        description=(
+            "Replace or merge the top-level workflow output projection. "
+            "Map graph source paths such as state.markdown to public output "
+            "fields such as markdown."
+        ),
+    )
+    async def set_workflow_output_map(
+        request: SetWorkflowOutputMapRequest,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.set_workflow_output_map(
+                workspace_id=request.workspace_id,
+                revision=request.revision,
+                output_map=request.output_map,
+                merge=request.merge,
+            )
+        )
+
     @server.tool(
         name="wf.workflow.bind",
         title="Bind Draft",

src/wf_transport_rpc_http/__init__.py

@@ -42,6 +42,7 @@ from .models import (
     SetDraftRouteParams,
     SetStepInputMapParams,
     SetStepOutputMapParams,
+    SetWorkflowOutputMapParams,
     StartRunParams,
     TraceRangeParams,
     ValidateDeploymentParams,
@@ -88,6 +89,7 @@ __all__ = [
     "SetDraftRouteParams",
     "SetStepInputMapParams",
     "SetStepOutputMapParams",
+    "SetWorkflowOutputMapParams",
     "StartRunParams",
     "TraceRangeParams",
     "ValidateDeploymentParams",

src/wf_transport_rpc_http/client/drafts.py

@@ -141,6 +141,24 @@ class RpcDraftClientMixin:
             },
         )
 
+    async def set_workflow_output_map(
+        self: RpcCaller,
+        *,
+        workspace_id: str,
+        revision: int,
+        output_map: dict[str, str],
+        merge: bool = False,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.draft_workspaces.set_workflow_output_map",
+            {
+                "workspace_id": workspace_id,
+                "revision": revision,
+                "output_map": output_map,
+                "merge": merge,
+            },
+        )
+
     async def bind_draft(
         self: RpcCaller,
         *,

src/wf_transport_rpc_http/methods/drafts.py

@@ -28,6 +28,7 @@ from ..models import (
     SetDraftRouteParams,
     SetStepInputMapParams,
     SetStepOutputMapParams,
+    SetWorkflowOutputMapParams,
     ValidateDraftParams,
     ValidateDraftWorkspaceParams,
 )
@@ -185,6 +186,23 @@ def register_methods(
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
+    @entrypoint.method(
+        name="workflow.draft_workspaces.set_workflow_output_map",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_draft_workspaces_set_workflow_output_map(
+        params: SetWorkflowOutputMapParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.set_workflow_output_map(
+                workspace_id=params.workspace_id,
+                revision=params.revision,
+                output_map=params.output_map,
+                merge=params.merge,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
     @entrypoint.method(
         name="workflow.draft_workspaces.bind",
         errors=[WorkflowRpcError],

src/wf_transport_rpc_http/models.py

@@ -141,6 +141,13 @@ class SetStepOutputMapParams(RpcParamsModel):
     merge: bool = False
 
 
+class SetWorkflowOutputMapParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    revision: int = Field(ge=1)
+    output_map: dict[str, str]
+    merge: bool = False
+
+
 class BindDraftParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
     revision: int = Field(ge=1)

tests/wf_api/test_drafts_service.py

@@ -1154,6 +1154,69 @@ async def test_compile_draft_workspace_invalid_returns_diagnostics(
     assert result["diagnostics"]
 
 
+@pytest.mark.asyncio
+async def test_set_workflow_output_map_replaces_top_level_output(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_out_replace")
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(workspace_id="report", draft=_echo_draft())
+
+    result = await api.set_workflow_output_map(
+        workspace_id="report",
+        revision=1,
+        output_map={"state.echoed": "echoed"},
+    )
+
+    assert result["revision"] == 2
+    assert result["status"] == "valid", result["diagnostics"]
+    fetched = await api.get_draft_workspace(workspace_id="report", include_draft=True)
+    assert fetched["draft"]["output"] == [{"path": "state.echoed", "target": "echoed"}]
+
+
+@pytest.mark.asyncio
+async def test_set_workflow_output_map_merges_top_level_output(tmp_path: Path) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_out_merge")
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
+    draft = {
+        **_echo_draft(),
+        "state_schema": {
+            "fields": {
+                "echoed": {"type": "string"},
+                "other": {"type": "string"},
+            }
+        },
+        "output_schema": {
+            "type": "object",
+            "properties": {
+                "echoed": {"type": "string"},
+                "kind": {"type": "string"},
+                "other": {"type": "string"},
+            },
+        },
+        "output": [
+            {"path": "state.echoed", "target": "echoed"},
+            {"value": "report", "target": "kind"},
+        ],
+    }
+    await api.create_draft_workspace(workspace_id="report", draft=draft)
+
+    result = await api.set_workflow_output_map(
+        workspace_id="report",
+        revision=1,
+        output_map={"state.other": "other"},
+        merge=True,
+    )
+
+    assert result["status"] == "valid", result["diagnostics"]
+    fetched = await api.get_draft_workspace(workspace_id="report", include_draft=True)
+    assert fetched["draft"]["output"] == [
+        {"path": "state.echoed", "target": "echoed"},
+        {"value": "report", "target": "kind"},
+        {"path": "state.other", "target": "other"},
+    ]
+
+
 # -- Browser-click test helpers for forward-route tests --
 
 

tests/wf_cli/test_app.py

@@ -140,17 +140,26 @@ def test_wf_draft_help_does_not_list_old_create_from_capability() -> None:
 def test_wf_draft_map_help_explains_replace_merge_and_validate() -> None:
     input_result = runner.invoke(app, ["draft", "set-input", "--help"])
     output_result = runner.invoke(app, ["draft", "set-output", "--help"])
+    workflow_output_result = runner.invoke(
+        app, ["draft", "set-workflow-output", "--help"]
+    )
 
     assert input_result.exit_code == 0
     assert output_result.exit_code == 0
+    assert workflow_output_result.exit_code == 0
     input_help = " ".join(input_result.output.split())
     output_help = " ".join(output_result.output.split())
+    workflow_output_help = " ".join(workflow_output_result.output.split())
     assert "replaces the full input map" in input_help
     assert "Use --merge only" in input_help
     assert "draft validate" in input_help
     assert "replaces the full output map" in output_help
     assert "Use --merge only" in output_help
     assert "draft validate" in output_help
+    assert "replaces the full workflow output map" in workflow_output_help
+    assert "Use --merge only" in workflow_output_help
+    assert "GRAPH_SOURCE=OUTPUT_FIELD" in workflow_output_help
+    assert "draft validate" in workflow_output_help
 
 
 def test_wf_draft_bind_help_explains_direction() -> None:

tests/wf_cli/test_remote_target.py

@@ -11,6 +11,7 @@ from typer.testing import CliRunner
 import wf_cli.context as cli_context
 from wf_api.models import RawWorkflowPlan
 from wf_cli.app import app
+from wf_cli.commands import drafts as drafts_module
 from wf_cli.context import CliContext, load_cli_context, load_local_cli_context
 from wf_core import END
 from wf_server import build_local_static_workflow_server
@@ -1143,6 +1144,51 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
     ]
 
 
+def test_wf_draft_set_workflow_output_uses_rpc_target(monkeypatch, tmp_path) -> None:
+    calls: list[dict[str, object]] = []
+
+    class FakeDrafts:
+        async def set_workflow_output_map(self, **kwargs: object) -> dict[str, object]:
+            calls.append(kwargs)
+            return {"workspace_id": "report", "revision": 2}
+
+    def _fake_context(ctx: object) -> CliContext:
+        return CliContext(
+            config_path=Path("dummy"),
+            service=None,
+            handlers=FakeDrafts(),  # type: ignore[arg-type]
+            source_admin=cast(Any, object()),
+            admin=cast(Any, object()),
+        )
+
+    monkeypatch.setattr(drafts_module, "load_cli_context", _fake_context)
+    runner = CliRunner()
+    result = runner.invoke(
+        app,
+        [
+            "--url",
+            "http://example.test/rpc",
+            "draft",
+            "set-workflow-output",
+            "report",
+            "--revision",
+            "1",
+            "--map",
+            "state.markdown=markdown",
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert calls == [
+        {
+            "workspace_id": "report",
+            "revision": 1,
+            "output_map": {"state.markdown": "markdown"},
+            "merge": False,
+        }
+    ]
+
+
 def test_wf_draft_remove_route_uses_rpc_target(monkeypatch, tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     _patch_rpc_client_to_server(monkeypatch, server)

tests/wf_mcp/server/test_config.py

@@ -56,6 +56,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.set_draft_route" in names
             assert "wf.workflow.set_step_input_map" in names
             assert "wf.workflow.set_step_output_map" in names
+            assert "wf.workflow.set_workflow_output_map" in names
             assert "wf.workflow.bind" in names
             assert "wf.workflow.add_step_from_capability" in names
             assert "wf.workflow.remove_draft_route" in names
@@ -117,6 +118,14 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             ].inputSchema
             set_input_request = set_input_schema["properties"]["request"]
             assert "merge" in set_input_request["properties"]
+            set_workflow_output_schema = tools_by_name[
+                "wf.workflow.set_workflow_output_map"
+            ].inputSchema
+            set_workflow_output_request = set_workflow_output_schema["properties"][
+                "request"
+            ]
+            assert "output_map" in set_workflow_output_request["properties"]
+            assert "merge" in set_workflow_output_request["properties"]
             add_step_schema = tools_by_name[
                 "wf.workflow.add_step_from_capability"
             ].inputSchema

tests/wf_mcp/server/test_tools.py

@@ -40,6 +40,7 @@ def test_server_search_mode_pins_stable_control_and_workflow_tools() -> None:
             assert "wf.workflow.set_draft_route" in names
             assert "wf.workflow.set_step_input_map" in names
             assert "wf.workflow.set_step_output_map" in names
+            assert "wf.workflow.set_workflow_output_map" in names
             assert "wf.workflow.create_minimal_draft_workspace" in names
             assert "wf.workflow.create_artifact_from_workspace" in names
             assert "wf.workflow.create_wrapper_from_workspace" in names

tests/wf_transport_rpc_http/test_app.py

@@ -775,6 +775,42 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
     )
 
 
+async def test_rpc_draft_workspace_set_workflow_output_map(tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        created = await _rpc(
+            client,
+            "workflow.draft_workspaces.create_from_capability",
+            {
+                "workspace_id": "output_ws",
+                "capability_name": "wf.std.constant",
+                "name": "output_test",
+            },
+        )
+
+        result = await _rpc(
+            client,
+            "workflow.draft_workspaces.set_workflow_output_map",
+            {
+                "workspace_id": "output_ws",
+                "revision": created["result"]["revision"],
+                "output_map": {"state.value": "value"},
+            },
+        )
+        fetched = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "output_ws", "include_draft": True},
+        )
+
+    assert result["result"]["revision"] == 2
+    assert fetched["result"]["draft"]["output"] == [
+        {"path": "state.value", "target": "value"},
+    ]
+
+
 async def test_rpc_draft_workspace_remove_route(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     app = create_rpc_app(server)

tests/wf_transport_rpc_http/test_client.py

@@ -448,6 +448,39 @@ async def test_rpc_client_creates_artifact_from_plan(tmp_path) -> None:
     assert inspected["id"] == "client_plan"
 
 
+async def test_rpc_client_set_workflow_output_map(tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="http://test"
+    ) as http_client:
+        client = RpcWorkflowApiClient(
+            url="http://test/rpc",
+            timeout_seconds=5,
+            http_client=http_client,
+        )
+        await client.create_draft_workspace_from_capability(
+            workspace_id="client_output_ws",
+            capability_name="wf.std.constant",
+            name="client_output",
+        )
+        result = await client.set_workflow_output_map(
+            workspace_id="client_output_ws",
+            revision=1,
+            output_map={"state.value": "value"},
+        )
+        fetched = await client.get_draft_workspace(
+            workspace_id="client_output_ws",
+            include_draft=True,
+        )
+
+    assert result["revision"] == 2
+    assert fetched["draft"]["output"] == [
+        {"path": "state.value", "target": "value"},
+    ]
+
+
 async def test_rpc_client_draft_workspace_focused_edit_methods(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     app = create_rpc_app(server)