feat: replace draft output bind with general bind

docs/current_roadmap.md

@@ -63,9 +63,10 @@ clear operator feedback before adding more architecture.
 - Completed: `wf schema` now lists workflow document/component models, emits
   compact JSON outlines for agent discovery, and emits valid self-contained
   JSON Schema with `--verbose`.
-- Completed: `wf draft bind-output-to-state` composes state schema projection
+- Completed: `wf draft bind --from ... --to ...` composes input/state/output
-  with output binding merge, reducing manual draft patch repairs in agent
+  schema projection with step binding merge, replacing the narrower
-  challenge runs.
+  `bind-output-to-state` helper and reducing manual draft patch repairs in
+  agent challenge runs.
 - Completed: `wf draft add-step-from-capability` inserts one explicit
   capability-backed step with route, input, and output-to-state schema/binding
   wiring in a single revision, reducing brittle JSON Patch authoring for
@@ -75,7 +76,7 @@ clear operator feedback before adding more architecture.
 - Completed: `wf draft compile` returns the compiled raw plan plus required
   capabilities without mutating or saving the draft workspace.
 - Completed: draft validation now preserves structured core validation issues
-  and adds exact `bind-output-to-state` repair hints for missing state fields.
+  and adds exact `wf draft bind` repair hints for missing state fields.
 - Keep status read-only; do not mutate registry, auth, config, or stores.
 
 ## Priority 2: Durable Run/Resume Hardening

docs/historical/superpowers/plans/2026-06-27-draft-bind-from-to.md

