fix: reject lossy step input map merges

src/wf_api/draft_authoring.py

@@ -212,23 +212,10 @@ class WorkflowDraftAuthoringApi:
         revision: int,
     ) -> WorkflowDraftWorkspace | dict[str, Any]:
         """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
-        return {
-            **summarize_draft_workspace(workspace),
-            "status": "conflict",
-            "diagnostics": [
-                {
-                    "code": "revision_conflict",
-                    "path": "revision",
-                    "message": (
-                        f"workspace {workspace.id!r} is at revision "
-                        f"{workspace.revision}, not {revision}"
-                    ),
-                }
-            ],
-        }
+        return self.drafts._workspace_if_revision_matches(
+            workspace_id=workspace_id,
+            revision=revision,
+        )
 
     def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
         try:

src/wf_api/drafts.py

@@ -8,9 +8,11 @@ from jsonschema import Draft202012Validator, SchemaError
 
 from wf_artifacts import (
     DraftWorkspaceStore,
+    WorkflowDraftWorkspace,
     compile_workflow_draft,
     patch_workflow_draft,
     replace_validated_draft_document,
+    summarize_draft_workspace,
     validate_workflow_draft,
 )
 from wf_artifacts import (
@@ -98,6 +100,31 @@ class WorkflowDraftApi:
             raise KeyError("draft workspace store is not configured")
         return self.context.draft_workspace_store
 
+    def _workspace_if_revision_matches(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+    ) -> WorkflowDraftWorkspace | dict[str, Any]:
+        """Load a workspace or return its canonical revision-conflict payload."""
+        workspace = self._draft_store().get_workspace(workspace_id)
+        if workspace.revision == revision:
+            return workspace
+        return {
+            **summarize_draft_workspace(workspace),
+            "status": "conflict",
+            "diagnostics": [
+                {
+                    "code": "revision_conflict",
+                    "path": "revision",
+                    "message": (
+                        f"workspace {workspace.id!r} is at revision "
+                        f"{workspace.revision}, not {revision}"
+                    ),
+                }
+            ],
+        }
+
     def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
         try:
             spec = self.context.specs.get_qualified_spec(qualified_name)
@@ -426,9 +453,18 @@ class WorkflowDraftApi:
     ) -> dict[str, Any]:
         input_values: dict[str, Any] = {}
         if merge:
-            existing_map, input_values = self._step_input_maps(
+            workspace = self._workspace_if_revision_matches(
                 workspace_id=workspace_id,
-                step_id=step_id,
+                revision=revision,
+            )
+            if isinstance(workspace, dict):
+                return workspace
+            step = _draft_step(workspace.draft, step_id)
+            existing_map, input_values = (
+                _require_lossless_step_input_map_round_trip(
+                    step.get("input", []),
+                    step_id=step_id,
+                )
             )
             input_map = {**existing_map, **input_map}
         return await self.patch_draft_workspace(
@@ -684,6 +720,22 @@ def _input_maps_from_payload(
     return input_map, input_values
 
 
+def _require_lossless_step_input_map_round_trip(
+    payload: object,
+    *,
+    step_id: str,
+) -> tuple[dict[str, str], dict[str, Any]]:
+    """Return compatibility maps only when they reproduce the binding list."""
+    input_map, input_values = _input_maps_from_payload(payload)
+    rebuilt = _draft_input_bindings_payload(input_map, input_values)
+    if rebuilt != payload:
+        raise ValueError(
+            f"step {step_id!r} inputs cannot be safely merged through a "
+            "compatibility map; replace the complete canonical binding list instead"
+        )
+    return input_map, input_values
+
+
 def _input_map_from_payload(payload: Any) -> dict[str, str]:
     """Read stored canonical input bindings back into a source -> local field map."""
     input_map: dict[str, str] = {}

tests/wf_api/test_drafts_service.py

@@ -1299,6 +1299,111 @@ async def test_step_map_helpers_merge_with_existing_bindings(tmp_path: Path) ->
     ]
 
 
+@pytest.mark.asyncio
+async def test_step_input_map_merge_rejects_canonical_source_fan_out(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "input_merge_fan_out")
+    api, _service, _authoring = _draft_api(artifact_store)
+    draft = _echo_draft()
+    draft["steps"]["echo"]["input"] = [
+        {"path": "input.text", "target": "message"},
+        {"path": "input.text", "target": "audit.message"},
+    ]
+    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_input_map(
+            workspace_id="echo_ws",
+            revision=1,
+            step_id="echo",
+            input_map={"input.extra": "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_input_map_merge_rejects_path_before_literal(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "input_merge_order")
+    api, _service, _authoring = _draft_api(artifact_store)
+    draft = _echo_draft()
+    draft["steps"]["echo"]["input"] = [
+        {"path": "input.text", "target": "message"},
+        {"value": "markdown", "target": "format"},
+    ]
+    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_input_map(
+            workspace_id="echo_ws",
+            revision=1,
+            step_id="echo",
+            input_map={"input.extra": "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_input_map_merge_checks_revision_before_lossless_preflight(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "input_merge_stale")
+    api, _service, _authoring = _draft_api(artifact_store)
+    draft = _echo_draft()
+    draft["steps"]["echo"]["input"] = [
+        {"path": "input.text", "target": "message"},
+        {"path": "input.text", "target": "audit.message"},
+    ]
+    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_input_map(
+        workspace_id="echo_ws",
+        revision=1,
+        step_id="echo",
+        input_map={"input.extra": "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")