feat: replace draft step output bindings

.superpowers/sdd/task-1-report.md

@@ -0,0 +1,86 @@
+# Task 1 Report: Atomic Capability-Aware Output Replacement
+
+## Changed Files
+
+- `src/wf_api/draft_authoring.py`
+  - Added stable-index overlap diagnostics for state targets.
+  - Added the atomic output-binding patch builder.
+  - Added `WorkflowDraftAuthoringApi.set_step_output_bindings` with revision-first
+    validation, capability source validation, schema projection, exact-equivalent
+    target reuse, no-op handling, and canonical ordered replacement.
+- `src/wf_api/surface.py`
+  - Added `set_step_output_bindings` to `WorkflowDraftSurface`, and therefore the
+    composed `WorkflowApiSurface`.
+- `src/wf_api/service.py`
+  - Added the `WorkflowApi` delegation method.
+- `tests/wf_api/test_drafts_service.py`
+  - Added canonical replacement and source fan-out coverage.
+  - Added nested and whole-payload schema projection coverage.
+  - Added exact-equivalent no-op, clear, overlap, duplicate, missing-source,
+    incompatible-target, missing-step, non-capability-step, stale-revision, and
+    no-mutation coverage.
+  - Added an explicit nested output schema contract matching the Task 1 brief.
+- `tests/core/test_atomic_state_patches.py`
+  - Added runtime source fan-out coverage asserting both state writes.
+
+## RED Evidence
+
+Command:
+
+```text
+uv run pytest tests/wf_api/test_drafts_service.py -q -k "step_output" --basetemp C:\\tmp\\pytest-task1-red
+```
+
+Result: 12 failed. Every failure reached the intended missing-feature error:
+`AttributeError: 'WorkflowApi' object has no attribute 'set_step_output_bindings'`.
+
+The first environment attempt could not start the local `uv` shim. A first
+rerun also found a missing `StatePath` test import; that test-only error was
+fixed before the canonical RED run above.
+
+## GREEN Evidence
+
+Focused API command:
+
+```text
+uv run pytest tests/wf_api/test_drafts_service.py -q -k "step_output" --basetemp C:\\tmp\\pytest-task1-green-api
+```
+
+Result: `12 passed in 9.51s`.
+
+Required combined command:
+
+```text
+uv run pytest tests/wf_api/test_drafts_service.py tests/core/test_atomic_state_patches.py -q --basetemp C:\\tmp\\pytest-task1-green-all
+```
+
+Result: `160 passed in 9.93s`.
+
+Additional verification:
+
+- `uv run ruff check` on all five Task 1 files: passed.
+- `uv run ruff format --check` on all five Task 1 files: passed.
+- `uv run basedpyright --level error` on all five Task 1 files: `0 errors, 0 warnings, 0 notes`.
+- `git diff --check`: passed.
+
+## Deviations
+
+- The test capability uses an explicit output schema contract rather than the
+  generated Pydantic schema so the assertions match the brief's exact nested
+  `title`/`markdown` contract and do not depend on generated metadata or `$ref`
+  names.
+- Pytest used `--basetemp C:\\tmp\\...` because the default system temp
+  cleanup failed with `PermissionError: [WinError 5]` in this environment.
+- RPC/MCP/CLI transport adapters were not changed; the brief limits Task 1 to
+  the Python authoring/runtime contract and explicitly lists only the five
+  implementation/test files.
+
+## Concerns
+
+Full repository `basedpyright --level error` remains red with four conformance
+errors in `src/wf_cli/context.py`, `tests/wf_api/test_surface_protocol.py`, and
+`tests/wf_transport_rpc_http/test_client.py`. These are downstream adapter
+typing failures because `RpcWorkflowApiClient` does not yet implement the new
+protocol method. The Task 1 scoped type check is clean; later transport work
+must add the corresponding RPC/client surface before the full type check can
+pass.

src/wf_api/draft_authoring.py

