fix: reject ambiguous output map merges

src/wf_api/drafts.py

@@ -489,8 +489,18 @@ class WorkflowDraftApi:
         merge: bool = False,
     ) -> dict[str, Any]:
         if merge:
+            workspace = self._workspace_if_revision_matches(
+                workspace_id=workspace_id,
+                revision=revision,
+            )
+            if isinstance(workspace, dict):
+                return workspace
+            step = _draft_step(workspace.draft, step_id)
             output_map = {
-                **self._step_output_map(workspace_id=workspace_id, step_id=step_id),
+                **_require_lossless_step_output_map_round_trip(
+                    step.get("output", []),
+                    step_id=step_id,
+                ),
                 **output_map,
             }
         return await self.patch_draft_workspace(
@@ -514,12 +524,39 @@ class WorkflowDraftApi:
         merge: bool = False,
     ) -> dict[str, Any]:
         output_bindings: list[dict[str, Any]]
+        workspace: WorkflowDraftWorkspace | None = None
         if merge:
-            workspace = self._draft_store().get_workspace(workspace_id)
+            checked_workspace = self._workspace_if_revision_matches(
+                workspace_id=workspace_id,
+                revision=revision,
+            )
+            if isinstance(checked_workspace, dict):
+                return checked_workspace
+            workspace = checked_workspace
             remaining = dict(output_map)
             output_bindings = []
             output_payload = workspace.draft.get("output")
             if isinstance(output_payload, list):
+                ambiguous = next(
+                    (
+                        source
+                        for source in output_map
+                        if sum(
+                            1
+                            for binding in output_payload
+                            if isinstance(binding, dict)
+                            and binding.get("path") == source
+                        )
+                        > 1
+                    ),
+                    None,
+                )
+                if ambiguous is not None:
+                    raise ValueError(
+                        f"workflow output source {ambiguous!r} has multiple "
+                        "bindings and cannot be updated through a compatibility "
+                        "map; replace the complete canonical binding list instead"
+                    )
                 for binding in output_payload:
                     if not isinstance(binding, dict):
                         continue
@@ -545,7 +582,8 @@ class WorkflowDraftApi:
                 {"path": source, "target": target}
                 for source, target in output_map.items()
             ]
-        workspace = self._draft_store().get_workspace(workspace_id)
+        if workspace is None:
+            workspace = self._draft_store().get_workspace(workspace_id)
         output_schema = self._workflow_output_schema_for_bindings(
             draft=workspace.draft,
             output_bindings=output_bindings,
@@ -627,22 +665,6 @@ class WorkflowDraftApi:
                 projected = updated
         return projected if changed else output_schema
 
-    def _step_input_maps(
-        self,
-        *,
-        workspace_id: str,
-        step_id: str,
-    ) -> tuple[dict[str, str], dict[str, Any]]:
-        workspace = self._draft_store().get_workspace(workspace_id)
-        step = _draft_step(workspace.draft, step_id)
-        return _input_maps_from_payload(step.get("input", []))
-
-    def _step_output_map(self, *, workspace_id: str, step_id: str) -> dict[str, str]:
-        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],
     source_path: str,
@@ -766,6 +788,22 @@ def _output_map_from_payload(payload: Any) -> dict[str, str]:
     return output_map
 
 
+def _require_lossless_step_output_map_round_trip(
+    payload: object,
+    *,
+    step_id: str,
+) -> dict[str, str]:
+    """Return a compatibility map only when it reproduces the output list."""
+    output_map = _output_map_from_payload(payload)
+    rebuilt = _draft_output_bindings_payload(output_map)
+    if rebuilt != payload:
+        raise ValueError(
+            f"step {step_id!r} outputs cannot be safely merged through a "
+            "compatibility map; replace the complete canonical binding list instead"
+        )
+    return output_map
+
+
 def _path_text(value: Any, *, expected_root: str | None = None) -> str:
     """Return compact dotted text for stored structural path JSON."""
     if isinstance(value, str):

tests/wf_api/test_drafts_service.py

