fix: harden capability step update semantics

docs/superpowers/plans/2026-07-26-capability-step-update.md

@@ -60,7 +60,7 @@ pytest, Ruff, basedpyright.
 - Preserves:
   `WorkflowDraftAuthoringApi.set_step_input_bindings(...)`.
 
-- [ ] **Step 1: Write failing update-model tests**
+- [x] **Step 1: Write failing update-model tests**
 
 Add focused tests in `tests/wf_api/test_drafts_service.py`:
 
@@ -91,7 +91,7 @@ def test_capability_step_update_rejects_invalid_patch(payload: object) -> None:
         CapabilityStepUpdate.model_validate(payload)
 ```
 
-- [ ] **Step 2: Run the model tests red**
+- [x] **Step 2: Run the model tests red**
 
 Run:
 
@@ -104,7 +104,7 @@ Run:
 
 Expected: collection fails because `CapabilityStepUpdate` does not exist.
 
-- [ ] **Step 3: Implement the update model**
+- [x] **Step 3: Implement the update model**
 
 Create `src/wf_api/draft_updates.py`:
 
@@ -139,7 +139,7 @@ class CapabilityStepUpdate(BaseModel):
 
 Export it from `src/wf_api/__init__.py`. Run the model tests green.
 
-- [ ] **Step 4: Write failing semantic update tests**
+- [x] **Step 4: Write failing semantic update tests**
 
 Add a helper that creates a capability-backed draft step containing:
 
@@ -218,7 +218,7 @@ Add separate tests for:
 - path/literal input replacement projects schemas and preserves order;
 - compile/run succeeds after one combined metadata/input update.
 
-- [ ] **Step 5: Run semantic tests red**
+- [x] **Step 5: Run semantic tests red**
 
 Run:
 
@@ -231,7 +231,7 @@ Run:
 
 Expected: failures because the authoring method does not exist.
 
-- [ ] **Step 6: Extract one shared input-binding preflight**
+- [x] **Step 6: Extract one shared input-binding preflight**
 
 In `src/wf_api/draft_authoring.py`, add a private result model:
 
@@ -280,7 +280,7 @@ the update method:
 
 Expected: PASS.
 
-- [ ] **Step 7: Implement atomic capability-step update**
+- [x] **Step 7: Implement atomic capability-step update**
 
 Implement:
 
@@ -339,7 +339,7 @@ values here come from `CapabilityStepUpdate`. Build one patch for changed
 schemas plus replacement of `/steps/<step_id>`. If the step payload and
 schemas are unchanged, return `summarize_draft_workspace(workspace)`.
 
-- [ ] **Step 8: Write failing creation-parity tests**
+- [x] **Step 8: Write failing creation-parity tests**
 
 Add tests proving:
 
@@ -371,7 +371,7 @@ rejecting `input_map` and `input_bindings` whenever both arguments are
 supplied, including an explicit empty compatibility map, and proving existing
 map-only callers retain behavior.
 
-- [ ] **Step 9: Extend creation through the shared preflight**
+- [x] **Step 9: Extend creation through the shared preflight**
 
 Extend the method signature:
 
@@ -399,7 +399,7 @@ and validate the resulting typed list, or accept `input_bindings` directly.
 Use `_project_step_input_bindings` for both. Persist metadata only when not
 `None`; `retry=0` must survive.
 
-- [ ] **Step 10: Verify and commit Task 1**
+- [x] **Step 10: Verify and commit Task 1**
 
 Run:
 
@@ -446,7 +446,7 @@ git commit -m "feat: update capability-backed draft steps"
 - Extends the existing add-capability RPC with metadata and canonical
   `input_bindings`.
 
-- [ ] **Step 1: Write failing RPC model tests**
+- [x] **Step 1: Write failing RPC model tests**
 
 Add tests for:
 
@@ -470,11 +470,11 @@ timeout, and unknown fields. Extend `AddStepFromCapabilityParams` tests for
 metadata, canonical path/value order, and simultaneous `input_map` plus
 `input_bindings`.
 
-- [ ] **Step 2: Run RPC model tests red**
+- [x] **Step 2: Run RPC model tests red**
 
 Run the focused new tests. Expected: import/model failures.
 
-- [ ] **Step 3: Add RPC models**
+- [x] **Step 3: Add RPC models**
 
 Add:
 
@@ -498,7 +498,7 @@ timeout_seconds: int | None = Field(default=None, gt=0)
 
 Add a model validator rejecting both input forms when both were supplied.
 
-- [ ] **Step 4: Write failing endpoint and client tests**
+- [x] **Step 4: Write failing endpoint and client tests**
 
 Add a real ASGI test that creates a capability step, updates metadata and a
 literal input through:
@@ -524,7 +524,7 @@ and asserts the emitted nested object is exactly:
 No default `timeout_seconds` or `input` keys may appear. Extend add-capability
 client tests for canonical input order and metadata.
 
-- [ ] **Step 5: Implement public and RPC adapters**
+- [x] **Step 5: Implement public and RPC adapters**
 
 Add the update method to `WorkflowDraftSurface` and `WorkflowApi`, delegating to
 `draft_authoring`.
@@ -538,7 +538,7 @@ update.model_dump(mode="json", exclude_unset=True)
 Extend every add-capability seam with the Task 1 fields. Compatibility map-only
 calls remain valid.
 
-- [ ] **Step 6: Verify and commit Task 2**
+- [x] **Step 6: Verify and commit Task 2**
 
 Run:
 

src/wf_api/draft_authoring.py

@@ -553,7 +553,7 @@ class WorkflowDraftAuthoringApi:
         step_id: str,
         update: CapabilityStepUpdate,
     ) -> dict[str, Any]:
-        """Patch capability metadata and optional canonical inputs atomically."""
+        """Return a workspace summary or conflict after one atomic step patch."""
         checked = self._workspace_if_revision_matches(
             workspace_id=workspace_id,
             revision=revision,
@@ -584,11 +584,22 @@ class WorkflowDraftAuthoringApi:
             changes["input"] = update.input
 
         changed = current.model_copy(update=changes)
-        step_payload = changed.model_dump(
+        if projected is None:
-            mode="json",
+            step_payload = dict(deepcopy(step))
-            by_alias=True,
+            for field in ("desc", "retry", "timeout_seconds"):
-            exclude_none=True,
+                if field not in update.model_fields_set:
-        )
+                    continue
+                value = getattr(update, field)
+                if value is None:
+                    step_payload.pop(field, None)
+                else:
+                    step_payload[field] = value
+        else:
+            step_payload = changed.model_dump(
+                mode="json",
+                by_alias=True,
+                exclude_none=True,
+            )
         input_schema = (
             projected.input_schema
             if projected is not None
@@ -599,40 +610,31 @@ class WorkflowDraftAuthoringApi:
             if projected is not None
             else _draft_schema(workspace.draft, "state_schema")
         )
+        removed_metadata_key = any(
+            field in update.model_fields_set
+            and getattr(update, field) is None
+            and field in step
+            for field in ("desc", "retry", "timeout_seconds")
+        )
         if (
-            step == step_payload
+            current == changed
+            and not removed_metadata_key
             and workspace.draft.get("input_schema", {}) == input_schema
             and workspace.draft.get("state_schema", {}) == state_schema
         ):
             return summarize_draft_workspace(workspace)
 
-        if projected is None:
+        next_draft = deepcopy(workspace.draft)
-            next_draft = deepcopy(workspace.draft)
+        next_steps = next_draft.get("steps")
-            next_draft["steps"][step_id] = step_payload
+        if not isinstance(next_steps, dict):
-            return await self.drafts.replace_validated_draft_document(
+            raise ValueError("draft steps must be an object")
-                workspace_id=workspace_id,
+        next_steps[step_id] = step_payload
-                revision=revision,
+        next_draft["input_schema"] = input_schema
-                draft=next_draft,
+        next_draft["state_schema"] = state_schema
-            )
+        return await self.drafts._replace_validated_draft_document(
-
-        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,
             revision=revision,
-            patch=patch,
+            draft=next_draft,
         )
 
     async def set_workflow_output_bindings(
@@ -1073,6 +1075,17 @@ class WorkflowDraftAuthoringApi:
             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")
+        metadata = {
+            field: value
+            for field, value in (
+                ("desc", desc),
+                ("retry", retry),
+                ("timeout_seconds", timeout_seconds),
+            )
+            if value is not None
+        }
+        if metadata:
+            CapabilityStepUpdate.model_validate(metadata)
 
         spec = self.context.specs.get_qualified_spec(capability_name)
         output_schema = (
@@ -1160,6 +1173,11 @@ class WorkflowDraftAuthoringApi:
             step_payload["retry"] = retry
         if timeout_seconds is not None:
             step_payload["timeout_seconds"] = timeout_seconds
+        step_payload = DraftUseStep.model_validate(step_payload).model_dump(
+            mode="json",
+            by_alias=True,
+            exclude_none=True,
+        )
 
         patch: list[dict[str, Any]] = [
             {

src/wf_api/drafts.py

@@ -19,8 +19,8 @@ from wf_artifacts import (
 from wf_artifacts import (
     patch_draft_workspace as patch_draft_workspace_record,
 )
-from wf_artifacts import (
+from wf_artifacts.draft_workspaces.api import (
-    replace_validated_draft_document as replace_validated_draft_document_record,
+    _replace_validated_draft_document as replace_validated_draft_document_record,
 )
 from wf_core.models.schemas import NodeDef
 from wf_core.models.steps import (
@@ -292,7 +292,7 @@ class WorkflowDraftApi:
             node_defs_for_draft=self._node_defs_for_draft,
         )
 
-    async def replace_validated_draft_document(
+    async def _replace_validated_draft_document(
         self,
         *,
         workspace_id: str,

src/wf_api/service.py

@@ -448,6 +448,7 @@ class WorkflowApi:
         step_id: str,
         update: CapabilityStepUpdate,
     ) -> dict[str, Any]:
+        """Return the updated workspace summary or a revision-conflict payload."""
         return await self.draft_authoring.update_capability_step(
             workspace_id=workspace_id,
             revision=revision,

src/wf_artifacts/__init__.py

@@ -12,7 +12,6 @@ from .draft_workspaces import (
     ensure_workspace_id,
     get_draft_workspace,
     patch_draft_workspace,
-    replace_validated_draft_document,
     summarize_draft_workspace,
 )
 from .drafts import (
@@ -92,7 +91,6 @@ __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

@@ -2,7 +2,6 @@ from .api import (
     create_draft_workspace,
     get_draft_workspace,
     patch_draft_workspace,
-    replace_validated_draft_document,
 )
 from .models import (
     WorkflowDraftWorkspace,
@@ -24,6 +23,5 @@ __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,7 +102,7 @@ def patch_draft_workspace(
     return summarize_draft_workspace(next_workspace)
 
 
-def replace_validated_draft_document(
+def _replace_validated_draft_document(
     store: DraftWorkspaceStore,
     *,
     workspace_id: str,
@@ -118,13 +118,13 @@ def replace_validated_draft_document(
     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")
+    WorkflowDraft.model_validate(draft)
-    if canonical_draft == workspace.draft:
+    if draft == workspace.draft:
         return summarize_draft_workspace(workspace)
     next_workspace = workspace.model_copy(
         update={
             "revision": workspace.revision + 1,
-            "draft": canonical_draft,
+            "draft": draft,
             "updated_at_epoch_ms": _now_ms(),
         }
     )

src/wf_transport_rpc_http/client/drafts.py

@@ -23,6 +23,7 @@ class RpcDraftClientMixin:
         workspace_id: str,
         include_draft: bool = False,
     ) -> dict[str, Any]:
+        """Return the remote workspace summary or revision-conflict payload."""
         return await self._call(
             "workflow.draft_workspaces.get",
             {"workspace_id": workspace_id, "include_draft": include_draft},

tests/wf_api/test_drafts_service.py

@@ -98,7 +98,7 @@ async def test_update_capability_step_changes_metadata_and_inputs_atomically(
     assert step["use"] == "demo.personal.echo_tool"
     assert step["desc"] == "New description"
     assert step["retry"] == 0
-    assert step["timeout_seconds"] is None
+    assert "timeout_seconds" not in step
     assert step["input"] == [{"value": "fixed", "target": "text"}]
     assert step["output"] == [{"source": "echoed", "target": "state.echoed"}]
     assert inspected["draft"]["routes"]["echo"] == {"ok": "__end__"}
@@ -128,7 +128,15 @@ async def test_update_capability_step_preserves_omitted_fields_and_exact_noop(
         workspace_id="echo",
         revision=1,
         step_id="echo",
-        update=CapabilityStepUpdate(retry=2),
+        update=CapabilityStepUpdate(
+            retry=2,
+            input=[
+                InputPathBinding(
+                    path=GraphSourcePath.input("text"),
+                    target=LocalPath.of("text"),
+                )
+            ],
+        ),
     )
     second = await authoring.update_capability_step(
         workspace_id="echo",
@@ -144,13 +152,13 @@ async def test_update_capability_step_preserves_omitted_fields_and_exact_noop(
     assert first["revision"] == 1
     assert second["revision"] == 2
     step = inspected["draft"]["steps"]["echo"]
-    assert step["desc"] is None
+    assert "desc" not in step
     assert step["retry"] == 2
     assert step["timeout_seconds"] == 15
 
 
 @pytest.mark.asyncio
-async def test_update_capability_step_clearing_absent_metadata_is_exact_noop(
+async def test_update_capability_step_removes_stored_null_metadata(
     tmp_path: Path,
 ) -> None:
     draft_api, _service, authoring = _draft_api(
@@ -158,6 +166,11 @@ async def test_update_capability_step_clearing_absent_metadata_is_exact_noop(
         register_echo=True,
     )
     await draft_api.create_draft_workspace(workspace_id="echo", draft=_echo_draft())
+    before = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
+    assert before["draft"]["steps"]["echo"]["retry"] is None
 
     result = await authoring.update_capability_step(
         workspace_id="echo",
@@ -165,8 +178,13 @@ async def test_update_capability_step_clearing_absent_metadata_is_exact_noop(
         step_id="echo",
         update=CapabilityStepUpdate(retry=None),
     )
+    after = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
 
-    assert result["revision"] == 1
+    assert result["revision"] == 2
+    assert "retry" not in after["draft"]["steps"]["echo"]
 
 
 @pytest.mark.asyncio
@@ -198,6 +216,46 @@ async def test_update_capability_step_metadata_does_not_resolve_capability(
     assert result["revision"] == 2
 
 
+@pytest.mark.asyncio
+async def test_update_capability_step_metadata_preserves_invalid_draft_shape(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "update_invalid_draft_shape"),
+        register_echo=True,
+    )
+    draft = _echo_draft()
+    draft["routes"]["echo"]["ok"] = "missing"
+    await draft_api.create_draft_workspace(workspace_id="echo", draft=draft)
+    before = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
+
+    result = await authoring.update_capability_step(
+        workspace_id="echo",
+        revision=1,
+        step_id="echo",
+        update=CapabilityStepUpdate(desc="Explain the invalid draft"),
+    )
+    after = await draft_api.get_draft_workspace(
+        workspace_id="echo",
+        include_draft=True,
+    )
+
+    assert result["revision"] == 2
+    assert after["draft"]["steps"]["echo"]["desc"] == "Explain the invalid draft"
+    assert (
+        after["draft"]["steps"]["echo"]["input"]
+        == before["draft"]["steps"]["echo"]["input"]
+    )
+    assert (
+        after["draft"]["steps"]["echo"]["output"]
+        == before["draft"]["steps"]["echo"]["output"]
+    )
+    assert after["draft"]["routes"] == before["draft"]["routes"]
+
+
 @pytest.mark.asyncio
 async def test_update_capability_step_stale_revision_wins_over_semantic_errors(
     tmp_path: Path,
@@ -345,6 +403,54 @@ async def test_add_step_from_capability_rejects_both_input_forms(
         )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("desc", "retry", "timeout_seconds"),
+    [
+        ("", None, None),
+        (None, -1, None),
+        (None, None, 0),
+    ],
+)
+async def test_add_step_from_capability_rejects_invalid_metadata_atomically(
+    tmp_path: Path,
+    desc: str | None,
+    retry: int | None,
+    timeout_seconds: int | None,
+) -> None:
+    workspace_id = f"invalid_add_metadata_{desc}_{retry}_{timeout_seconds}"
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / workspace_id),
+        register_echo=True,
+    )
+    await draft_api.create_draft_workspace(
+        workspace_id=workspace_id,
+        draft=_echo_draft(),
+    )
+    before = await draft_api.get_draft_workspace(
+        workspace_id=workspace_id,
+        include_draft=True,
+    )
+
+    with pytest.raises(ValidationError):
+        await authoring.add_step_from_capability(
+            workspace_id=workspace_id,
+            revision=1,
+            step_id="other",
+            capability_name="demo.personal.echo_tool",
+            routes={"ok": "__end__"},
+            desc=desc,
+            retry=retry,
+            timeout_seconds=timeout_seconds,
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id=workspace_id,
+        include_draft=True,
+    )
+    assert after == before
+
+
 def _echo_draft() -> dict[str, Any]:
     return {
         "name": "echo",