@@ -102,6 +102,24 @@ def _overlapping_input_targets_error(
     raise AssertionError("overlap error requested without overlapping targets")
 
 
+def _overlapping_output_targets_error(
+    bindings: Sequence[OutputBinding],
+) -> ValueError:
+    """Describe the first overlapping state-target pair with stable indexes."""
+    for left_index, left in enumerate(bindings):
+        for right_index in range(left_index + 1, len(bindings)):
+            right = bindings[right_index]
+            # StatePath is a separate typed path, so serialized state.* values
+            # provide the shared synthetic root expected by paths_overlap.
+            if paths_overlap(str(left.target), str(right.target)):
+                return ValueError(
+                    f"bindings[{left_index}].target {str(left.target)!r} "
+                    f"overlaps bindings[{right_index}].target "
+                    f"{str(right.target)!r}"
+                )
+    raise AssertionError("overlap error requested without overlapping targets")
+
+
 def _step_input_bindings_patch(
     *,
     workspace: WorkflowDraftWorkspace,
@@ -128,6 +146,27 @@ def _step_input_bindings_patch(
     return patch
 
 
+def _step_output_bindings_patch(
+    *,
+    workspace: WorkflowDraftWorkspace,
+    step_id: str,
+    bindings: list[dict[str, Any]],
+    state_schema: dict[str, Any],
+) -> list[dict[str, Any]]:
+    """Build one atomic patch for state schema and canonical step outputs."""
+    patch: list[dict[str, Any]] = []
+    if workspace.draft.get("state_schema", {}) != state_schema:
+        patch.append({"op": "replace", "path": "/state_schema", "value": state_schema})
+    patch.append(
+        {
+            "op": "replace",
+            "path": f"/steps/{escape_json_pointer(step_id)}/output",
+            "value": bindings,
+        }
+    )
+    return patch
+
+
 class WorkflowDraftAuthoringApi:
     """Capability-aware semantic edits over revisioned workflow drafts."""
 
@@ -459,6 +498,85 @@ class WorkflowDraftAuthoringApi:
             patch=patch,
         )
 
+    async def set_step_output_bindings(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        bindings: Sequence[OutputBinding],
+    ) -> dict[str, Any]:
+        """Replace one capability step's canonical output bindings 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} does not declare a capability use"
+            )
+        spec = self.context.specs.get_qualified_spec(capability_name)
+        capability_schema = (
+            spec.output_schema_contract or spec.output_model.model_json_schema()
+        )
+
+        targets = [str(binding.target) for binding in bindings]
+        if has_overlapping_paths(targets):
+            raise _overlapping_output_targets_error(bindings)
+
+        projected_state = _draft_schema(workspace.draft, "state_schema")
+        for index, binding in enumerate(bindings):
+            source_parts = binding.source.parts
+            try:
+                schema_fragment_at_path(
+                    capability_schema,
+                    source_parts,
+                    label="capability output schema",
+                )
+            except ValueError as exc:
+                raise ValueError(
+                    f"bindings[{index}].source {str(binding.source)!r} "
+                    f"is not declared by capability {capability_name!r}: {exc}"
+                ) from exc
+
+            target_parts = binding.target.parts
+            try:
+                projected_state = project_schema_path_to_schema_path(
+                    target_schema=projected_state,
+                    source_schema=capability_schema,
+                    source_parts=source_parts,
+                    target_parts=target_parts,
+                    allow_existing_equivalent=True,
+                )
+            except ValueError as exc:
+                raise ValueError(
+                    f"bindings[{index}].target {str(binding.target)!r} "
+                    f"cannot receive source {str(binding.source)!r}: {exc}"
+                ) from exc
+
+        payload = [binding.model_dump(mode="json") for binding in bindings]
+        if (
+            step.get("output", []) == payload
+            and workspace.draft.get("state_schema", {}) == projected_state
+        ):
+            return summarize_draft_workspace(workspace)
+
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=_step_output_bindings_patch(
+                workspace=workspace,
+                step_id=step_id,
+                bindings=payload,
+                state_schema=projected_state,
+            ),
+        )
+
     async def bind_draft(
         self,
         *,

src/wf_api/service.py

@@ -5,7 +5,7 @@ from typing import Any
 
 from wf_artifacts import ArtifactKind
 from wf_artifacts.drafts.models import DraftStep
-from wf_core.models.steps import InputBinding
+from wf_core.models.steps import InputBinding, OutputBinding
 
 from .artifacts import WorkflowArtifactApi
 from .capabilities import WorkflowCapabilityApi
@@ -424,6 +424,21 @@ class WorkflowApi:
             bindings=bindings,
         )
 
+    async def set_step_output_bindings(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        bindings: Sequence[OutputBinding],
+    ) -> dict[str, Any]:
+        return await self.draft_authoring.set_step_output_bindings(
+            workspace_id=workspace_id,
+            revision=revision,
+            step_id=step_id,
+            bindings=bindings,
+        )
+
     async def set_step_output_map(
         self,
         *,

src/wf_api/surface.py

@@ -5,7 +5,7 @@ from typing import Any, Protocol
 
 from wf_artifacts import ArtifactKind
 from wf_artifacts.drafts.models import DraftStep
-from wf_core.models.steps import InputBinding
+from wf_core.models.steps import InputBinding, OutputBinding
 
 from .draft_authoring import RouteSource
 from .runs import TraceRangeLike
@@ -147,6 +147,15 @@ class WorkflowDraftSurface(Protocol):
         bindings: Sequence[InputBinding],
     ) -> dict[str, Any]: ...
 
+    async def set_step_output_bindings(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        bindings: Sequence[OutputBinding],
+    ) -> dict[str, Any]: ...
+
     async def set_step_output_map(
         self,
         *,

tests/core/test_atomic_state_patches.py

@@ -182,6 +182,24 @@ def test_full_workflow_execution_writes_canonical_output_bindings() -> None:
     assert run.trace[0].state_changes["state.person.name"] == "Ada"
 
 
+def test_output_bindings_apply_one_source_to_multiple_state_targets() -> None:
+    workflow = _workflow()
+    state = {"person": {"name": "old"}}
+
+    apply_output_bindings(
+        workflow,
+        [
+            _binding("person.name", "state.person.name"),
+            _binding("person.name", "state.person.extra"),
+        ],
+        {"person": {"name": "Ada"}},
+        state,
+    )
+
+    assert state["person"]["name"] == "Ada"
+    assert state["person"]["extra"] == "Ada"
+
+
 def test_build_output_patch_does_not_mutate_until_commit() -> None:
     workflow = _workflow(fields={"person.name": StateField(type="string")})
     state = {"person": {"name": "old"}}

tests/wf_api/test_drafts_service.py

@@ -15,8 +15,8 @@ from wf_api.service import WorkflowApi
 from wf_artifacts import FileDraftWorkspaceStore, FileWorkflowArtifactStore
 from wf_artifacts.drafts.models import DraftStep
 from wf_authoring import node
-from wf_core.models.steps import InputPathBinding, InputValueBinding
+from wf_core.models.steps import InputPathBinding, InputValueBinding, OutputBinding
-from wf_core.paths import GraphSourcePath, LocalPath
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 from wf_mcp.broker import WfMcpService
 from wf_mcp.broker.service.workflow_operation_context import context_from_service
 from wf_mcp.models import ConnectionConfig
@@ -86,6 +86,7 @@ class _NestedReportInput(BaseModel):
 
 
 class _ReportOutputValue(BaseModel):
+    title: str
     markdown: str
 
 
@@ -96,7 +97,10 @@ class _NestedReportOutput(BaseModel):
 @node(name="nested_report", outcomes=("ok",))
 def _nested_report(payload: _NestedReportInput) -> _NestedReportOutput:
     return _NestedReportOutput(
-        report=_ReportOutputValue(markdown=f"# {payload.report.title}")
+        report=_ReportOutputValue(
+            title=payload.report.title,
+            markdown=f"# {payload.report.title}",
+        )
     )
 
 
@@ -207,6 +211,41 @@ async def _create_structured_binding_api(
     return draft_api, service, WorkflowApi(authoring.context)
 
 
+async def _create_nested_output_binding_api(
+    tmp_path: Path,
+    workspace_id: str,
+) -> tuple[WorkflowDraftApi, WfMcpService, WorkflowApi]:
+    draft_api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / workspace_id),
+        register_echo=True,
+    )
+    service.register_specs(
+        "demo.personal",
+        replace(
+            _nested_report,
+            output_schema_contract={
+                "type": "object",
+                "properties": {
+                    "report": {
+                        "type": "object",
+                        "properties": {
+                            "title": {"type": "string"},
+                            "markdown": {"type": "string"},
+                        },
+                        "required": ["title", "markdown"],
+                    }
+                },
+                "required": ["report"],
+            },
+        ),
+    )
+    await draft_api.create_draft_workspace(
+        workspace_id=workspace_id,
+        draft=_nested_report_draft(),
+    )
+    return draft_api, service, WorkflowApi(authoring.context)
+
+
 @pytest.mark.asyncio
 async def test_patch_draft_applies_json_patch(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch")
@@ -1506,6 +1545,395 @@ async def test_set_step_input_bindings_preserves_source_fan_out(tmp_path: Path)
     ]
 
 
+@pytest.mark.asyncio
+async def test_set_step_output_bindings_replaces_in_order_and_preserves_source_fan_out(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        "draft-output-bindings",
+    )
+
+    result = await api.set_step_output_bindings(
+        workspace_id="draft-output-bindings",
+        revision=1,
+        step_id="render",
+        bindings=[
+            OutputBinding(
+                source=LocalPath.parse("report.title"),
+                target=StatePath.parse("state.report.title"),
+            ),
+            OutputBinding(
+                source=LocalPath.parse("report.title"),
+                target=StatePath.parse("state.audit.title"),
+            ),
+        ],
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="draft-output-bindings",
+        include_draft=True,
+    )
+
+    assert result["revision"] == 2
+    assert inspected["draft"]["steps"]["render"]["output"] == [
+        {"source": "report.title", "target": "state.report.title"},
+        {"source": "report.title", "target": "state.audit.title"},
+    ]
+    state_schema = inspected["draft"]["state_schema"]
+    assert state_schema["properties"]["report"]["properties"]["title"] == {
+        "type": "string"
+    }
+    assert state_schema["properties"]["audit"]["properties"]["title"] == {
+        "type": "string"
+    }
+
+
+@pytest.mark.asyncio
+async def test_set_step_output_bindings_projects_whole_capability_payload(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        "whole-output-binding",
+    )
+
+    await api.set_step_output_bindings(
+        workspace_id="whole-output-binding",
+        revision=1,
+        step_id="render",
+        bindings=[
+            OutputBinding(
+                source=LocalPath.root(),
+                target=StatePath.parse("state.raw_result"),
+            )
+        ],
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="whole-output-binding",
+        include_draft=True,
+    )
+
+    raw_result_schema = inspected["draft"]["state_schema"]["properties"]["raw_result"]
+    assert raw_result_schema["type"] == "object"
+    assert raw_result_schema["properties"]["report"]["properties"]["title"] == {
+        "type": "string"
+    }
+
+
+@pytest.mark.asyncio
+async def test_set_step_output_bindings_accepts_exact_existing_target_and_is_noop(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        "equivalent-output-binding",
+    )
+    binding = OutputBinding(
+        source=LocalPath.parse("report.title"),
+        target=StatePath.parse("state.report.title"),
+    )
+    first = await api.set_step_output_bindings(
+        workspace_id="equivalent-output-binding",
+        revision=1,
+        step_id="render",
+        bindings=[binding],
+    )
+    second = await api.set_step_output_bindings(
+        workspace_id="equivalent-output-binding",
+        revision=first["revision"],
+        step_id="render",
+        bindings=[binding],
+    )
+
+    assert first["revision"] == 2
+    assert second["revision"] == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("bindings", "message"),
+    [
+        (
+            [
+                OutputBinding(
+                    source=LocalPath.parse("report.missing"),
+                    target=StatePath.parse("state.report.missing"),
+                )
+            ],
+            r"bindings\[0\]\.source 'report\.missing' is not declared",
+        ),
+        (
+            [
+                OutputBinding(
+                    source=LocalPath.parse("report.title"),
+                    target=StatePath.parse("state.report.title"),
+                ),
+                OutputBinding(
+                    source=LocalPath.parse("report.markdown"),
+                    target=StatePath.parse("state.report.title"),
+                ),
+            ],
+            r"bindings\[0\]\.target 'state\.report\.title' overlaps "
+            r"bindings\[1\]\.target 'state\.report\.title'",
+        ),
+        (
+            [
+                OutputBinding(
+                    source=LocalPath.parse("report.title"),
+                    target=StatePath.parse("state.report"),
+                ),
+                OutputBinding(
+                    source=LocalPath.parse("report.markdown"),
+                    target=StatePath.parse("state.report.title"),
+                ),
+            ],
+            r"bindings\[0\]\.target 'state\.report' overlaps "
+            r"bindings\[1\]\.target 'state\.report\.title'",
+        ),
+    ],
+)
+async def test_set_step_output_bindings_rejects_semantic_errors_without_mutation(
+    tmp_path: Path,
+    bindings: list[OutputBinding],
+    message: str,
+) -> None:
+    workspace_id = f"invalid_output_bindings_{len(message)}"
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        workspace_id,
+    )
+    before = await draft_api.get_draft_workspace(
+        workspace_id=workspace_id,
+        include_draft=True,
+    )
+
+    with pytest.raises(ValueError, match=message):
+        await api.set_step_output_bindings(
+            workspace_id=workspace_id,
+            revision=1,
+            step_id="render",
+            bindings=bindings,
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id=workspace_id,
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_set_step_output_bindings_rejects_incompatible_existing_target(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        "incompatible-output-binding",
+    )
+    await draft_api.patch_draft_workspace(
+        workspace_id="incompatible-output-binding",
+        revision=1,
+        patch=[
+            {
+                "op": "replace",
+                "path": "/state_schema",
+                "value": {
+                    "type": "object",
+                    "properties": {
+                        "report": {
+                            "type": "object",
+                            "properties": {"title": {"type": "integer"}},
+                        }
+                    },
+                },
+            }
+        ],
+    )
+    before = await draft_api.get_draft_workspace(
+        workspace_id="incompatible-output-binding",
+        include_draft=True,
+    )
+
+    with pytest.raises(
+        ValueError,
+        match=(
+            r"bindings\[0\]\.target 'state\.report\.title' cannot receive "
+            r"source 'report\.title'"
+        ),
+    ):
+        await api.set_step_output_bindings(
+            workspace_id="incompatible-output-binding",
+            revision=2,
+            step_id="render",
+            bindings=[
+                OutputBinding(
+                    source=LocalPath.parse("report.title"),
+                    target=StatePath.parse("state.report.title"),
+                )
+            ],
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id="incompatible-output-binding",
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_set_step_output_bindings_clears_outputs_without_removing_projection(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        "clear-output-bindings",
+    )
+    first = await api.set_step_output_bindings(
+        workspace_id="clear-output-bindings",
+        revision=1,
+        step_id="render",
+        bindings=[
+            OutputBinding(
+                source=LocalPath.parse("report.title"),
+                target=StatePath.parse("state.report.title"),
+            )
+        ],
+    )
+    cleared = await api.set_step_output_bindings(
+        workspace_id="clear-output-bindings",
+        revision=first["revision"],
+        step_id="render",
+        bindings=[],
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="clear-output-bindings",
+        include_draft=True,
+    )
+
+    assert cleared["revision"] == 3
+    assert inspected["draft"]["steps"]["render"]["output"] == []
+    assert (
+        inspected["draft"]["state_schema"]["properties"]["report"]["properties"][
+            "title"
+        ]["type"]
+        == "string"
+    )
+    assert set(
+        inspected["draft"]["state_schema"]["properties"]["report"]["properties"][
+            "title"
+        ]
+    ) == {"type"}
+
+
+@pytest.mark.asyncio
+async def test_set_step_output_bindings_rejects_missing_step_without_mutation(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        "missing-output-step",
+    )
+    before = await draft_api.get_draft_workspace(
+        workspace_id="missing-output-step",
+        include_draft=True,
+    )
+
+    with pytest.raises(KeyError, match="missing"):
+        await api.set_step_output_bindings(
+            workspace_id="missing-output-step",
+            revision=1,
+            step_id="missing",
+            bindings=[],
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id="missing-output-step",
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_set_step_output_bindings_rejects_non_capability_step_without_mutation(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        "non-capability-output-step",
+    )
+    await draft_api.patch_draft_workspace(
+        workspace_id="non-capability-output-step",
+        revision=1,
+        patch=[{"op": "replace", "path": "/steps/render", "value": {"join": {}}}],
+    )
+    before = await draft_api.get_draft_workspace(
+        workspace_id="non-capability-output-step",
+        include_draft=True,
+    )
+
+    with pytest.raises(ValueError, match="does not declare a capability use"):
+        await api.set_step_output_bindings(
+            workspace_id="non-capability-output-step",
+            revision=2,
+            step_id="render",
+            bindings=[],
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id="non-capability-output-step",
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "bindings",
+    [
+        [
+            OutputBinding(
+                source=LocalPath.parse("report.missing"),
+                target=StatePath.parse("state.report.missing"),
+            )
+        ],
+        [
+            OutputBinding(
+                source=LocalPath.parse("report.title"),
+                target=StatePath.parse("state.report"),
+            ),
+            OutputBinding(
+                source=LocalPath.parse("report.markdown"),
+                target=StatePath.parse("state.report.title"),
+            ),
+        ],
+    ],
+)
+async def test_set_step_output_bindings_stale_revision_precedes_semantic_errors(
+    tmp_path: Path,
+    bindings: list[OutputBinding],
+) -> None:
+    draft_api, _service, api = await _create_nested_output_binding_api(
+        tmp_path,
+        f"stale-output-binding-{len(bindings)}",
+    )
+    workspace_id = f"stale-output-binding-{len(bindings)}"
+
+    result = await api.set_step_output_bindings(
+        workspace_id=workspace_id,
+        revision=2,
+        step_id="render",
+        bindings=bindings,
+    )
+
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id=workspace_id,
+        include_draft=True,
+    )
+    assert inspected["revision"] == 1
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("bindings", "message"),