@@ -0,0 +1,841 @@
+# Draft Bind From/To Implementation Plan
+
+> **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:** Add a general `wf draft bind --from <path> --to <path>` operation that handles capability-aware input/output binding plus schema projection.
+
+**Architecture:** Extend the existing semantic draft authoring service rather than adding another parallel helper. Generalize schema projection in `wf_api.schema_projection`, then expose one `bind_draft` method through API, service, RPC, MCP, and CLI. Remove the recently-added narrow `bind_output_to_state` surface instead of preserving ghost compatibility.
+
+**Tech Stack:** Python 3.14, Pydantic, Typer, JSON-RPC, MCP tool models, `jsonschema.Draft202012Validator`, pytest, basedpyright.
+
+---
+
+## File Map
+
+- Modify `src/wf_api/schema_projection.py`: general schema projection helper with nested target insertion.
+- Modify `src/wf_api/draft_authoring.py`: add `bind_draft`, remove `bind_output_to_state`.
+- Modify `src/wf_api/service.py` and `src/wf_api/surface.py`: facade/protocol methods.
+- Modify `src/wf_transport_rpc_http/models.py`, `methods/drafts.py`, `client/drafts.py`, and `__init__.py`: RPC DTO/method/client/export.
+- Modify `src/wf_mcp/workflow_surface/models.py` and `tools.py`: MCP request/tool.
+- Modify `src/wf_cli/commands/drafts.py`: add `wf draft bind` and remove `bind-output-to-state`.
+- Modify `src/wf_api/drafts.py`: update repair hints from `bind-output-to-state` to `bind`.
+- Modify docs and skills: `docs/wf_cli.md`, `docs/current_roadmap.md`, `docs/superpowers/specs/2026-06-27-draft-semantic-authoring-boundary.md`, `skills/wf-cli/SKILL.md`, `skills/wf-workflow/references/draft-workspaces.md`, `skills/wf-workflow/references/workflow-lifecycle.md`.
+
+---
+
+### Task 1: Generalize Schema Projection
+
+**Files:**
+- Modify: `src/wf_api/schema_projection.py`
+- Test: `tests/wf_api/test_schema_projection.py`
+
+- [ ] **Step 1: Add failing tests for nested insertion and overwrite rejection**
+
+Append tests:
+
+```python
+def test_project_schema_property_inserts_nested_path_and_defs() -> None:
+    projected = project_property_to_schema_path(
+        target_schema={"type": "object", "properties": {}},
+        source_schema={
+            "type": "object",
+            "properties": {"after": {"$ref": "#/$defs/Snapshot"}},
+            "$defs": {
+                "Snapshot": {
+                    "type": "object",
+                    "properties": {"clicked": {"type": "boolean"}},
+                }
+            },
+        },
+        source_field="after",
+        target_parts=("session", "after"),
+    )
+
+    assert projected["properties"]["session"]["type"] == "object"
+    assert projected["properties"]["session"]["properties"]["after"] == {
+        "$ref": "#/$defs/Snapshot"
+    }
+    assert projected["$defs"]["Snapshot"]["properties"]["clicked"] == {
+        "type": "boolean"
+    }
+
+
+def test_project_schema_property_rejects_existing_nested_target() -> None:
+    with pytest.raises(ValueError, match="schema path 'session.after' already exists"):
+        project_property_to_schema_path(
+            target_schema={
+                "type": "object",
+                "properties": {
+                    "session": {
+                        "type": "object",
+                        "properties": {"after": {"type": "string"}},
+                    }
+                },
+            },
+            source_schema={
+                "type": "object",
+                "properties": {"after": {"type": "object"}},
+            },
+            source_field="after",
+            target_parts=("session", "after"),
+        )
+
+
+def test_project_schema_property_rejects_non_object_ancestor() -> None:
+    with pytest.raises(ValueError, match="schema path 'session' is not an object"):
+        project_property_to_schema_path(
+            target_schema={
+                "type": "object",
+                "properties": {"session": {"type": "string"}},
+            },
+            source_schema={
+                "type": "object",
+                "properties": {"after": {"type": "object"}},
+            },
+            source_field="after",
+            target_parts=("session", "after"),
+        )
+```
+
+- [ ] **Step 2: Run tests and confirm failure**
+
+Run:
+
+```powershell
+uv run pytest tests/wf_api/test_schema_projection.py -q
+```
+
+Expected: failures because `project_property_to_schema_path` does not exist.
+
+- [ ] **Step 3: Implement general helper and keep the root-state convenience function**
+
+In `src/wf_api/schema_projection.py`, add:
+
+```python
+def project_property_to_schema_path(
+    *,
+    target_schema: JsonObject,
+    source_schema: JsonObject,
+    source_field: str,
+    target_parts: tuple[str, ...],
+) -> JsonObject:
+    """Copy one source property schema into a target JSON Schema object path."""
+    if not target_parts:
+        raise ValueError("target schema path must not be empty")
+    _check_schema("target_schema", target_schema)
+    _check_schema("source_schema", source_schema)
+    source_properties = source_schema.get("properties")
+    if not isinstance(source_properties, dict) or source_field not in source_properties:
+        raise ValueError(f"source field {source_field!r} is not declared")
+    source_property = source_properties[source_field]
+    if not isinstance(source_property, dict):
+        raise ValueError(f"source field {source_field!r} is not a JSON Schema object")
+
+    projected = deepcopy(target_schema)
+    _ensure_object_schema(projected, "target_schema")
+    parent = projected
+    for index, part in enumerate(target_parts[:-1]):
+        properties = _properties_for_object(parent, ".".join(target_parts[:index]) or "target_schema")
+        child = properties.get(part)
+        if child is None:
+            child = {"type": "object", "properties": {}}
+            properties[part] = child
+        if not isinstance(child, dict):
+            raise ValueError(f"schema path {'.'.join(target_parts[: index + 1])!r} is not an object")
+        _ensure_object_schema(child, ".".join(target_parts[: index + 1]))
+        parent = child
+
+    properties = _properties_for_object(parent, ".".join(target_parts[:-1]) or "target_schema")
+    leaf = target_parts[-1]
+    if leaf in properties:
+        raise ValueError(f"schema path {'.'.join(target_parts)!r} already exists")
+    properties[leaf] = deepcopy(source_property)
+
+    _merge_definition_block(projected, source_schema, "$defs")
+    _merge_definition_block(projected, source_schema, "definitions")
+    _check_schema("projected target_schema", projected)
+    return projected
+
+
+def _ensure_object_schema(schema: JsonObject, label: str) -> None:
+    schema_type = schema.get("type")
+    if schema_type is not None and schema_type != "object":
+        raise ValueError(f"{label} must be an object schema")
+    schema.setdefault("type", "object")
+
+
+def _properties_for_object(schema: JsonObject, label: str) -> JsonObject:
+    properties = schema.setdefault("properties", {})
+    if not isinstance(properties, dict):
+        raise ValueError(f"{label}.properties must be an object")
+    return properties
+```
+
+Then rewrite `project_output_property_to_state_schema` as:
+
+```python
+def project_output_property_to_state_schema(
+    *,
+    state_schema: JsonObject,
+    output_schema: JsonObject,
+    output_field: str,
+    state_field: str,
+) -> JsonObject:
+    """Root state projection convenience wrapper."""
+    return project_property_to_schema_path(
+        target_schema=state_schema,
+        source_schema=output_schema,
+        source_field=output_field,
+        target_parts=(state_field,),
+    )
+```
+
+- [ ] **Step 4: Run schema projection tests**
+
+Run:
+
+```powershell
+uv run pytest tests/wf_api/test_schema_projection.py -q
+```
+
+Expected: all pass.
+
+- [ ] **Step 5: Commit**
+
+```powershell
+git add src/wf_api/schema_projection.py tests/wf_api/test_schema_projection.py
+git commit -m "feat: generalize draft schema projection"
+```
+
+---
+
+### Task 2: Add API-Level `bind_draft`
+
+**Files:**
+- Modify: `src/wf_api/draft_authoring.py`
+- Test: `tests/wf_api/test_drafts_service.py`
+
+- [ ] **Step 1: Add failing API tests**
+
+Add tests:
+
+```python
+@pytest.mark.asyncio
+async def test_bind_draft_workflow_input_to_step_input_projects_input_schema(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_input")
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
+    service.register_specs("demo.personal", _snapshot_tool)
+    await api.create_draft_workspace(
+        workspace_id="bind_ws",
+        draft=_echo_draft(),
+    )
+
+    result = await authoring.bind_draft(
+        workspace_id="bind_ws",
+        revision=1,
+        step_id="echo",
+        source_path="input.message",
+        target_path="local.message",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="bind_ws", include_draft=True)
+
+    assert result["revision"] == 2
+    assert workspace["draft"]["input_schema"]["properties"]["message"]["type"] == "string"
+    assert workspace["draft"]["steps"]["echo"]["input"] == [
+        {"target": "message", "path": "input.message"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_bind_draft_output_to_nested_state_projects_state_schema(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_output_nested")
+    api, service, authoring = _draft_api(artifact_store, register_echo=True)
+    service.register_specs("demo.personal", _snapshot_tool)
+    await api.create_draft_workspace(
+        workspace_id="snapshot_ws",
+        draft={
+            "name": "snapshot",
+            "input_schema": {"type": "object", "properties": {}},
+            "state_schema": {"type": "object", "properties": {}},
+            "output_schema": {"type": "object", "properties": {}},
+            "start": "snap",
+            "steps": {"snap": {"use": "demo.personal.snapshot_tool", "input": [], "output": []}},
+            "routes": {"snap": {"ok": "__end__"}},
+        },
+    )
+
+    result = await authoring.bind_draft(
+        workspace_id="snapshot_ws",
+        revision=1,
+        step_id="snap",
+        source_path="local.after",
+        target_path="state.session.after",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="snapshot_ws", include_draft=True)
+
+    assert result["revision"] == 2
+    assert (
+        workspace["draft"]["state_schema"]["properties"]["session"]["properties"]["after"]["$ref"]
+        == "#/$defs/_Snapshot"
+    )
+    assert workspace["draft"]["steps"]["snap"]["output"] == [
+        {"source": "after", "target": "state.session.after"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_bind_draft_rejects_unsupported_direction(tmp_path: Path) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_bad_direction")
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(workspace_id="bind_ws", draft=_echo_draft())
+
+    with pytest.raises(ValueError, match="unsupported bind direction"):
+        await authoring.bind_draft(
+            workspace_id="bind_ws",
+            revision=1,
+            step_id="echo",
+            source_path="input.message",
+            target_path="state.message",
+        )
+```
+
+- [ ] **Step 2: Run tests and confirm failure**
+
+Run:
+
+```powershell
+uv run pytest tests/wf_api/test_drafts_service.py -q -k "bind_draft"
+```
+
+Expected: failures because `bind_draft` does not exist.
+
+- [ ] **Step 3: Implement `bind_draft`**
+
+In `src/wf_api/draft_authoring.py`, import:
+
+```python
+from wf_core.paths import GraphSourcePath, LocalPath
+from .schema_projection import (
+    project_output_property_to_state_schema,
+    project_property_to_schema_path,
+)
+```
+
+Add helpers:
+
+```python
+def _graph_parts(path: str) -> tuple[str, tuple[str, ...]]:
+    parsed = GraphSourcePath.parse(path)
+    return parsed.root, parsed.parts
+
+
+def _local_field(path: str) -> str:
+    parsed = LocalPath.parse(path)
+    if len(parsed.parts) != 1:
+        raise ValueError("local path must name one capability field")
+    return parsed.parts[0]
+```
+
+Add method:
+
+```python
+async def bind_draft(
+    self,
+    *,
+    workspace_id: str,
+    revision: int,
+    step_id: str,
+    source_path: str,
+    target_path: str,
+) -> dict[str, Any]:
+    """Bind a graph path to/from one capability local field with schema projection."""
+    workspace = self.drafts._draft_store().get_workspace(workspace_id)
+    step = draft_step(workspace.draft, step_id)
+    capability_name = step.get("use")
+    if not isinstance(capability_name, str) or not capability_name:
+        raise ValueError(f"draft step {step_id!r} does not declare a capability use")
+    spec = self.context.specs.get_qualified_spec(capability_name)
+
+    source_root, source_parts = _graph_parts(source_path) if not source_path.startswith("local.") else ("local", LocalPath.parse(source_path).parts)
+    target_root, target_parts = _graph_parts(target_path) if not target_path.startswith("local.") else ("local", LocalPath.parse(target_path).parts)
+
+    if target_root == "local" and source_root in {"input", "state"}:
+        local_field = _local_field(target_path)
+        input_schema = spec.input_schema_contract or spec.input_model.model_json_schema()
+        schema_key = "input_schema" if source_root == "input" else "state_schema"
+        target_schema = workspace.draft.get(schema_key, {})
+        if not isinstance(target_schema, dict):
+            raise ValueError(f"draft {schema_key} must be an object")
+        projected = project_property_to_schema_path(
+            target_schema=target_schema,
+            source_schema=input_schema,
+            source_field=local_field,
+            target_parts=source_parts,
+        )
+        input_map = {
+            **_draft_input_maps(workspace.draft, step_id),
+            source_path: local_field,
+        }
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[
+                {"op": "replace", "path": f"/{schema_key}", "value": projected},
+                {
+                    "op": "replace",
+                    "path": f"/steps/{escape_json_pointer(step_id)}/input",
+                    "value": input_bindings_payload(input_map, {}),
+                },
+            ],
+        )
+
+    if source_root == "local" and target_root in {"state", "output"}:
+        local_field = _local_field(source_path)
+        output_schema = spec.output_schema_contract or spec.output_model.model_json_schema()
+        schema_key = "state_schema" if target_root == "state" else "output_schema"
+        target_schema = workspace.draft.get(schema_key, {})
+        if not isinstance(target_schema, dict):
+            raise ValueError(f"draft {schema_key} must be an object")
+        projected = project_property_to_schema_path(
+            target_schema=target_schema,
+            source_schema=output_schema,
+            source_field=local_field,
+            target_parts=target_parts,
+        )
+        output_map = {
+            **self.drafts._step_output_map(workspace_id=workspace_id, step_id=step_id),
+            local_field: target_path,
+        }
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[
+                {"op": "replace", "path": f"/{schema_key}", "value": projected},
+                {
+                    "op": "replace",
+                    "path": f"/steps/{escape_json_pointer(step_id)}/output",
+                    "value": output_bindings_payload(output_map),
+                },
+            ],
+        )
+
+    raise ValueError(f"unsupported bind direction: {source_path!r} -> {target_path!r}")
+```
+
+Delete the old `bind_output_to_state` method from `WorkflowDraftAuthoringApi`.
+It is superseded by `bind_draft` and should not remain as a compatibility shim.
+
+- [ ] **Step 4: Run API tests**
+
+Run:
+
+```powershell
+uv run pytest tests/wf_api/test_drafts_service.py -q -k "bind_draft or bind_output_to_state"
+```
+
+Expected: new `bind_draft` tests pass. Existing `bind_output_to_state` tests
+should be updated or removed in this task because the old method is gone.
+
+- [ ] **Step 5: Commit**
+
+```powershell
+git add src/wf_api/draft_authoring.py tests/wf_api/test_drafts_service.py
+git commit -m "feat: replace narrow draft output bind with general bind"
+```
+
+---
+
+### Task 3: Expose Bind Through Facade, RPC, MCP, And CLI
+
+**Files:**
+- Modify: `src/wf_api/service.py`, `src/wf_api/surface.py`
+- Modify: `src/wf_transport_rpc_http/models.py`, `src/wf_transport_rpc_http/methods/drafts.py`, `src/wf_transport_rpc_http/client/drafts.py`, `src/wf_transport_rpc_http/__init__.py`
+- Modify: `src/wf_mcp/workflow_surface/models.py`, `src/wf_mcp/workflow_surface/tools.py`
+- Modify: `src/wf_cli/commands/drafts.py`
+- Test: `tests/wf_transport_rpc_http/test_app.py`, `tests/wf_transport_rpc_http/test_client.py`, `tests/wf_cli/test_app.py`, `tests/wf_cli/test_remote_target.py`, `tests/wf_mcp/server/test_config.py`
+
+- [ ] **Step 1: Add failing surface tests**
+
+In RPC app/client tests, replace existing `bind_output_to_state` coverage with:
+
+```python
+{
+    "workspace_id": "rpc_ws",
+    "revision": 1,
+    "step_id": "call",
+    "source_path": "local.echoed",
+    "target_path": "state.echoed",
+}
+```
+
+Expected method name:
+
+```text
+workflow.draft_workspaces.bind
+```
+
+In CLI remote target test, invoke:
+
+```python
+[
+    "draft",
+    "bind",
+    "rpc_ws",
+    "--revision",
+    "1",
+    "--step",
+    "call",
+    "--from",
+    "local.echoed",
+    "--to",
+    "state.echoed",
+]
+```
+
+In MCP config test, assert:
+
+```python
+assert "wf.workflow.bind" in names
+```
+
+- [ ] **Step 2: Run tests and confirm failure**
+
+Run:
+
+```powershell
+uv run pytest tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_cli/test_app.py tests/wf_cli/test_remote_target.py tests/wf_mcp/server/test_config.py -q -k "bind"
+```
+
+Expected: failures because public bind surfaces do not exist.
+
+- [ ] **Step 3: Add facade and protocol**
+
+In `src/wf_api/surface.py`, add:
+
+```python
+async def bind_draft(
+    self,
+    *,
+    workspace_id: str,
+    revision: int,
+    step_id: str,
+    source_path: str,
+    target_path: str,
+) -> dict[str, Any]: ...
+```
+
+Remove the old `bind_output_to_state` protocol method.
+
+In `src/wf_api/service.py`, add delegate:
+
+```python
+async def bind_draft(
+    self,
+    *,
+    workspace_id: str,
+    revision: int,
+    step_id: str,
+    source_path: str,
+    target_path: str,
+) -> dict[str, Any]:
+    return await self.draft_authoring.bind_draft(
+        workspace_id=workspace_id,
+        revision=revision,
+        step_id=step_id,
+        source_path=source_path,
+        target_path=target_path,
+    )
+```
+
+Remove the old `bind_output_to_state` facade method.
+
+- [ ] **Step 4: Add RPC model/method/client**
+
+In `src/wf_transport_rpc_http/models.py`:
+
+```python
+class BindDraftParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    revision: int = Field(ge=1)
+    step_id: str = Field(min_length=1)
+    source_path: str = Field(min_length=1)
+    target_path: str = Field(min_length=1)
+```
+
+In `src/wf_transport_rpc_http/methods/drafts.py`:
+
+```python
+@entrypoint.method(
+    name="workflow.draft_workspaces.bind",
+    errors=[WorkflowRpcError],
+)
+async def workflow_draft_workspaces_bind(
+    params: BindDraftParams = RpcParams(),
+) -> dict[str, Any]:
+    try:
+        return await server.api.bind_draft(
+            workspace_id=params.workspace_id,
+            revision=params.revision,
+            step_id=params.step_id,
+            source_path=params.source_path,
+            target_path=params.target_path,
+        )
+    except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+        raise_workflow_rpc_error(exc)
+```
+
+In `src/wf_transport_rpc_http/client/drafts.py`:
+
+```python
+async def bind_draft(
+    self,
+    *,
+    workspace_id: str,
+    revision: int,
+    step_id: str,
+    source_path: str,
+    target_path: str,
+) -> dict[str, Any]:
+    return await self._request(
+        "workflow.draft_workspaces.bind",
+        {
+            "workspace_id": workspace_id,
+            "revision": revision,
+            "step_id": step_id,
+            "source_path": source_path,
+            "target_path": target_path,
+        },
+    )
+```
+
+Export `BindDraftParams` from `src/wf_transport_rpc_http/__init__.py`.
+
+Remove `BindOutputToStateParams`, the
+`workflow.draft_workspaces.bind_output_to_state` RPC method, and
+`RpcDraftClientMixin.bind_output_to_state`.
+
+- [ ] **Step 5: Add MCP request/tool**
+
+In `src/wf_mcp/workflow_surface/models.py`:
+
+```python
+class BindDraftRequest(BaseModel):
+    """Typed MCP request for binding one draft step path with schema projection."""
+
+    workspace_id: WorkspaceId
+    revision: int = Field(ge=1, description="Expected current workspace revision.")
+    step_id: str = Field(description="Capability-backed draft step id.")
+    source_path: str = Field(description="Source path, for example input.x or local.y.")
+    target_path: str = Field(description="Target path, for example local.x or state.y.")
+```
+
+In `src/wf_mcp/workflow_surface/tools.py`, register `wf.workflow.bind` beside
+the other draft authoring tools.
+
+Remove `BindOutputToStateRequest` and the `wf.workflow.bind_output_to_state`
+tool.
+
+- [ ] **Step 6: Add CLI command**
+
+In `src/wf_cli/commands/drafts.py`, add:
+
+```python
+@app.command("bind")
+def bind_draft(
+    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.")
+    ],
+    step_id: Annotated[str, typer.Option("--step", help="Draft step id.")],
+    source_path: Annotated[
+        str,
+        typer.Option("--from", help="Source path, for example input.x or local.y."),
+    ],
+    target_path: Annotated[
+        str,
+        typer.Option("--to", help="Target path, for example local.x or state.y."),
+    ],
+) -> None:
+    """Bind a capability step path and project the matching schema.
+
+    Direction matters. Use input/state -> local for step inputs and local ->
+    state/output for step outputs. Run `wf draft validate <workspace_id>` after
+    this command.
+    """
+    context = load_cli_context(ctx)
+    emit_json(
+        run_cli_operation(
+            context,
+            context.handlers.bind_draft(
+                workspace_id=workspace_id,
+                revision=revision,
+                step_id=step_id,
+                source_path=source_path,
+                target_path=target_path,
+            ),
+        )
+    )
+```
+
+Then remove the old `@app.command("bind-output-to-state")` command.
+
+- [ ] **Step 7: Run surface tests**
+
+Run:
+
+```powershell
+uv run pytest tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_cli/test_app.py tests/wf_cli/test_remote_target.py tests/wf_mcp/server/test_config.py -q -k "bind"
+```
+
+Expected: all selected tests pass.
+
+- [ ] **Step 8: Commit**
+
+```powershell
+git add src/wf_api/service.py src/wf_api/surface.py src/wf_transport_rpc_http src/wf_mcp/workflow_surface src/wf_cli/commands/drafts.py tests/wf_transport_rpc_http tests/wf_cli tests/wf_mcp
+git commit -m "feat: expose general draft bind across transports"
+```
+
+---
+
+### Task 4: Update Repair Hints, Docs, And Skills
+
+**Files:**
+- Modify: `src/wf_api/drafts.py`
+- Modify: `docs/wf_cli.md`, `docs/current_roadmap.md`, `docs/superpowers/specs/2026-06-27-draft-semantic-authoring-boundary.md`
+- Modify: `skills/wf-cli/SKILL.md`, `skills/wf-workflow/references/draft-workspaces.md`, `skills/wf-workflow/references/workflow-lifecycle.md`
+- Test: `tests/wf_api/test_drafts_service.py`, `tests/wf_cli/test_remote_target.py`
+
+- [ ] **Step 1: Update repair hint tests**
+
+Change expectations from:
+
+```python
+"wf draft bind-output-to-state snapshot_ws --revision 1 --step snap --output after --state state.after"
+```
+
+to:
+
+```python
+"wf draft bind snapshot_ws --revision 1 --step snap --from local.after --to state.after"
+```
+
+- [ ] **Step 2: Update repair hint implementation**
+
+In `src/wf_api/drafts.py`, update `_draft_repair_hint` return value:
+
+```python
+return (
+    f"wf draft bind {workspace_id} --revision {revision} "
+    f"--step {step_id} --from local.{output_field} --to {state_path}"
+)
+```
+
+- [ ] **Step 3: Update docs and skills**
+
+Docs should state:
+
+```markdown
+Use `wf draft bind --from ... --to ...` for schema-aware step wiring.
+`bind-output-to-state` has been removed; use `bind --from local.<field>
+--to state.<field>` instead.
+```
+
+Include examples:
+
+```powershell
+wf draft bind browser_ws --revision 2 --step click --from input.simulate --to local.simulate
+wf draft bind browser_ws --revision 3 --step click --from local.after --to state.after
+wf draft bind report_ws --revision 4 --step render --from local.markdown --to output.markdown
+```
+
+- [ ] **Step 4: Run tests**
+
+Run:
+
+```powershell
+uv run pytest tests/wf_api/test_drafts_service.py tests/wf_cli/test_remote_target.py -q -k "repair_hint or bind"
+```
+
+Expected: all selected tests pass.
+
+- [ ] **Step 5: Commit**
+
+```powershell
+git add src/wf_api/drafts.py docs/wf_cli.md docs/current_roadmap.md docs/superpowers/specs/2026-06-27-draft-semantic-authoring-boundary.md skills/wf-cli/SKILL.md skills/wf-workflow/references/draft-workspaces.md skills/wf-workflow/references/workflow-lifecycle.md tests/wf_api/test_drafts_service.py tests/wf_cli/test_remote_target.py
+git commit -m "docs: document general draft bind helper"
+```
+
+---
+
+### Task 5: Final Verification
+
+**Files:**
+- No new files.
+
+- [ ] **Step 1: Run focused test set**
+
+Run:
+
+```powershell
+uv run pytest tests/wf_api/test_schema_projection.py tests/wf_api/test_drafts_service.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_cli/test_app.py tests/wf_cli/test_remote_target.py tests/wf_mcp/server/test_config.py -q -k "bind or schema_projection or repair_hint"
+```
+
+Expected: all selected tests pass.
+
+- [ ] **Step 2: Run lint, format, and type checks**
+
+Run:
+
+```powershell
+uv run ruff check
+uv run ruff format --check
+uv run basedpyright --level error
+git diff --check
+```
+
+Expected:
+
+- Ruff clean.
+- Format clean.
+- Basedpyright reports `0 errors`.
+- `git diff --check` has no whitespace errors. CRLF warnings are acceptable on Windows.
+
+- [ ] **Step 3: Optional live smoke**
+
+If `wf-rpc-server --config wf.config.json --host 127.0.0.1 --port 8765` is running, run:
+
+```powershell
+uv run wf --url http://127.0.0.1:8765/rpc draft bind --help
+```
+
+Then create a temporary draft and verify:
+
+```powershell
+uv run wf --url http://127.0.0.1:8765/rpc draft bind <workspace_id> --revision <n> --step <step_id> --from local.<field> --to state.<field>
+```
+
+Expected: command routes through RPC and returns a revised workspace summary.
+
+- [ ] **Step 4: Commit final cleanup if needed**
+
+```powershell
+git status --short
+git add <only files changed by cleanup>
+git commit -m "fix: polish draft bind implementation"
+```
+
+Skip this commit if the tree is already clean after Task 4.
+
+---
+
+## Self-Review Notes
+
+- Spec coverage: tasks cover schema projection, API, transports, CLI, repair hints, docs, and deletion of the old narrow helper.
+- Scope control: this removes `bind-output-to-state` because it is recent, narrow, and not a durable external contract.
+- Risk: nested schema projection must stay object-property only. Do not support array item projection in this slice.

