feat: update capability-backed draft steps

src/wf_api/__init__.py

@@ -18,6 +18,7 @@ from .constants import (
 )
 from .deployments import WorkflowDeploymentApi
 from .draft_authoring import RouteSource
+from .draft_updates import CapabilityStepUpdate
 from .drafts import WorkflowDraftApi
 from .durable_context import durable_workflow_api, require_workflow_stores
 from .listing import matches_query, paged_list_payload
@@ -72,6 +73,7 @@ __all__ = [
     "DEFAULT_CALL_STEP_ID",
     "AuthRecord",
     "AuthStore",
+    "CapabilityStepUpdate",
     "builtin_sources",
     "get_qualified_spec",
     "matches_query",

src/wf_api/draft_authoring.py

@@ -52,6 +52,7 @@ from .draft_payloads import (
     output_bindings_payload,
     state_root_field,
 )
+from .draft_updates import CapabilityStepUpdate
 from .drafts import (
     WorkflowDraftApi,
     _draft_input_maps,
@@ -184,6 +185,15 @@ def _step_output_bindings_patch(
     return patch
 
 
+@dataclass(frozen=True)
+class _ProjectedStepInputBindings:
+    """Canonical step inputs plus workflow schemas projected from their sources."""
+
+    payload: list[dict[str, Any]]
+    input_schema: dict[str, Any]
+    state_schema: dict[str, Any]
+
+
 class WorkflowDraftAuthoringApi:
     """Capability-aware semantic edits over revisioned workflow drafts."""
 
@@ -436,6 +446,40 @@ class WorkflowDraftAuthoringApi:
             raise ValueError(
                 f"draft step {step_id!r} does not declare a capability use"
             )
+        projected = self._project_step_input_bindings(
+            workspace=workspace,
+            capability_name=capability_name,
+            bindings=bindings,
+        )
+
+        if (
+            step.get("input", []) == projected.payload
+            and workspace.draft.get("input_schema", {}) == projected.input_schema
+            and workspace.draft.get("state_schema", {}) == projected.state_schema
+        ):
+            return summarize_draft_workspace(workspace)
+
+        patch = _step_input_bindings_patch(
+            workspace=workspace,
+            step_id=step_id,
+            bindings=projected.payload,
+            input_schema=projected.input_schema,
+            state_schema=projected.state_schema,
+        )
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
+        )
+
+    def _project_step_input_bindings(
+        self,
+        *,
+        workspace: WorkflowDraftWorkspace,
+        capability_name: str,
+        bindings: Sequence[InputBinding],
+    ) -> _ProjectedStepInputBindings:
+        """Validate canonical inputs and project missing workflow source schemas."""
         spec = self.context.specs.get_qualified_spec(capability_name)
         capability_schema = (
             spec.input_schema_contract or spec.input_model.model_json_schema()
@@ -495,19 +539,95 @@ class WorkflowDraftAuthoringApi:
                     projected_state = target_schema
 
         payload = [binding.model_dump(mode="json") for binding in bindings]
+        return _ProjectedStepInputBindings(
+            payload=payload,
+            input_schema=projected_input,
+            state_schema=projected_state,
+        )
+
+    async def update_capability_step(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        update: CapabilityStepUpdate,
+    ) -> dict[str, Any]:
+        """Patch capability metadata and optional canonical inputs atomically."""
+        checked = self._workspace_if_revision_matches(
+            workspace_id=workspace_id,
+            revision=revision,
+        )
+        if isinstance(checked, dict):
+            return checked
+        workspace = checked
+        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} is not capability-backed")
+        current = DraftUseStep.model_validate(step)
+
+        changes: dict[str, object] = {}
+        for field in ("desc", "retry", "timeout_seconds"):
+            if field in update.model_fields_set:
+                changes[field] = getattr(update, field)
+
+        projected: _ProjectedStepInputBindings | None = None
+        if "input" in update.model_fields_set:
+            if update.input is None:
+                raise AssertionError("validated capability update has null input")
+            projected = self._project_step_input_bindings(
+                workspace=workspace,
+                capability_name=capability_name,
+                bindings=update.input,
+            )
+            changes["input"] = update.input
+
+        changed = current.model_copy(update=changes)
+        step_payload = changed.model_dump(
+            mode="json",
+            by_alias=True,
+            exclude_none=True,
+        )
+        input_schema = (
+            projected.input_schema
+            if projected is not None
+            else _draft_schema(workspace.draft, "input_schema")
+        )
+        state_schema = (
+            projected.state_schema
+            if projected is not None
+            else _draft_schema(workspace.draft, "state_schema")
+        )
         if (
-            step.get("input", []) == payload
+            step == step_payload
-            and workspace.draft.get("input_schema", {}) == projected_input
+            and workspace.draft.get("input_schema", {}) == input_schema
-            and workspace.draft.get("state_schema", {}) == projected_state
+            and workspace.draft.get("state_schema", {}) == state_schema
         ):
             return summarize_draft_workspace(workspace)
 
-        patch = _step_input_bindings_patch(
+        if projected is None:
-            workspace=workspace,
+            next_draft = deepcopy(workspace.draft)
-            step_id=step_id,
+            next_draft["steps"][step_id] = step_payload
-            bindings=payload,
+            return await self.drafts.replace_validated_draft_document(
-            input_schema=projected_input,
+                workspace_id=workspace_id,
-            state_schema=projected_state,
+                revision=revision,
+                draft=next_draft,
+            )
+
+        patch: list[dict[str, Any]] = []
+        for key, value in (
+            ("input_schema", input_schema),
+            ("state_schema", state_schema),
+        ):
+            if workspace.draft.get(key, {}) != value:
+                patch.append({"op": "replace", "path": f"/{key}", "value": value})
+        patch.append(
+            {
+                "op": "replace",
+                "path": f"/steps/{escape_json_pointer(step_id)}",
+                "value": step_payload,
+            }
         )
         return await self.drafts.patch_draft_workspace(
             workspace_id=workspace_id,
@@ -927,7 +1047,11 @@ class WorkflowDraftAuthoringApi:
         route_from_outcome: str = DEFAULT_OK_OUTCOME,
         routes: dict[str, str] | None = None,
         input_map: dict[str, str] | None = None,
+        input_bindings: Sequence[InputBinding] | None = None,
         bind_outputs: dict[str, str] | None = None,
+        desc: str | None = None,
+        retry: int | None = None,
+        timeout_seconds: int | None = None,
     ) -> dict[str, Any]:
         """Add one capability step plus explicit route/map/schema wiring.
 
@@ -947,6 +1071,8 @@ class WorkflowDraftAuthoringApi:
             raise ValueError("draft steps must be an object")
         if step_id in steps:
             raise ValueError(f"draft step {step_id!r} already exists")
+        if input_map is not None and input_bindings is not None:
+            raise ValueError("input_map and input_bindings are mutually exclusive")
 
         spec = self.context.specs.get_qualified_spec(capability_name)
         output_schema = (
@@ -999,46 +1125,20 @@ class WorkflowDraftAuthoringApi:
                     f"routes for {missing_outcomes}"
                 )
 
-        input_map = input_map or {}
+        if input_bindings is None:
+            canonical_inputs = TypeAdapter(list[InputBinding]).validate_python(
+                input_bindings_payload(input_map or {}, {})
+            )
+        else:
+            canonical_inputs = list(input_bindings)
         bind_outputs = bind_outputs or {}
-        projected_input_schema = workspace.draft.get("input_schema", {})
+        projected_inputs = self._project_step_input_bindings(
-        if not isinstance(projected_input_schema, dict):
+            workspace=workspace,
-            raise ValueError("draft input_schema must be an object")
+            capability_name=capability_name,
-        projected_state_schema = state_schema
+            bindings=canonical_inputs,
-        input_schema = (
-            spec.input_schema_contract or spec.input_model.model_json_schema()
         )
-        for graph_path, local_path in input_map.items():
+        projected_input_schema = projected_inputs.input_schema
-            try:
+        projected_state_schema = projected_inputs.state_schema
-                source_root, source_parts = _graph_parts(graph_path)
-                local_parts = LocalPath.parse(local_path).parts
-            except ValueError:
-                continue
-            if source_root not in {"input", "state"}:
-                continue
-            if not local_parts:
-                # `.` binds the whole graph value to the whole node input. It has
-                # no capability-property path from which to project one field.
-                continue
-            schema_key = "input_schema" if source_root == "input" else "state_schema"
-            target_schema = (
-                projected_input_schema
-                if source_root == "input"
-                else projected_state_schema
-            )
-            if schema_path_exists(target_schema, source_parts):
-                continue
-            projected = project_schema_path_to_schema_path(
-                target_schema=target_schema,
-                source_schema=input_schema,
-                source_parts=local_parts,
-                target_parts=source_parts,
-                allow_existing_equivalent=True,
-            )
-            if schema_key == "input_schema":
-                projected_input_schema = projected
-            else:
-                projected_state_schema = projected
         for output_field, path in bind_outputs.items():
             sf = state_root_field(path)
             projected_state_schema = project_output_property_to_state_schema(
@@ -1049,15 +1149,23 @@ class WorkflowDraftAuthoringApi:
                 allow_existing_equivalent=True,
             )
 
+        step_payload: dict[str, Any] = {
+            "use": capability_name,
+            "input": projected_inputs.payload,
+            "output": output_bindings_payload(bind_outputs),
+        }
+        if desc is not None:
+            step_payload["desc"] = desc
+        if retry is not None:
+            step_payload["retry"] = retry
+        if timeout_seconds is not None:
+            step_payload["timeout_seconds"] = timeout_seconds
+
         patch: list[dict[str, Any]] = [
             {
                 "op": "add",
                 "path": f"/steps/{escape_json_pointer(step_id)}",
-                "value": {
+                "value": step_payload,
-                    "use": capability_name,
-                    "input": input_bindings_payload(input_map, {}),
-                    "output": output_bindings_payload(bind_outputs),
-                },
             },
             {
                 "op": "add",

src/wf_api/draft_updates.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from typing import Self
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from wf_core.models.steps import InputBinding
+
+
+class CapabilityStepUpdate(BaseModel):
+    """Presence-aware patch for one existing capability-backed draft step."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    desc: str | None = Field(default=None, min_length=1)
+    retry: int | None = Field(default=None, ge=0)
+    timeout_seconds: int | None = Field(default=None, gt=0)
+    input: list[InputBinding] | None = None
+
+    @model_validator(mode="after")
+    def validate_patch_shape(self) -> Self:
+        if not self.model_fields_set:
+            raise ValueError("capability step update requires at least one field")
+        if "input" in self.model_fields_set and self.input is None:
+            raise ValueError("capability step update input must be a list")
+        return self

src/wf_api/drafts.py

@@ -19,6 +19,9 @@ from wf_artifacts import (
 from wf_artifacts import (
     patch_draft_workspace as patch_draft_workspace_record,
 )
+from wf_artifacts import (
+    replace_validated_draft_document as replace_validated_draft_document_record,
+)
 from wf_core.models.schemas import NodeDef
 from wf_core.models.steps import (
     InputBinding,
@@ -289,6 +292,21 @@ class WorkflowDraftApi:
             node_defs_for_draft=self._node_defs_for_draft,
         )
 
+    async def replace_validated_draft_document(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        draft: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Persist a focused, structurally validated edit without provider lookup."""
+        return replace_validated_draft_document_record(
+            self._draft_store(),
+            workspace_id=workspace_id,
+            revision=revision,
+            draft=draft,
+        )
+
     async def set_draft_name(
         self,
         *,

src/wf_artifacts/__init__.py

@@ -12,6 +12,7 @@ from .draft_workspaces import (
     ensure_workspace_id,
     get_draft_workspace,
     patch_draft_workspace,
+    replace_validated_draft_document,
     summarize_draft_workspace,
 )
 from .drafts import (
@@ -91,6 +92,7 @@ __all__ = [
     "logical_ref_for_concrete_ref",
     "normalize_plan_node_refs",
     "patch_draft_workspace",
+    "replace_validated_draft_document",
     "patch_workflow_draft",
     "summarize_draft_workspace",
     "validate_deployment_dependencies",

src/wf_artifacts/draft_workspaces/__init__.py

@@ -1,4 +1,9 @@
-from .api import create_draft_workspace, get_draft_workspace, patch_draft_workspace
+from .api import (
+    create_draft_workspace,
+    get_draft_workspace,
+    patch_draft_workspace,
+    replace_validated_draft_document,
+)
 from .models import (
     WorkflowDraftWorkspace,
     ensure_workspace_id,
@@ -19,5 +24,6 @@ __all__ = [
     "ensure_workspace_id",
     "get_draft_workspace",
     "patch_draft_workspace",
+    "replace_validated_draft_document",
     "summarize_draft_workspace",
 ]

src/wf_artifacts/draft_workspaces/api.py

@@ -102,6 +102,39 @@ def patch_draft_workspace(
     return summarize_draft_workspace(next_workspace)
 
 
+def replace_validated_draft_document(
+    store: DraftWorkspaceStore,
+    *,
+    workspace_id: str,
+    revision: int,
+    draft: JsonObject,
+) -> JsonObject:
+    """Atomically replace a structurally validated draft without resolving providers.
+
+    This is for focused edits whose changed fields have already been validated and
+    cannot alter capability contracts. The workspace's existing semantic validation
+    snapshot is preserved.
+    """
+    workspace = store.get_workspace(workspace_id)
+    if workspace.revision != revision:
+        return _revision_conflict_payload(workspace, revision)
+    canonical_draft = WorkflowDraft.model_validate(draft).model_dump(mode="json")
+    if canonical_draft == workspace.draft:
+        return summarize_draft_workspace(workspace)
+    next_workspace = workspace.model_copy(
+        update={
+            "revision": workspace.revision + 1,
+            "draft": canonical_draft,
+            "updated_at_epoch_ms": _now_ms(),
+        }
+    )
+    try:
+        store.replace_workspace(next_workspace, expected_revision=revision)
+    except DraftWorkspaceConflictError as exc:
+        return _revision_conflict_payload(exc.workspace, revision)
+    return summarize_draft_workspace(next_workspace)
+
+
 def get_draft_workspace(
     store: DraftWorkspaceStore,
     *,

tests/wf_api/test_drafts_service.py

@@ -5,10 +5,11 @@ from pathlib import Path
 from typing import Any, Literal, cast
 
 import pytest
-from pydantic import BaseModel, Field, TypeAdapter
+from pydantic import BaseModel, Field, TypeAdapter, ValidationError
 
 from tests.wf_mcp.test_support import echo_tool
 from wf_api.draft_authoring import RouteSource, WorkflowDraftAuthoringApi
+from wf_api.draft_updates import CapabilityStepUpdate
 from wf_api.drafts import WorkflowDraftApi
 from wf_api.models import RawWorkflowPlan
 from wf_api.service import WorkflowApi
@@ -29,6 +30,321 @@ from wf_mcp.storage import FileStore
 from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
 
 
+def test_capability_step_update_preserves_field_presence() -> None:
+    update = CapabilityStepUpdate.model_validate(
+        {"desc": None, "retry": 0, "input": []}
+    )
+
+    assert update.model_fields_set == {"desc", "retry", "input"}
+    assert update.desc is None
+    assert update.retry == 0
+    assert update.input == []
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {},
+        {"input": None},
+        {"desc": ""},
+        {"retry": -1},
+        {"timeout_seconds": 0},
+        {"unknown": True},
+    ],
+)
+def test_capability_step_update_rejects_invalid_patch(payload: object) -> None:
+    with pytest.raises(ValidationError):
+        CapabilityStepUpdate.model_validate(payload)
+
+
+@pytest.mark.asyncio
+async def test_update_capability_step_changes_metadata_and_inputs_atomically(
+    tmp_path: Path,
+) -> None:
+    draft_api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "update_capability"),
+        register_echo=True,
+    )
+    draft = _echo_draft()
+    draft["steps"]["echo"].update(
+        {
+            "desc": "Old description",
+            "retry": 1,
+            "timeout_seconds": 10,
+        }
+    )
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=draft)
+
+    result = await authoring.update_capability_step(
+        workspace_id="echo",
+        revision=1,
+        step_id="echo",
+        update=CapabilityStepUpdate.model_validate(
+            {
+                "desc": "New description",
+                "retry": 0,
+                "timeout_seconds": None,
+                "input": [{"value": "fixed", "target": "text"}],
+            }
+        ),
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
+    step = inspected["draft"]["steps"]["echo"]
+
+    assert result["revision"] == 2
+    assert step["use"] == "demo.personal.echo_tool"
+    assert step["desc"] == "New description"
+    assert step["retry"] == 0
+    assert step["timeout_seconds"] is None
+    assert step["input"] == [{"value": "fixed", "target": "text"}]
+    assert step["output"] == [{"source": "echoed", "target": "state.echoed"}]
+    assert inspected["draft"]["routes"]["echo"] == {"ok": "__end__"}
+
+    compiled = await draft_api.compile_draft_workspace(workspace_id="echo")
+    run = await service.run_workflow_from_plan(
+        RawWorkflowPlan.model_validate(compiled["compiled_plan"]),
+        {"text": "ignored"},
+    )
+    assert run.error is None
+    assert run.output == {"echoed": "fixed"}
+
+
+@pytest.mark.asyncio
+async def test_update_capability_step_preserves_omitted_fields_and_exact_noop(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "update_capability_noop"),
+        register_echo=True,
+    )
+    draft = _echo_draft()
+    draft["steps"]["echo"].update({"desc": "Keep", "retry": 2, "timeout_seconds": 15})
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=draft)
+
+    first = await authoring.update_capability_step(
+        workspace_id="echo",
+        revision=1,
+        step_id="echo",
+        update=CapabilityStepUpdate(retry=2),
+    )
+    second = await authoring.update_capability_step(
+        workspace_id="echo",
+        revision=first["revision"],
+        step_id="echo",
+        update=CapabilityStepUpdate(desc=None),
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
+
+    assert first["revision"] == 1
+    assert second["revision"] == 2
+    step = inspected["draft"]["steps"]["echo"]
+    assert step["desc"] is None
+    assert step["retry"] == 2
+    assert step["timeout_seconds"] == 15
+
+
+@pytest.mark.asyncio
+async def test_update_capability_step_clearing_absent_metadata_is_exact_noop(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "update_capability_null_noop"),
+        register_echo=True,
+    )
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=_echo_draft())
+
+    result = await authoring.update_capability_step(
+        workspace_id="echo",
+        revision=1,
+        step_id="echo",
+        update=CapabilityStepUpdate(retry=None),
+    )
+
+    assert result["revision"] == 1
+
+
+@pytest.mark.asyncio
+async def test_update_capability_step_metadata_does_not_resolve_capability(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "update_capability_metadata"),
+        register_echo=True,
+    )
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=_echo_draft())
+
+    def fail_lookup(_provider: object, _capability_name: str) -> None:
+        raise AssertionError("metadata-only update resolved the capability")
+
+    monkeypatch.setattr(
+        type(authoring.context.specs),
+        "get_qualified_spec",
+        fail_lookup,
+    )
+    result = await authoring.update_capability_step(
+        workspace_id="echo",
+        revision=1,
+        step_id="echo",
+        update=CapabilityStepUpdate(desc="Metadata only"),
+    )
+
+    assert result["revision"] == 2
+
+
+@pytest.mark.asyncio
+async def test_update_capability_step_stale_revision_wins_over_semantic_errors(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "update_capability_stale"),
+        register_echo=True,
+    )
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=_echo_draft())
+
+    result = await authoring.update_capability_step(
+        workspace_id="echo",
+        revision=2,
+        step_id="missing",
+        update=CapabilityStepUpdate(
+            input=[InputValueBinding(target=LocalPath.of("missing"), value="bad")]
+        ),
+    )
+
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+
+
+@pytest.mark.asyncio
+async def test_update_capability_step_rejects_wrong_kind_and_invalid_input_atomically(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "update_capability_invalid"),
+        register_echo=True,
+    )
+    draft = _echo_draft()
+    draft["steps"]["joined"] = {"join": ["echo"]}
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=draft)
+    before = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
+
+    with pytest.raises(ValueError, match="not capability-backed"):
+        await authoring.update_capability_step(
+            workspace_id="echo",
+            revision=1,
+            step_id="joined",
+            update=CapabilityStepUpdate(desc="No"),
+        )
+    with pytest.raises(ValueError, match=r"bindings\[0\]\.target"):
+        await authoring.update_capability_step(
+            workspace_id="echo",
+            revision=1,
+            step_id="echo",
+            update=CapabilityStepUpdate(
+                desc="Must not persist",
+                input=[
+                    InputValueBinding(
+                        target=LocalPath.of("missing"),
+                        value="bad",
+                    )
+                ],
+            ),
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_add_step_from_capability_accepts_metadata_and_canonical_inputs(
+    tmp_path: Path,
+) -> None:
+    draft_api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "add_capability_parity"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _structured_report)
+    draft = _structured_report_draft()
+    draft["steps"] = {}
+    draft["routes"] = {}
+    await draft_api.create_draft_workspace(workspace_id="report", draft=draft)
+
+    result = await authoring.add_step_from_capability(
+        workspace_id="report",
+        revision=1,
+        step_id="report",
+        capability_name="demo.personal.structured_report",
+        routes={"ok": "__end__"},
+        desc="Publish report",
+        retry=0,
+        timeout_seconds=30,
+        input_bindings=[
+            InputPathBinding(
+                path=GraphSourcePath.state("report", "title"),
+                target=LocalPath.of("request", "title"),
+            ),
+            InputValueBinding(
+                target=LocalPath.of("request", "format"),
+                value="markdown",
+            ),
+        ],
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="report",
+        include_draft=True,
+    )
+    step = inspected["draft"]["steps"]["report"]
+
+    assert result["revision"] == 2
+    assert step["desc"] == "Publish report"
+    assert step["retry"] == 0
+    assert step["timeout_seconds"] == 30
+    assert step["input"] == [
+        {"path": "state.report.title", "target": "request.title"},
+        {"value": "markdown", "target": "request.format"},
+    ]
+    assert (
+        inspected["draft"]["state_schema"]["properties"]["report"]["properties"][
+            "title"
+        ]["type"]
+        == "string"
+    )
+
+
+@pytest.mark.asyncio
+async def test_add_step_from_capability_rejects_both_input_forms(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "add_capability_exclusive"),
+        register_echo=True,
+    )
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=_echo_draft())
+
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        await authoring.add_step_from_capability(
+            workspace_id="echo",
+            revision=1,
+            step_id="other",
+            capability_name="demo.personal.echo_tool",
+            routes={"ok": "__end__"},
+            input_map={},
+            input_bindings=[],
+        )
+
+
 def _echo_draft() -> dict[str, Any]:
     return {
         "name": "echo",