fix: preserve canonical bindings in focused binds

docs/historical/superpowers/plans/2026-07-29-safe-compatibility-merges.md

@@ -1,5 +1,7 @@
 # Safe Compatibility Merges Implementation Plan
 
+**Status:** Completed
+
 > **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:** Prevent compatibility-map merges from silently changing canonical step and workflow-output bindings.
@@ -31,7 +33,7 @@
 - Produces: `_require_lossless_step_input_map_round_trip(payload, *, step_id)`
 - Preserves: `WorkflowDraftApi.set_step_input_map`
 
-- [ ] **Step 1: Write failing tests for lossless and lossy input merges**
+- [x] **Step 1: Write failing tests for lossless and lossy input merges**
 
 Add focused tests beside
 `test_step_map_helpers_merge_with_existing_bindings`:
@@ -93,7 +95,7 @@ representable case.
 Add a stale-revision test where the draft contains fan-out but the request uses
 an old revision. Assert a conflict payload, not the lossless-merge `ValueError`.
 
-- [ ] **Step 2: Run the focused tests and verify RED**
+- [x] **Step 2: Run the focused tests and verify RED**
 
 Run:
 
@@ -104,7 +106,7 @@ uv run pytest tests/wf_api/test_drafts_service.py -k "step_input_map_merge" -q
 Expected: the fan-out and interleaving tests fail because the merge succeeds or
 rewrites the list; the stale-revision test fails if preflight runs first.
 
-- [ ] **Step 3: Add the revision preflight and exact round-trip guard**
+- [x] **Step 3: Add the revision preflight and exact round-trip guard**
 
 In `WorkflowDraftApi`, add:
 
@@ -153,7 +155,7 @@ Update `set_step_input_map` so `merge=True`:
 
 Do not run the guard for replacement mode (`merge=False`).
 
-- [ ] **Step 4: Run focused tests and verify GREEN**
+- [x] **Step 4: Run focused tests and verify GREEN**
 
 Run:
 
@@ -163,7 +165,7 @@ uv run pytest tests/wf_api/test_drafts_service.py -k "step_input_map_merge or st
 
 Expected: all selected tests pass.
 
-- [ ] **Step 5: Commit Task 1**
+- [x] **Step 5: Commit Task 1**
 
 ```bash
 git add src/wf_api/drafts.py src/wf_api/draft_authoring.py tests/wf_api/test_drafts_service.py
@@ -181,7 +183,7 @@ git commit -m "fix: reject lossy step input map merges"
 - Preserves: `WorkflowDraftApi.set_step_output_map`
 - Preserves: `WorkflowDraftApi.set_workflow_output_map`
 
-- [ ] **Step 1: Write failing step-output fan-out tests**
+- [x] **Step 1: Write failing step-output fan-out tests**
 
 Create a draft whose step output contains:
 
@@ -211,7 +213,7 @@ full workspace and revision remain unchanged.
 Add a stale-revision variant and retain the existing unique-source merge test as
 the positive case.
 
-- [ ] **Step 2: Write failing workflow-output ambiguity tests**
+- [x] **Step 2: Write failing workflow-output ambiguity tests**
 
 Create workflow output bindings:
 
@@ -229,7 +231,7 @@ Add two tests:
 2. merging `{"state.other": "other"}` succeeds while preserving both
    `state.title` bindings, the literal, and their order.
 
-- [ ] **Step 3: Run focused tests and verify RED**
+- [x] **Step 3: Run focused tests and verify RED**
 
 Run:
 
@@ -240,7 +242,7 @@ uv run pytest tests/wf_api/test_drafts_service.py -k "step_output_map_merge or w
 Expected: fan-out step output is collapsed and the ambiguous workflow-output
 source updates multiple bindings.
 
-- [ ] **Step 4: Implement minimal guards**
+- [x] **Step 4: Implement minimal guards**
 
 Add:
 
@@ -291,7 +293,7 @@ if ambiguous is not None:
 
 Leave unrequested duplicate sources and literal records untouched.
 
-- [ ] **Step 5: Run focused and API regression tests**
+- [x] **Step 5: Run focused and API regression tests**
 
 Run:
 
@@ -301,7 +303,7 @@ uv run pytest tests/wf_api/test_drafts_service.py -k "map_helpers or set_step_in
 
 Expected: all selected tests pass.
 
-- [ ] **Step 6: Commit Task 2**
+- [x] **Step 6: Commit Task 2**
 
 ```bash
 git add src/wf_api/drafts.py tests/wf_api/test_drafts_service.py