docs/superpowers/specs/2026-06-27-draft-bind-from-to.md

@@ -0,0 +1,211 @@
+# Draft Bind From/To Design
+
+## Status
+
+Implemented. This design replaces `bind-output-to-state` with the general
+`bind --from ... --to ...` operation. `bind-output-to-state` was introduced
+recently as an incomplete narrow helper and should be removed rather than
+preserved as compatibility.
+
+## Problem
+
+Agents authoring drafts repeatedly hit the same boundary: step input/output
+bindings are not just path edits. They often require workflow-level schema
+declarations too.
+
+Today:
+
+- `wf draft set-input` edits a step input map, but it does not declare
+  `input_schema.properties.<field>`.
+- `wf draft bind-output-to-state` declares one root state field and writes a
+  step output binding, but only covers `local.output -> state.field`. It is too
+  narrow now that input-side schema projection is needed too.
+- Agents interpret this split inconsistently, then fall back to raw JSON Patch.
+
+The product needs one capability-aware operation with the mental model:
+
+```text
+bind <destination> from <source>
+```
+
+## User-Facing Shape
+
+Primary CLI:
+
+```powershell
+wf draft bind <workspace_id> `
+  --revision <n> `
+  --step <step_id> `
+  --from <path> `
+  --to <path>
+```
+
+Examples:
+
+```powershell
+# Workflow input into a capability input.
+wf draft bind browser_ws --revision 2 --step click `
+  --from input.simulate `
+  --to local.simulate
+```
+
+```powershell
+# Capability output into workflow state.
+wf draft bind browser_ws --revision 3 --step click `
+  --from local.after `
+  --to state.after
+```
+
+```powershell
+# Capability output into final workflow output.
+wf draft bind report_ws --revision 7 --step render `
+  --from local.markdown `
+  --to output.markdown
+```
+
+## Semantics
+
+`bind` is capability-aware and step-scoped. The step must be a capability-backed
+draft step with a `use` field.
+
+Supported directions:
+
+| From | To | Draft edit | Schema projection |
+|---|---|---|---|
+| `input.*` | `local.*` | add/merge step input binding | copy capability input schema for `local.*` into workflow `input_schema` at `input.*` |
+| `state.*` | `local.*` | add/merge step input binding | copy capability input schema for `local.*` into workflow `state_schema` at `state.*` |
+| `local.*` | `state.*` | add/merge step output binding | copy capability output schema for `local.*` into workflow `state_schema` at `state.*` |
+| `local.*` | `output.*` | add/merge step output binding | copy capability output schema for `local.*` into workflow `output_schema` at `output.*` |
+
+Unsupported directions should fail clearly:
+
+- `local.* -> local.*`
+- `input.* -> state.*`
+- `state.* -> output.*`
+- `output.* -> local.*`
+- Any path root outside `input`, `state`, `output`, or `local`
+
+## Schema Projection
+
+The schema projection helper should be generalized from the current
+output-to-state helper.
+
+Requirements:
+
+- Use `jsonschema.Draft202012Validator.check_schema` to validate input schemas
+  and projected schemas. Do not hand-roll JSON Schema validation.
+- Copy the selected local field schema from the capability input/output schema.
+- Preserve `$defs` and `definitions`, rejecting conflicting definitions.
+- Insert the copied field schema at the target workflow path.
+- Support nested target paths such as `state.options.timeout_seconds`.
+- Create missing ancestor object schemas only when the ancestor does not exist.
+- Reject inserting through an existing non-object ancestor.
+- Reject overwriting an existing target property by default.
+
+Nested insertion example:
+
+```json
+{
+  "type": "object",
+  "properties": {
+    "options": {
+      "type": "object",
+      "properties": {
+        "timeout_seconds": { "type": "number" }
+      }
+    }
+  }
+}
+```
+
+This is a schema projection operation, not a general JSON Schema editor. It only
+copies one capability local input/output field schema to one workflow graph path.
+
+## API Shape
+
+Add the general method:
+
+```python
+async def bind_draft(
+    self,
+    *,
+    workspace_id: str,
+    revision: int,
+    step_id: str,
+    source_path: str,
+    target_path: str,
+) -> dict[str, Any]:
+    ...
+```
+
+Name notes:
+
+- User-facing CLI says `bind --from ... --to ...`.
+- Python/RPC names should avoid reserved words and use
+  `source_path` / `target_path`.
+- Remove `bind_output_to_state` end-to-end. It has no long-lived compatibility
+  contract and is replaced by `bind_draft`.
+
+## RPC And MCP
+
+New JSON-RPC method:
+
+```text
+workflow.draft_workspaces.bind
+```
+
+Params:
+
+```json
+{
+  "workspace_id": "browser_ws",
+  "revision": 3,
+  "step_id": "click",
+  "source_path": "local.after",
+  "target_path": "state.after"
+}
+```
+
+New MCP tool:
+
+```text
+wf.workflow.bind
+```
+
+Remove existing RPC/MCP `bind_output_to_state` surfaces during this slice.
+
+## Repair Hints
+
+Existing invalid-destination hints currently recommend:
+
+```text
+wf draft bind-output-to-state ...
+```
+
+They should instead recommend:
+
+```text
+wf draft bind <workspace_id> --revision <n> --step <step_id> --from local.<field> --to state.<field>
+```
+
+Future missing-input-schema diagnostics can also point to:
+
+```text
+wf draft bind <workspace_id> --revision <n> --step <step_id> --from input.<field> --to local.<field>
+```
+
+## Non-Goals
+
+- Do not preserve `bind-output-to-state` as a compatibility alias.
+- Do not add arbitrary schema editing commands.
+- Do not support array item schema projection.
+- Do not infer routes.
+- Do not mutate bindings without revision checks.
+
+## Acceptance Criteria
+
+- Agents can fix browser-click input schema issues with `wf draft bind --from
+  input.simulate --to local.simulate` instead of raw JSON Patch.
+- Agents can bind outputs with `wf draft bind --from local.after --to
+  state.after`; the old `bind-output-to-state` command/method/tool is gone.
+- Nested target paths are handled safely or rejected with clear errors.