@@ -1404,6 +1404,84 @@ async def test_step_input_map_merge_checks_revision_before_lossless_preflight(
     assert after == before
 
 
+@pytest.mark.asyncio
+async def test_step_output_map_merge_rejects_canonical_source_fan_out(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "output_merge_fan_out")
+    api, _service, _authoring = _draft_api(artifact_store)
+    draft = _echo_draft()
+    draft["state_schema"] = {
+        "fields": {
+            "echoed": {"type": "string"},
+            "other": {"type": "string"},
+        }
+    }
+    draft["steps"]["echo"]["output"] = [
+        {"source": "echoed", "target": "state.echoed"},
+        {"source": "echoed", "target": "state.other"},
+    ]
+    await api.create_draft_workspace(workspace_id="echo_ws", draft=draft)
+    before = await api.get_draft_workspace(
+        workspace_id="echo_ws",
+        include_draft=True,
+    )
+
+    with pytest.raises(ValueError, match="complete canonical binding list"):
+        await api.set_step_output_map(
+            workspace_id="echo_ws",
+            revision=1,
+            step_id="echo",
+            output_map={"extra": "state.extra"},
+            merge=True,
+        )
+
+    after = await api.get_draft_workspace(
+        workspace_id="echo_ws",
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_step_output_map_merge_checks_revision_before_lossless_preflight(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "output_merge_stale")
+    api, _service, _authoring = _draft_api(artifact_store)
+    draft = _echo_draft()
+    draft["steps"]["echo"]["output"] = [
+        {"source": "echoed", "target": "state.echoed"},
+        {"source": "echoed", "target": "state.other"},
+    ]
+    await api.create_draft_workspace(workspace_id="echo_ws", draft=draft)
+    await api.set_draft_name(
+        workspace_id="echo_ws",
+        revision=1,
+        name="echo_v2",
+    )
+    before = await api.get_draft_workspace(
+        workspace_id="echo_ws",
+        include_draft=True,
+    )
+
+    result = await api.set_step_output_map(
+        workspace_id="echo_ws",
+        revision=1,
+        step_id="echo",
+        output_map={"extra": "state.extra"},
+        merge=True,
+    )
+
+    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_validate_draft_workspace_refreshes_status(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_validate_workspace")
@@ -4938,6 +5016,102 @@ async def test_set_workflow_output_map_merges_top_level_output(tmp_path: Path) -
     ]
 
 
+@pytest.mark.asyncio
+async def test_workflow_output_map_merge_rejects_requested_fan_out_source(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "workflow_merge_fan_out")
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
+    draft = {
+        **_echo_draft(),
+        "output_schema": {
+            "type": "object",
+            "properties": {
+                "first": {"type": "string"},
+                "second": {"type": "string"},
+                "kind": {"type": "string"},
+            },
+        },
+        "output": [
+            {"path": "state.echoed", "target": "first"},
+            {"path": "state.echoed", "target": "second"},
+            {"value": "markdown", "target": "kind"},
+        ],
+    }
+    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 bindings"):
+        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 after == before
+
+
+@pytest.mark.asyncio
+async def test_workflow_output_map_merge_preserves_unrequested_fan_out_and_literals(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        tmp_path / "workflow_merge_unrequested_fan_out"
+    )
+    api, _service, _authoring = _draft_api(artifact_store, register_echo=True)
+    draft = {
+        **_echo_draft(),
+        "state_schema": {
+            "fields": {
+                "echoed": {"type": "string"},
+                "other": {"type": "string"},
+            }
+        },
+        "output_schema": {
+            "type": "object",
+            "properties": {
+                "first": {"type": "string"},
+                "second": {"type": "string"},
+                "kind": {"type": "string"},
+                "other": {"type": "string"},
+            },
+        },
+        "output": [
+            {"path": "state.echoed", "target": "first"},
+            {"path": "state.echoed", "target": "second"},
+            {"value": "markdown", "target": "kind"},
+        ],
+    }
+    await api.create_draft_workspace(workspace_id="report", draft=draft)
+
+    result = await api.set_workflow_output_map(
+        workspace_id="report",
+        revision=1,
+        output_map={"state.other": "other"},
+        merge=True,
+    )
+
+    fetched = await api.get_draft_workspace(
+        workspace_id="report",
+        include_draft=True,
+    )
+    assert result["revision"] == 2
+    assert fetched["draft"]["output"] == [
+        {"path": "state.echoed", "target": "first"},
+        {"path": "state.echoed", "target": "second"},
+        {"value": "markdown", "target": "kind"},
+        {"path": "state.other", "target": "other"},
+    ]
+
+
 @pytest.mark.asyncio
 async def test_set_workflow_output_map_projects_missing_output_schema(
     tmp_path: Path,