@@ -319,7 +321,7 @@ git commit -m "fix: reject ambiguous output map merges"
 - Consumes: compatibility merge errors from Tasks 1-2
 - Produces: CLI help that describes rejection rather than acceptable loss
 
-- [ ] **Step 1: Write the failing CLI help assertion**
+- [x] **Step 1: Write the failing CLI help assertion**
 
 Update `test_wf_draft_map_help_explains_replace_merge_and_validate` to require:
 
@@ -332,7 +334,7 @@ assert "ambiguous fan-out sources" in workflow_output_help
 Remove assertions that describe compatibility merge as merely
 `potentially lossy`.
 
-- [ ] **Step 2: Run the CLI help test and verify RED**
+- [x] **Step 2: Run the CLI help test and verify RED**
 
 Run:
 
@@ -342,7 +344,7 @@ uv run pytest tests/wf_cli/test_app.py::test_wf_draft_map_help_explains_replace_
 
 Expected: FAIL because current help still says `potentially lossy`.
 
-- [ ] **Step 3: Update command help and issue state**
+- [x] **Step 3: Update command help and issue state**
 
 Revise the `set-input`, `set-output`, and `set-workflow-output` docstrings and
 `--merge` help text. State that merge:
@@ -355,7 +357,7 @@ Revise the `set-input`, `set-output`, and `set-workflow-output` docstrings and
 Mark the `ISSUES.md` compatibility step map item complete and summarize the
 guarded behavior without claiming the map representation gained fan-out.
 
-- [ ] **Step 4: Run focused verification**
+- [x] **Step 4: Run focused verification**
 
 Run:
 
@@ -369,9 +371,18 @@ git diff --check
 Expected: tests pass, Ruff reports no issues, basedpyright reports zero errors,
 and `git diff --check` emits no errors.
 
-- [ ] **Step 5: Commit Task 3**
+- [x] **Step 5: Commit Task 3**
 
 ```bash
 git add src/wf_cli/commands/drafts.py tests/wf_cli/test_app.py ISSUES.md
 git commit -m "docs: explain safe compatibility merges"
 ```
+
+## Post-Review Hardening
+
+The final review found that `bind_draft` still reconstructed canonical input
+and output lists through compatibility maps. The completed slice therefore
+also updates focused binds directly on typed canonical lists, preserving
+unrelated fan-out, literals, and ordering while rejecting ambiguous requested
+sources before mutation. Remote CLI regressions cover canonical-replacement
+guidance for step inputs, step outputs, and workflow outputs.

docs/superpowers/plans/2026-07-29-draft-document-transfer.md

@@ -484,7 +484,7 @@ git commit -m "feat: import draft documents"
 **Files:**
 - Modify: `skills/wf-cli/SKILL.md`
 - Modify: `docs/current_roadmap.md`
-- Move: `docs/superpowers/plans/2026-07-29-safe-compatibility-merges.md`
+- Reference: `docs/historical/superpowers/plans/2026-07-29-safe-compatibility-merges.md`
 - Move: `docs/superpowers/plans/2026-07-29-draft-document-transfer.md`
 
 **Interfaces:**
@@ -506,8 +506,8 @@ stored with diagnostics for repair.
 
 - [ ] **Step 2: Update roadmap and archive completed plans**
 