docs/superpowers/specs/2026-06-27-draft-semantic-authoring-boundary.md

@@ -109,7 +109,7 @@ The public draft surface is documented in descending order of preference.
 
 - `create-from-capability`
 - `add-step-from-capability`
-- `bind-output-to-state`
+- `bind`
 - `branch`
 - `handle`
 
@@ -123,9 +123,10 @@ graph intent. They are the preferred agent authoring surface.
 - `set-input`
 - `set-output`
 
-These remain available for precise repairs. `set-output` does not project the
+These remain available for precise repairs. `set-input` and `set-output` do
-destination state schema; callers should prefer `bind-output-to-state` when
+not project workflow schemas; callers should prefer `bind` when a capability
-writing a capability output into state.
+input/output should also declare the matching workflow input, state, or output
+schema.
 
 ### Escape Hatch
 
@@ -213,11 +214,22 @@ pairs remain unchanged.
 
 `handle` is not a join. It creates ordinary directed edges to one target.
 
-### Bind Output To State
+### Bind
 
-`bind-output-to-state` remains the capability-aware schema propagation
+`bind` is the capability-aware schema propagation operation. It projects the
-operation. It projects the selected output property and required `$defs` into
+selected capability local input/output property and required `$defs` into the
-the root state schema, then merges the output binding in the same revision.
+workflow input, state, or output schema, then merges the matching step input or
+output binding in the same revision.
+
+```powershell
+wf draft bind WORKSPACE --revision 4 --step wait `
+  --from input.simulate `
+  --to local.simulate
+
+wf draft bind WORKSPACE --revision 5 --step wait `
+  --from local.after `
+  --to state.after
+```
 
 The partial `add-state-from-output` operation is removed from API, RPC, MCP,
 CLI, docs, and skills. It was superseded before acquiring a real caller or

