fix: gate capability draft edits on revision

src/wf_api/draft_authoring.py

@@ -107,7 +107,7 @@ class WorkflowDraftAuthoringApi:
         workspace_id: str,
         revision: int,
     ) -> WorkflowDraftWorkspace | dict[str, Any]:
-        """Load a workspace for no-op edits while still enforcing optimistic locks."""
+        """Load a workspace and enforce optimistic locking before semantic preflight."""
         workspace = self.drafts._draft_store().get_workspace(workspace_id)
         if workspace.revision == revision:
             return workspace
@@ -330,7 +330,13 @@ class WorkflowDraftAuthoringApi:
         target_path: str,
     ) -> dict[str, Any]:
         """Bind a graph path to/from one capability local field, projecting missing schema when needed."""
-        workspace = self.drafts._draft_store().get_workspace(workspace_id)
+        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:
@@ -535,7 +541,13 @@ class WorkflowDraftAuthoringApi:
         one revision so callers do not have to interleave add-step, route,
         input-map, state-schema, and output-map operations by hand.
         """
-        workspace = self.drafts._draft_store().get_workspace(workspace_id)
+        checked = self._workspace_if_revision_matches(
+            workspace_id=workspace_id,
+            revision=revision,
+        )
+        if isinstance(checked, dict):
+            return checked
+        workspace = checked
         steps = workspace.draft.get("steps")
         if not isinstance(steps, dict):
             raise ValueError("draft steps must be an object")

tests/wf_api/test_drafts_service.py

@@ -535,6 +535,38 @@ async def test_patch_draft_workspace_updates_revision(tmp_path: Path) -> None:
     assert patched["status"] == "valid"
 
 
+@pytest.mark.asyncio
+async def test_patch_draft_workspace_stale_revision_does_not_mutate(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        tmp_path / "drafts_patch_workspace_stale"
+    )
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(
+        workspace_id="echo_ws",
+        draft=_echo_draft(),
+    )
+    before = await api.get_draft_workspace(
+        workspace_id="echo_ws",
+        include_draft=True,
+    )
+
+    result = await api.patch_draft_workspace(
+        workspace_id="echo_ws",
+        revision=2,
+        patch=[{"op": "replace", "path": "/name", "value": "must_not_apply"}],
+    )
+
+    after = await api.get_draft_workspace(
+        workspace_id="echo_ws",
+        include_draft=True,
+    )
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+    assert after == before
+
+
 @pytest.mark.asyncio
 async def test_draft_workspace_patch_helpers_update_revision_and_bindings(
     tmp_path: Path,
@@ -1286,6 +1318,51 @@ async def test_add_step_stale_revision_wins_over_content_preflight(
     assert after == before
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["bind", "capability_add"])
+async def test_capability_aware_edits_stale_revision_wins_over_semantic_errors(
+    tmp_path: Path,
+    operation: str,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        tmp_path / f"draft_capability_stale_{operation}"
+    )
+    api, _service, authoring = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(
+        workspace_id="draft_ws",
+        draft=_echo_draft(),
+    )
+    before = await api.get_draft_workspace(
+        workspace_id="draft_ws",
+        include_draft=True,
+    )
+
+    if operation == "bind":
+        result = await authoring.bind_draft(
+            workspace_id="draft_ws",
+            revision=2,
+            step_id="missing",
+            source_path="input.text",
+            target_path="local.text",
+        )
+    else:
+        result = await authoring.add_step_from_capability(
+            workspace_id="draft_ws",
+            revision=2,
+            step_id="new_step",
+            capability_name="missing.connection.unknown_tool",
+        )
+
+    after = await api.get_draft_workspace(
+        workspace_id="draft_ws",
+        include_draft=True,
+    )
+    assert result["status"] == "conflict"
+    assert result["revision"] == before["revision"]
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+    assert after == before
+
+
 @pytest.mark.asyncio
 async def test_add_step_adds_missing_incoming_route_parent_atomically(
     tmp_path: Path,