-Add one completed roadmap item linking to both historical plan paths. Move both
+Add one completed roadmap item linking to both historical plan paths. The safe
-plans under:
+compatibility plan is already archived; move the draft-transfer plan under:
 
 ```text
 docs/historical/superpowers/plans/

src/wf_api/draft_authoring.py

@@ -57,7 +57,6 @@ from .drafts import (
     WorkflowDraftApi,
     _draft_input_maps,
     _draft_output_map,
-    _input_map_from_payload,
 )
 from .operation_context import WorkflowOperationContext
 from .schema_projection import (
@@ -87,6 +86,119 @@ def _draft_schema(draft: Mapping[str, Any], key: str) -> dict[str, Any]:
     return deepcopy(value)
 
 
+def _upsert_input_path_binding(
+    payload: object,
+    *,
+    binding: InputPathBinding,
+    step_id: str,
+) -> list[dict[str, Any]]:
+    """Update one graph source without lowering unrelated canonical bindings."""
+    bindings = TypeAdapter(list[InputBinding]).validate_python(payload)
+    matching = [
+        index
+        for index, existing in enumerate(bindings)
+        if isinstance(existing, InputPathBinding) and existing.path == binding.path
+    ]
+    if len(matching) > 1:
+        raise ValueError(
+            f"step {step_id!r} source {str(binding.path)!r} has multiple input "
+            "bindings; replace the complete canonical binding list instead"
+        )
+    if matching:
+        bindings[matching[0]] = binding
+    else:
+        bindings.append(binding)
+    return [existing.model_dump(mode="json") for existing in bindings]
+
+
+def _upsert_step_output_binding(
+    payload: object,
+    *,
+    binding: OutputBinding,
+    step_id: str,
+) -> tuple[list[dict[str, Any]], str | None]:
+    """Return updated canonical outputs and the source's previous state target."""
+    bindings = TypeAdapter(list[OutputBinding]).validate_python(payload)
+    matching = [
+        index
+        for index, existing in enumerate(bindings)
+        if existing.source == binding.source
+    ]
+    if len(matching) > 1:
+        raise ValueError(
+            f"step {step_id!r} source {str(binding.source)!r} has multiple output "
+            "bindings; replace the complete canonical binding list instead"
+        )
+    previous_target = str(bindings[matching[0]].target) if matching else None
+    if matching:
+        bindings[matching[0]] = binding
+    else:
+        bindings.append(binding)
+    return (
+        [existing.model_dump(mode="json") for existing in bindings],
+        previous_target,
+    )
+
+
+def _rebind_workflow_output(
+    payload: object,
+    *,
+    previous_state_path: str | None,
+    state_path: str,
+    output_target: str,
+) -> list[dict[str, Any]]:
+    """Rebind one public target without guessing across source fan-out.
+
+    A previous state source and the requested public target can identify two
+    different records. Both are replaced because the bind operation transfers
+    ownership of that local output to the new state-to-public projection. The
+    replacement occupies the earliest removed position so unrelated canonical
+    bindings retain their relative order.
+    """
+    bindings = TypeAdapter(list[InputBinding]).validate_python(payload)
+    previous_matches = [
+        index
+        for index, binding in enumerate(bindings)
+        if (
+            previous_state_path is not None
+            and isinstance(binding, InputPathBinding)
+            and str(binding.path) == previous_state_path
+        )
+    ]
+    if len(previous_matches) > 1:
+        raise ValueError(
+            f"state source {previous_state_path!r} has multiple public output "
+            "bindings; replace the complete canonical binding list instead"
+        )
+    target_matches = [
+        index
+        for index, binding in enumerate(bindings)
+        if str(binding.target) == output_target
+    ]
+    if len(target_matches) > 1:
+        raise ValueError(
+            f"public output target {output_target!r} has multiple bindings; replace "
+            "the complete canonical binding list instead"
+        )
+
+    replaced_indices = set(previous_matches) | set(target_matches)
+    replacement = InputPathBinding(
+        path=GraphSourcePath.parse(state_path),
+        target=LocalPath.parse(output_target),
+    )
+    if not replaced_indices:
+        bindings.append(replacement)
+    else:
+        insert_at = min(replaced_indices)
+        bindings = [
+            binding
+            for index, binding in enumerate(bindings)
+            if index not in replaced_indices
+        ]
+        bindings.insert(insert_at, replacement)
+    return [binding.model_dump(mode="json") for binding in bindings]
+
+
 def _overlapping_input_binding_targets_error(
     bindings: Sequence[InputBinding],
 ) -> ValueError:
@@ -870,7 +982,6 @@ class WorkflowDraftAuthoringApi:
             target_root, target_parts = _graph_parts(target_path)
 
         if target_root == "local" and source_root in {"input", "state"}:
-            local_path = format_toml_path_segments(target_parts)
             input_schema = (
                 spec.input_schema_contract or spec.input_model.model_json_schema()
             )
@@ -887,10 +998,14 @@ class WorkflowDraftAuthoringApi:
                     source_parts=target_parts,
                     target_parts=source_parts,
                 )
-            input_map = {
+            input_bindings = _upsert_input_path_binding(
-                **_input_map_from_payload(step.get("input", [])),
+                step.get("input", []),
-                source_path: local_path,
+                binding=InputPathBinding(
-            }
+                    path=GraphSourcePath.parse(source_path),
+                    target=LocalPath(target_parts),
+                ),
+                step_id=step_id,
+            )
             return await self.drafts.patch_draft_workspace(
                 workspace_id=workspace_id,
                 revision=revision,
@@ -899,7 +1014,7 @@ class WorkflowDraftAuthoringApi:
                     {
                         "op": "replace",
                         "path": f"/steps/{escape_json_pointer(step_id)}/input",
-                        "value": input_bindings_payload(input_map, {}),
+                        "value": input_bindings,
                     },
                 ],
             )
@@ -934,36 +1049,18 @@ class WorkflowDraftAuthoringApi:
                 allow_existing_equivalent=True,
             )
 
-            current_output_map = self.drafts._step_output_map(
+            step_output_bindings, previous_state_path = _upsert_step_output_binding(
-                workspace_id=workspace_id, step_id=step_id
+                step.get("output", []),
+                binding=OutputBinding.model_validate(
+                    {"source": local_path, "target": state_path_str}
+                ),
+                step_id=step_id,
             )
-            previous_state_path = current_output_map.get(local_path)
+            output_bindings = _rebind_workflow_output(
-            output_map = {
+                workspace.draft.get("output", []),
-                **current_output_map,
+                previous_state_path=previous_state_path,
-                local_path: state_path_str,
+                state_path=state_path_str,
-            }
+                output_target=output_target_str,
-
-            existing_output = workspace.draft.get("output")
-            if isinstance(existing_output, list):
-                output_bindings = [
-                    b
-                    for b in existing_output
-                    if not (
-                        isinstance(b, dict)
-                        and (
-                            b.get("target") == output_target_str
-                            or b.get("path") == state_path_str
-                            or (
-                                previous_state_path is not None
-                                and b.get("path") == previous_state_path
-                            )
-                        )
-                    )
-                ]
-            else:
-                output_bindings = []
-            output_bindings.append(
-                {"path": state_path_str, "target": output_target_str}
             )
 
             return await self.drafts.patch_draft_workspace(
@@ -983,7 +1080,7 @@ class WorkflowDraftAuthoringApi:
                     {
                         "op": "replace",
                         "path": f"/steps/{escape_json_pointer(step_id)}/output",
-                        "value": output_bindings_payload(output_map),
+                        "value": step_output_bindings,
                     },
                     {"op": "replace", "path": "/output", "value": output_bindings},
                 ],
@@ -1004,12 +1101,13 @@ class WorkflowDraftAuthoringApi:
                 target_parts=target_parts,
                 allow_existing_equivalent=True,
             )
-            output_map = {
+            output_bindings, _previous_state_path = _upsert_step_output_binding(
-                **self.drafts._step_output_map(
+                step.get("output", []),
-                    workspace_id=workspace_id, step_id=step_id
+                binding=OutputBinding.model_validate(
+                    {"source": local_path, "target": target_path}
                 ),
-                local_path: target_path,
+                step_id=step_id,
-            }
+            )
             return await self.drafts.patch_draft_workspace(
                 workspace_id=workspace_id,
                 revision=revision,
@@ -1018,7 +1116,7 @@ class WorkflowDraftAuthoringApi:
                     {
                         "op": "replace",
                         "path": f"/steps/{escape_json_pointer(step_id)}/output",
-                        "value": output_bindings_payload(output_map),
+                        "value": output_bindings,
                     },
                 ],
             )

src/wf_api/drafts.py

@@ -663,12 +663,6 @@ class WorkflowDraftApi:
                 projected = updated
         return projected if changed else output_schema
 
-    def _step_output_map(self, *, workspace_id: str, step_id: str) -> dict[str, str]:
-        """Read one step's outputs for legacy focused binding helpers."""
-        workspace = self._draft_store().get_workspace(workspace_id)
-        step = _draft_step(workspace.draft, step_id)
-        return _output_map_from_payload(step.get("output", []))
-
 
 def _workflow_source_schema(
     draft: Mapping[str, Any],

tests/wf_api/test_drafts_service.py

@@ -1299,6 +1299,40 @@ async def test_step_map_helpers_merge_with_existing_bindings(tmp_path: Path) ->
     ]
 
 
+@pytest.mark.asyncio
+async def test_step_input_map_merge_preserves_literal_prefix(
+    tmp_path: Path,
+) -> None:
+    api, _service, _authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "input_merge_literal_prefix")
+    )
+    draft = _echo_draft()
+    draft["steps"]["echo"]["input"] = [
+        {"value": "seed", "target": "fallback"},
+        {"path": "input.text", "target": "text"},
+    ]
+    await api.create_draft_workspace(workspace_id="echo_ws", draft=draft)
+
+    result = await api.set_step_input_map(
+        workspace_id="echo_ws",
+        revision=1,
+        step_id="echo",
+        input_map={"input.extra": "extra"},
+        merge=True,
+    )
+
+    fetched = await api.get_draft_workspace(
+        workspace_id="echo_ws",
+        include_draft=True,
+    )
+    assert result["revision"] == 2
+    assert fetched["draft"]["steps"]["echo"]["input"] == [
+        {"value": "seed", "target": "fallback"},
+        {"path": "input.text", "target": "text"},
+        {"path": "input.extra", "target": "extra"},
+    ]
+
+
 @pytest.mark.asyncio
 async def test_step_input_map_merge_rejects_canonical_source_fan_out(
     tmp_path: Path,
@@ -1799,6 +1833,45 @@ async def test_bind_draft_workflow_input_to_step_input_can_repeat(
     ]
 
 
+@pytest.mark.asyncio
+async def test_bind_draft_input_preserves_literals_and_unrelated_fan_out(
+    tmp_path: Path,
+) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "bind_preserves_canonical_inputs"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _structured_report)
+    draft = _structured_report_draft()
+    draft["state_schema"] = {
+        "type": "object",
+        "properties": {"shared": {"type": "string"}},
+    }
+    draft["steps"]["report"]["input"] = [
+        {"path": "state.shared", "target": "request.title"},
+        {"path": "state.shared", "target": "request.format"},
+        {"value": "audit", "target": "audit.note"},
+    ]
+    await api.create_draft_workspace(workspace_id="report", draft=draft)
+
+    result = await authoring.bind_draft(
+        workspace_id="report",
+        revision=1,
+        step_id="report",
+        source_path="input.body",
+        target_path="local.request.body",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="report", include_draft=True)
+
+    assert result["revision"] == 2
+    assert workspace["draft"]["steps"]["report"]["input"] == [
+        {"path": "state.shared", "target": "request.title"},
+        {"path": "state.shared", "target": "request.format"},
+        {"value": "audit", "target": "audit.note"},
+        {"path": "input.body", "target": "request.body"},
+    ]
+
+
 @pytest.mark.asyncio
 async def test_bind_draft_workflow_state_to_step_input_reuses_existing_schema(
     tmp_path: Path,
@@ -2028,6 +2101,51 @@ async def test_bind_draft_lowers_nested_local_output_to_public_output(
     )
 
 
+@pytest.mark.asyncio
+async def test_bind_draft_output_preserves_unrelated_source_fan_out(
+    tmp_path: Path,
+) -> None:
+    api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "bind_preserves_canonical_outputs"),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _nested_report)
+    draft = _nested_report_draft()
+    draft["state_schema"] = {
+        "type": "object",
+        "properties": {
+            "report": {
+                "type": "object",
+                "properties": {
+                    "title": {"type": "string"},
+                    "audit_title": {"type": "string"},
+                },
+            }
+        },
+    }
+    draft["steps"]["render"]["output"] = [
+        {"source": "report.title", "target": "state.report.title"},
+        {"source": "report.title", "target": "state.report.audit_title"},
+    ]
+    await api.create_draft_workspace(workspace_id="nested", draft=draft)
+
+    result = await authoring.bind_draft(
+        workspace_id="nested",
+        revision=1,
+        step_id="render",
+        source_path="local.report.markdown",
+        target_path="state.report.markdown",
+    )
+    workspace = await api.get_draft_workspace(workspace_id="nested", include_draft=True)
+
+    assert result["revision"] == 2
+    assert workspace["draft"]["steps"]["render"]["output"] == [
+        {"source": "report.title", "target": "state.report.title"},
+        {"source": "report.title", "target": "state.report.audit_title"},
+        {"source": "report.markdown", "target": "state.report.markdown"},
+    ]
+
+
 @pytest.mark.asyncio
 async def test_bind_draft_rebinds_nested_local_public_output(tmp_path: Path) -> None:
     api, service, authoring = _draft_api(
@@ -5059,6 +5177,48 @@ async def test_workflow_output_map_merge_rejects_requested_fan_out_source(
     assert after == before
 
 
+@pytest.mark.asyncio
+async def test_workflow_output_map_merge_checks_revision_before_ambiguity(
+    tmp_path: Path,
+) -> None:
+    api, _service, _authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "workflow_merge_stale_fan_out"),
+        register_echo=True,
+    )
+    draft = {
+        **_echo_draft(),
+        "output": [
+            {"path": "state.echoed", "target": "first"},
+            {"path": "state.echoed", "target": "second"},
+        ],
+    }
+    await api.create_draft_workspace(workspace_id="report", draft=draft)
+    await api.set_draft_name(
+        workspace_id="report",
+        revision=1,
+        name="report_v2",
+    )
+    before = await api.get_draft_workspace(
+        workspace_id="report",
+        include_draft=True,
+    )
+
+    result = await api.set_workflow_output_map(
+        workspace_id="report",
+        revision=1,
+        output_map={"state.echoed": "renamed"},
+        merge=True,
+    )
+
+    after = await api.get_draft_workspace(
+        workspace_id="report",
+        include_draft=True,
+    )
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+    assert after == before
+
+
 @pytest.mark.asyncio
 async def test_workflow_output_map_merge_preserves_unrequested_fan_out_and_literals(
     tmp_path: Path,
@@ -5621,6 +5781,47 @@ async def test_bind_draft_replaces_previous_public_output_for_local_field(
     ]
 
 
+@pytest.mark.asyncio
+async def test_bind_draft_rejects_ambiguous_previous_public_output_fan_out(
+    tmp_path: Path,
+) -> None:
+    api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "bind_ambiguous_public_output"),
+        register_echo=True,
+    )
+    draft = _echo_draft()
+    draft["state_schema"] = {
+        "type": "object",
+        "properties": {"old": {"type": "string"}},
+    }
+    draft["output_schema"] = {
+        "type": "object",
+        "properties": {
+            "first": {"type": "string"},
+            "second": {"type": "string"},
+        },
+    }
+    draft["steps"]["echo"]["output"] = [{"source": "echoed", "target": "state.old"}]
+    draft["output"] = [
+        {"path": "state.old", "target": "first"},
+        {"path": "state.old", "target": "second"},
+    ]
+    await api.create_draft_workspace(workspace_id="report", draft=draft)
+    before = await api.get_draft_workspace(workspace_id="report", include_draft=True)
+
+    with pytest.raises(ValueError, match="multiple public output bindings"):
+        await authoring.bind_draft(
+            workspace_id="report",
+            revision=1,
+            step_id="echo",
+            source_path="local.echoed",
+            target_path="output.echoed",
+        )
+
+    after = await api.get_draft_workspace(workspace_id="report", include_draft=True)
+    assert after == before
+
+
 @pytest.mark.asyncio
 async def test_remove_draft_binding_rejects_non_object_entries(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bad_bindings")

tests/wf_cli/test_remote_target.py

@@ -1397,6 +1397,83 @@ def test_wf_draft_set_workflow_output_merge_uses_compatibility_rpc_target(
     ]
 
 
+def test_wf_draft_set_workflow_output_merge_reports_canonical_replacement(
+    monkeypatch, tmp_path
+) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    _patch_rpc_client_to_server(monkeypatch, server)
+    rpc_methods: list[str] = []
+    original_call = RpcClientTransport._call
+
+    async def recording_call(
+        self: RpcClientTransport, method: str, params: dict[str, Any]
+    ) -> dict[str, Any]:
+        rpc_methods.append(method)
+        return await original_call(self, method, params)
+
+    monkeypatch.setattr(RpcClientTransport, "_call", recording_call)
+    config_path = tmp_path / "wf.json"
+    config_path.write_text('{"version": 1}', encoding="utf-8")
+    bindings_path = tmp_path / "workflow-output-bindings.json"
+    bindings_path.write_text(
+        json.dumps(
+            [
+                {"path": "state.value", "target": "first"},
+                {"path": "state.value", "target": "second"},
+            ]
+        ),
+        encoding="utf-8",
+    )
+    runner = CliRunner()
+    base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
+    created = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "create",
+            "report",
+            "--capability",
+            "wf.std.constant",
+        ],
+    )
+    replaced = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "set-workflow-output",
+            "report",
+            "--revision",
+            "1",
+            "--bindings-file",
+            str(bindings_path),
+        ],
+    )
+    rpc_methods.clear()
+
+    merged = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "set-workflow-output",
+            "report",
+            "--revision",
+            "2",
+            "--map",
+            "state.value=renamed",
+            "--merge",
+        ],
+    )
+
+    assert created.exit_code == 0, created.output
+    assert replaced.exit_code == 0, replaced.output
+    assert merged.exit_code == 1
+    assert "complete canonical binding list" in merged.output
+    assert rpc_methods == ["workflow.draft_workspaces.set_workflow_output_map"]
+
+
 def test_wf_draft_remove_route_uses_rpc_target(monkeypatch, tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     _patch_rpc_client_to_server(monkeypatch, server)
@@ -1636,6 +1713,28 @@ def test_wf_draft_set_input_replaces_canonical_bindings_over_rpc(
     ]
     assert rpc_methods.count("workflow.draft_workspaces.set_step_input_bindings") == 1
 
+    rpc_methods.clear()
+    merged = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "set-input",
+            "filter_ws",
+            "--revision",
+            "2",
+            "--step",
+            "call",
+            "--map",
+            "input.other=other",
+            "--merge",
+        ],
+    )
+
+    assert merged.exit_code == 1
+    assert "complete canonical binding list" in merged.output
+    assert rpc_methods == ["workflow.draft_workspaces.set_step_input_map"]
+
 
 def test_wf_draft_set_output_replaces_canonical_bindings_over_rpc(
     monkeypatch, tmp_path