docs/wf_cli.md

@@ -313,24 +313,25 @@ 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 Output To State
+### Bind A Step Path
 
-Use `bind-output-to-state` when a step output should become workflow state and
+Use `bind` when a capability step input/output binding also needs workflow
-the state schema should match that capability output field.
+schema projection. The selected step must be capability-backed (`use: ...`)
-The selected step must be capability-backed (`use: ...`) because the command
+because the command derives the schema from that capability. Direction matters:
-derives the output schema from that capability. Use JSON Patch for uncommon
+use `input.*` or `state.*` to `local.*` for step inputs, and `local.*` to
-control-flow or non-capability draft steps.
+`state.*` or `output.*` for step outputs.
 
 ```bash
-wf draft bind-output-to-state concat_ws --revision 6 --step call --output value --state state.value
+wf draft bind concat_ws --revision 6 --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 validate concat_ws
 ```
 
 The command combines two common edits:
 
-- It copies the selected capability output field schema into the root state
+- It copies the selected capability local field schema into the workflow input,
-  field.
+  state, or output schema at the graph path.
-- It merges the output binding `local.<output> -> state.<field>` for the step.
+- It merges the matching step input or output binding.
 
 Use `set-route` separately for outcome routing.
 

skills/wf-cli/SKILL.md

@@ -47,7 +47,8 @@ wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map
 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>
-wf draft bind-output-to-state <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
+wf draft bind <workspace_id> --revision <n> --step <step_id> --from local.<field> --to state.<field>
+wf draft bind <workspace_id> --revision <n> --step <step_id> --from input.<field> --to local.<field>
 wf draft add-step-from-capability <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
 wf draft validate <workspace_id>
 wf draft save <workspace_id> --artifact <artifact_id> --version <n> --title <title>
@@ -82,11 +83,10 @@ For `draft set-input` and `draft set-output`, repeated `--map` flags in one
 command define the complete replacement map. If you split map edits across
 multiple commands, pass `--merge` or the later command replaces the earlier map.
 
-Prefer `draft bind-output-to-state` when a step output should write to a new
+Prefer `draft bind` when a capability step binding also needs schema
-root state field. It declares the matching state schema and merges the output
+projection. Use `input/state -> local` for step inputs and `local ->
-binding in one revision-checked edit.
+state/output` for step outputs. It requires a capability-backed step with
-`bind-output-to-state` requires a capability-backed step with `use`; use JSON
+`use`; use JSON Patch for non-capability/control draft steps.
-Patch for non-capability/control draft steps.
 
 To add a capability step, prefer `wf draft add-step-from-capability` over raw
 JSON Patch when the route, input bindings, and output-to-state bindings are

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

@@ -73,7 +73,7 @@ Prefer focused helpers over JSON Patch for common edits:
 - `set_draft_route`
 - `set_step_input_map`
 - `set_step_output_map`
-- `bind_output_to_state`
+- `bind_draft`
 - `add_step_from_capability`
 - `branch_draft`
 - `handle_draft`
@@ -91,7 +91,8 @@ wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map
 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>
-wf draft bind-output-to-state <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
+wf draft bind <workspace_id> --revision <n> --step <step_id> --from local.<field> --to state.<field>
+wf draft bind <workspace_id> --revision <n> --step <step_id> --from input.<field> --to local.<field>
 wf draft add-step-from-capability <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
 ```
 
@@ -105,18 +106,19 @@ Without `--merge`, `set-input` and `set-output` replace the whole map for that
 step. Use repeated `--map` flags in one command for a complete replacement. Use
 `--merge` only when adding/updating entries over multiple revisions.
 
-- `bind_output_to_state`
+- `bind_draft`
 
-  Declares one root state field from a step capability output schema and merges
+  Declares a workflow input/state/output schema field from a capability local
-  `local.<output> -> state.<field>` into that step's output map. Prefer this
+  input/output schema and merges the matching step binding. Use `input/state ->
-  over manual JSON Patch when validation says a state output target is missing
+  local` for step inputs and `local -> state/output` for step outputs. Prefer
-  from `state_schema`.
+  this over manual JSON Patch when validation says a target schema field is
-  The selected step must have `use` so the helper can find the capability
+  missing. The selected step must have `use` so the helper can find the
-  output schema. It intentionally rejects non-capability/control steps instead
+  capability schema. It intentionally rejects non-capability/control steps
-  of guessing.
+  instead of guessing.
 
 ```bash
-wf draft bind-output-to-state <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
+wf draft bind <workspace_id> --revision <n> --step <step_id> --from local.<field> --to state.<field>
+wf draft bind <workspace_id> --revision <n> --step <step_id> --from input.<field> --to local.<field>
 wf draft validate <workspace_id>
 ```
 
@@ -156,8 +158,7 @@ wf draft validate <workspace_id>
   a `compiled_plan`.
 
 Validation repair hints are product guidance. If a diagnostic suggests
-`bind-output-to-state`, use it before hand-editing `state_schema` or step output
+`wf draft bind`, use it before hand-editing schemas or step bindings.
-bindings.
 
 Use JSON Patch for structural edits the helpers do not cover.
 

skills/wf-workflow/references/workflow-lifecycle.md

@@ -18,10 +18,10 @@ validated, runnable deployment.
      for common edits.
    - `set-input` and `set-output` replace full maps by default; pass `--merge`
      only when adding or updating one entry across a later revision.
-    - Before output-mapping into a new state field, declare it with
+   - Before mapping into a new workflow input, state, or output field, prefer
-      `bind-output-to-state` when it should mirror a capability output property.
+     `wf draft bind --from ... --to ...` when it should mirror a capability
-      It declares the matching state schema and merges the output binding in one
+     local input/output property. It declares the matching schema and merges
-      revision-checked edit.
+     the binding in one revision-checked edit.
    - When adding a new capability-backed step, prefer:
      ```bash
      wf draft add-step-from-capability ...

src/wf_api/draft_authoring.py

@@ -180,15 +180,27 @@ class WorkflowDraftAuthoringApi:
         step = draft_step(workspace.draft, step_id)
         capability_name = step.get("use")
         if not isinstance(capability_name, str) or not capability_name:
-            raise ValueError(f"draft step {step_id!r} does not declare a capability use")
+            raise ValueError(
+                f"draft step {step_id!r} does not declare a capability use"
+            )
         spec = self.context.specs.get_qualified_spec(capability_name)
 
-        source_root, source_parts = _graph_parts(source_path) if not source_path.startswith("local.") else ("local", LocalPath.parse(source_path).parts)
+        source_root, source_parts = (
-        target_root, target_parts = _graph_parts(target_path) if not target_path.startswith("local.") else ("local", LocalPath.parse(target_path).parts)
+            _graph_parts(source_path)
+            if not source_path.startswith("local.")
+            else ("local", LocalPath.parse(source_path).parts)
+        )
+        target_root, target_parts = (
+            _graph_parts(target_path)
+            if not target_path.startswith("local.")
+            else ("local", LocalPath.parse(target_path).parts)
+        )
 
         if target_root == "local" and source_root in {"input", "state"}:
             local_field = _local_field(target_path)
-            input_schema = spec.input_schema_contract or spec.input_model.model_json_schema()
+            input_schema = (
+                spec.input_schema_contract or spec.input_model.model_json_schema()
+            )
             schema_key = "input_schema" if source_root == "input" else "state_schema"
             target_schema = workspace.draft.get(schema_key, {})
             if not isinstance(target_schema, dict):
@@ -218,7 +230,9 @@ class WorkflowDraftAuthoringApi:
 
         if source_root == "local" and target_root in {"state", "output"}:
             local_field = _local_field(source_path)
-            output_schema = spec.output_schema_contract or spec.output_model.model_json_schema()
+            output_schema = (
+                spec.output_schema_contract or spec.output_model.model_json_schema()
+            )
             schema_key = "state_schema" if target_root == "state" else "output_schema"
             target_schema = workspace.draft.get(schema_key, {})
             if not isinstance(target_schema, dict):
@@ -230,7 +244,9 @@ class WorkflowDraftAuthoringApi:
                 target_parts=target_parts,
             )
             output_map = {
-                **self.drafts._step_output_map(workspace_id=workspace_id, step_id=step_id),
+                **self.drafts._step_output_map(
+                    workspace_id=workspace_id, step_id=step_id
+                ),
                 local_field: target_path,
             }
             return await self.drafts.patch_draft_workspace(
@@ -246,7 +262,9 @@ class WorkflowDraftAuthoringApi:
                 ],
             )
 
-        raise ValueError(f"unsupported bind direction: {source_path!r} -> {target_path!r}")
+        raise ValueError(
+            f"unsupported bind direction: {source_path!r} -> {target_path!r}"
+        )
 
     async def add_step_from_capability(
         self,

src/wf_api/draft_payloads.py

@@ -39,7 +39,10 @@ def input_bindings_payload(
 def output_bindings_payload(output_map: dict[str, str]) -> list[dict[str, Any]]:
     """Serialize draft output maps into canonical string-path binding payloads."""
     return [
-        {"source": _local_path_payload(source), "target": _graph_source_path_payload(target)}
+        {
+            "source": _local_path_payload(source),
+            "target": _graph_source_path_payload(target),
+        }
         for source, target in output_map.items()
     ]
 

src/wf_api/drafts.py

@@ -473,6 +473,6 @@ def _draft_repair_hint(
     if not isinstance(output_field, str) or not isinstance(state_path, str):
         return None
     return (
-        f"wf draft bind-output-to-state {workspace_id} --revision {revision} "
+        f"wf draft bind {workspace_id} --revision {revision} "
-        f"--step {step_id} --output {output_field} --state {state_path}"
+        f"--step {step_id} --from local.{output_field} --to {state_path}"
     )

src/wf_api/schema_projection.py

@@ -31,17 +31,23 @@ def project_property_to_schema_path(
     _ensure_object_schema(projected, "target_schema")
     parent = projected
     for index, part in enumerate(target_parts[:-1]):
-        properties = _properties_for_object(parent, ".".join(target_parts[:index]) or "target_schema")
+        properties = _properties_for_object(
+            parent, ".".join(target_parts[:index]) or "target_schema"
+        )
         child = properties.get(part)
         if child is None:
             child = {"type": "object", "properties": {}}
             properties[part] = child
         if not isinstance(child, dict):
-            raise ValueError(f"schema path {'.'.join(target_parts[: index + 1])!r} is not an object")
+            raise ValueError(
+                f"schema path {'.'.join(target_parts[: index + 1])!r} is not an object"
+            )
         _ensure_object_schema(child, ".".join(target_parts[: index + 1]))
         parent = child
 
-    properties = _properties_for_object(parent, ".".join(target_parts[:-1]) or "target_schema")
+    properties = _properties_for_object(
+        parent, ".".join(target_parts[:-1]) or "target_schema"
+    )
     leaf = target_parts[-1]
     if leaf in properties:
         raise ValueError(f"schema path {'.'.join(target_parts)!r} already exists")
@@ -76,15 +82,21 @@ def project_output_property_to_state_schema(
         if msg.startswith("source field ") and "is not declared" in msg:
             raise ValueError(f"output field {output_field!r} is not declared") from exc
         if msg.startswith("source field ") and "not a JSON Schema" in msg:
-            raise ValueError(f"output field {output_field!r} is not a JSON Schema object") from exc
+            raise ValueError(
+                f"output field {output_field!r} is not a JSON Schema object"
+            ) from exc
         if "schema path 'target_schema'" in msg and "is not an object" in msg:
             raise ValueError("state_schema must be an object schema") from exc
         if msg.startswith("schema path ") and "already exists" in msg:
             raise ValueError(f"state field {state_field!r} already exists") from exc
         if "target_schema is not valid JSON Schema" in msg:
-            raise ValueError(f"state_schema is not valid JSON Schema: {msg.split(': ', 1)[1]}") from exc
+            raise ValueError(
+                f"state_schema is not valid JSON Schema: {msg.split(': ', 1)[1]}"
+            ) from exc
         if "source_schema is not valid JSON Schema" in msg:
-            raise ValueError(f"output_schema is not valid JSON Schema: {msg.split(': ', 1)[1]}") from exc
+            raise ValueError(
+                f"output_schema is not valid JSON Schema: {msg.split(': ', 1)[1]}"
+            ) from exc
         raise
 
 

src/wf_cli/commands/drafts.py

@@ -299,42 +299,39 @@ def set_step_output_map(
     )
 
 
-@app.command("bind-output-to-state")
+@app.command("bind")
-def bind_output_to_state(
+def bind_draft(
     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.")
     ],
     step_id: Annotated[str, typer.Option("--step", help="Draft step id.")],
-    output_field: Annotated[
+    source_path: Annotated[
         str,
-        typer.Option("--output", help="Top-level capability output field."),
+        typer.Option("--from", help="Source path, for example input.x or local.y."),
     ],
-    state_path: Annotated[
+    target_path: Annotated[
         str,
-        typer.Option("--state", help="Root state path, for example state.after."),
+        typer.Option("--to", help="Target path, for example local.x or state.y."),
     ],
 ) -> None:
-    """Declare state schema and bind one step output to that state field.
+    """Bind a capability step path and project the matching schema.
 
-    This is the common command to run before validation when a step output
+    Direction matters. Use input/state -> local for step inputs and local ->
-    should write to a new state field. It copies the selected capability output
+    state/output for step outputs. Run `wf draft validate <workspace_id>` after
-    field schema into state_schema and merges the output binding
+    this command.
-    local.<output> -> state.<field>.
-
-    Run `wf draft validate <workspace_id>` after this command.
     """
     context = load_cli_context(ctx)
     emit_json(
         run_cli_operation(
             context,
-            context.handlers.bind_output_to_state(
+            context.handlers.bind_draft(
                 workspace_id=workspace_id,
                 revision=revision,
                 step_id=step_id,
-                output_field=output_field,
+                source_path=source_path,
-                state_path=state_path,
+                target_path=target_path,
             ),
         )
     )

src/wf_mcp/workflow_surface/models.py

@@ -243,18 +243,14 @@ class SetStepOutputMapRequest(BaseModel):
     )
 
 
-class BindOutputToStateRequest(BaseModel):
+class BindDraftRequest(BaseModel):
-    """Typed MCP request for binding one step output to one root state field."""
+    """Typed MCP request for binding one draft step path with schema projection."""
 
     workspace_id: WorkspaceId
     revision: int = Field(ge=1, description="Expected current workspace revision.")
-    step_id: str = Field(description="Draft step id whose capability output is used.")
+    step_id: str = Field(description="Capability-backed draft step id.")
-    output_field: str = Field(
+    source_path: str = Field(description="Source path, for example input.x or local.y.")
-        description="Top-level output field to bind, for example after."
+    target_path: str = Field(description="Target path, for example local.x or state.y.")
-    )
-    state_path: str = Field(
-        description="Root state path to declare and bind, for example state.after."
-    )
 
 
 class AddStepFromCapabilityRequest(BaseModel):

src/wf_mcp/workflow_surface/tools.py

@@ -13,7 +13,7 @@ from wf_mcp.broker.service.workflow_operation_context import context_from_servic
 
 from .models import (
     AddStepFromCapabilityRequest,
-    BindOutputToStateRequest,
+    BindDraftRequest,
     BranchDraftRequest,
     CallCapabilityResult,
     CompileDraftWorkspaceRequest,
@@ -461,23 +461,24 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
         )
 
     @server.tool(
-        name="wf.workflow.bind_output_to_state",
+        name="wf.workflow.bind",
-        title="Bind Output To State",
+        title="Bind Draft",
         description=(
-            "Declare one root state field from a draft step capability output "
+            "Bind a capability step path and project the matching schema. "
-            "schema and bind local.<output> to that state path."
+            "Use input/state -> local for step inputs and local -> state/output "
+            "for step outputs."
         ),
     )
-    async def bind_output_to_state(
+    async def bind_draft(
-        request: BindOutputToStateRequest,
+        request: BindDraftRequest,
     ) -> DraftWorkspaceResult:
         return DraftWorkspaceResult.model_validate(
-            await handlers.bind_output_to_state(
+            await handlers.bind_draft(
                 workspace_id=request.workspace_id,
                 revision=request.revision,
                 step_id=request.step_id,
-                output_field=request.output_field,
+                source_path=request.source_path,
-                state_path=request.state_path,
+                target_path=request.target_path,
             )
         )
 

src/wf_transport_rpc_http/__init__.py

@@ -6,6 +6,7 @@ from .errors import WorkflowRpcError
 from .models import (
     AddStepFromCapabilityParams,
     AdminEmptyParams,
+    BindDraftParams,
     BranchDraftParams,
     CallCapabilityParams,
     CompileDraftWorkspaceParams,
@@ -48,6 +49,7 @@ from .models import (
 __all__ = [
     "AddStepFromCapabilityParams",
     "AdminEmptyParams",
+    "BindDraftParams",
     "BranchDraftParams",
     "CallCapabilityParams",
     "CompileDraftWorkspaceParams",

src/wf_transport_rpc_http/client/drafts.py

@@ -141,23 +141,23 @@ class RpcDraftClientMixin:
             },
         )
 
-    async def bind_output_to_state(
+    async def bind_draft(
         self: RpcCaller,
         *,
         workspace_id: str,
         revision: int,
         step_id: str,
-        output_field: str,
+        source_path: str,
-        state_path: str,
+        target_path: str,
     ) -> dict[str, Any]:
         return await self._call(
-            "workflow.draft_workspaces.bind_output_to_state",
+            "workflow.draft_workspaces.bind",
             {
                 "workspace_id": workspace_id,
                 "revision": revision,
                 "step_id": step_id,
-                "output_field": output_field,
+                "source_path": source_path,
-                "state_path": state_path,
+                "target_path": target_path,
             },
         )
 

src/wf_transport_rpc_http/methods/drafts.py

@@ -9,7 +9,7 @@ from wf_server import WorkflowServer
 from ..errors import WorkflowRpcError, raise_workflow_rpc_error
 from ..models import (
     AddStepFromCapabilityParams,
-    BindOutputToStateParams,
+    BindDraftParams,
     BranchDraftParams,
     CompileDraftWorkspaceParams,
     CreateArtifactFromWorkspaceParams,
@@ -183,19 +183,19 @@ def register_methods(
             raise_workflow_rpc_error(exc)
 
     @entrypoint.method(
-        name="workflow.draft_workspaces.bind_output_to_state",
+        name="workflow.draft_workspaces.bind",
         errors=[WorkflowRpcError],
     )
-    async def workflow_draft_workspaces_bind_output_to_state(
+    async def workflow_draft_workspaces_bind(
-        params: BindOutputToStateParams = RpcParams(),
+        params: BindDraftParams = RpcParams(),
     ) -> dict[str, Any]:
         try:
-            return await server.api.bind_output_to_state(
+            return await server.api.bind_draft(
                 workspace_id=params.workspace_id,
                 revision=params.revision,
                 step_id=params.step_id,
-                output_field=params.output_field,
+                source_path=params.source_path,
-                state_path=params.state_path,
+                target_path=params.target_path,
             )
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)

src/wf_transport_rpc_http/models.py

@@ -141,12 +141,12 @@ class SetStepOutputMapParams(RpcParamsModel):
     merge: bool = False
 
 
-class BindOutputToStateParams(RpcParamsModel):
+class BindDraftParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
     revision: int = Field(ge=1)
     step_id: str = Field(min_length=1)
-    output_field: str = Field(min_length=1)
+    source_path: str = Field(min_length=1)
-    state_path: str = Field(min_length=1)
+    target_path: str = Field(min_length=1)
 
 
 class AddStepFromCapabilityParams(RpcParamsModel):

tests/wf_api/test_drafts_service.py

@@ -342,7 +342,7 @@ async def test_validate_draft_workspace_refreshes_status(tmp_path: Path) -> None
 
 
 @pytest.mark.asyncio
-async def test_validate_draft_workspace_suggests_bind_output_to_state(
+async def test_validate_draft_workspace_suggests_bind(
     tmp_path: Path,
 ) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_repair_hint")
@@ -381,8 +381,8 @@ async def test_validate_draft_workspace_suggests_bind_output_to_state(
     assert diagnostic["code"] == "invalid_destination_path"
     assert diagnostic["step_id"] == "snap"
     assert diagnostic["repair_hint"] == (
-        "wf draft bind-output-to-state snapshot_ws --revision 1 "
+        "wf draft bind snapshot_ws --revision 1 "
-        "--step snap --output after --state state.after"
+        "--step snap --from local.after --to state.after"
     )
 
 
@@ -560,7 +560,9 @@ async def test_bind_draft_workflow_input_to_step_input_projects_input_schema(
         source_path="input.text",
         target_path="local.text",
     )
-    workspace = await api.get_draft_workspace(workspace_id="bind_ws", include_draft=True)
+    workspace = await api.get_draft_workspace(
+        workspace_id="bind_ws", include_draft=True
+    )
 
     assert result["revision"] == 2
     assert workspace["draft"]["input_schema"]["properties"]["text"]["type"] == "string"
@@ -584,7 +586,13 @@ async def test_bind_draft_output_to_nested_state_projects_state_schema(
             "state_schema": {"type": "object", "properties": {}},
             "output_schema": {"type": "object", "properties": {}},
             "start": "snap",
-            "steps": {"snap": {"use": "demo.personal.snapshot_tool", "input": [], "output": []}},
+            "steps": {
+                "snap": {
+                    "use": "demo.personal.snapshot_tool",
+                    "input": [],
+                    "output": [],
+                }
+            },
             "routes": {"snap": {"ok": "__end__"}},
         },
     )
@@ -596,11 +604,15 @@ async def test_bind_draft_output_to_nested_state_projects_state_schema(
         source_path="local.after",
         target_path="state.session.after",
     )
-    workspace = await api.get_draft_workspace(workspace_id="snapshot_ws", include_draft=True)
+    workspace = await api.get_draft_workspace(
+        workspace_id="snapshot_ws", include_draft=True
+    )
 
     assert result["revision"] == 2
     assert (
-        workspace["draft"]["state_schema"]["properties"]["session"]["properties"]["after"]["$ref"]
+        workspace["draft"]["state_schema"]["properties"]["session"]["properties"][
+            "after"
+        ]["$ref"]
         == "#/$defs/_Snapshot"
     )
     assert workspace["draft"]["steps"]["snap"]["output"] == [

tests/wf_cli/test_app.py

@@ -144,13 +144,13 @@ def test_wf_draft_map_help_explains_replace_merge_and_validate() -> None:
     assert "draft validate" in output_help
 
 
-def test_wf_draft_bind_output_to_state_help_explains_composed_edit() -> None:
+def test_wf_draft_bind_help_explains_direction() -> None:
-    result = runner.invoke(app, ["draft", "bind-output-to-state", "--help"])
+    result = runner.invoke(app, ["draft", "bind", "--help"])
 
     assert result.exit_code == 0
     output = " ".join(result.output.split())
-    assert "state schema" in output
+    assert "--from" in output
-    assert "output binding" in output
+    assert "--to" in output
     assert "validate" in output
 
 

tests/wf_cli/test_remote_target.py

@@ -689,9 +689,9 @@ def test_wf_remote_draft_artifact_deploy_lifecycle(monkeypatch, tmp_path) -> Non
     )
     assert invalid_validated.exit_code == 0, invalid_validated.output
     assert '"status": "invalid"' in invalid_validated.output
-    assert "bind-output-to-state repair_ws --revision 2" in invalid_validated.output
+    assert "bind repair_ws --revision 2" in invalid_validated.output
     assert (
-        "--step call --output value --state state.missing" in invalid_validated.output
+        "--step call --from local.value --to state.missing" in invalid_validated.output
     )
 
     saved_artifact = runner.invoke(
@@ -1139,7 +1139,7 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
     ]
 
 
-def test_wf_draft_bind_output_to_state_uses_rpc_target(monkeypatch, tmp_path) -> None:
+def test_wf_draft_bind_uses_rpc_target(monkeypatch, tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     _patch_rpc_client_to_server(monkeypatch, server)
     config_path = tmp_path / "wf.json"
@@ -1166,16 +1166,16 @@ def test_wf_draft_bind_output_to_state_uses_rpc_target(monkeypatch, tmp_path) ->
         [
             *base_args,
             "draft",
-            "bind-output-to-state",
+            "bind",
             "snapshot_ws",
             "--revision",
             "1",
             "--step",
             "call",
-            "--output",
+            "--from",
-            "value",
+            "local.value",
-            "--state",
+            "--to",
-            "state.value",
+            "state.result",
         ],
     )
 

tests/wf_mcp/server/test_config.py

@@ -56,7 +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.bind_output_to_state" in names
+            assert "wf.workflow.bind" in names
             assert "wf.workflow.add_step_from_capability" in names
             assert "wf.workflow.create_minimal_draft_workspace" in names
             assert "wf.workflow.create_draft_workspace_from_capability" in names

tests/wf_transport_rpc_http/test_app.py

@@ -724,13 +724,13 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
         )
         state_bound = await _rpc(
             client,
-            "workflow.draft_workspaces.bind_output_to_state",
+            "workflow.draft_workspaces.bind",
             {
                 "workspace_id": "focused_ws",
                 "revision": 7,
                 "step_id": "call",
-                "output_field": "value",
+                "source_path": "local.value",
-                "state_path": "state.extra_value",
+                "target_path": "state.extra_value",
             },
         )
         fetched = await _rpc(

tests/wf_transport_rpc_http/test_client.py

@@ -501,12 +501,12 @@ async def test_rpc_client_draft_workspace_focused_edit_methods(tmp_path) -> None
             output_map={"extra": "state.extra"},
             merge=True,
         )
-        state_bound = await client.bind_output_to_state(
+        state_bound = await client.bind_draft(
             workspace_id="client_focused_ws",
             revision=7,
             step_id="call",
-            output_field="value",
+            source_path="local.value",
-            state_path="state.extra_value",
+            target_path="state.extra_value",
         )
 
     assert named["revision"] == 2