fix(workflows): address Copilot review findings in merge engine

- Apply inserts before winning replace to prevent anchor-not-found errors
  when replace changes step ID (r3594064604)
- Track attribution recursively for nested steps in composite inserts/replaces
  so workflow resolve attributes all child steps correctly (r3594064638)
- Add regression tests for both fixes

Refs: PR #3557 review discussion

Assisted-by: GitHub Copilot (model: qwen3.7-plus, autonomous)

src/specify_cli/workflows/overlays/merge.py

@@ -102,6 +102,61 @@ def _init_sources_recursively(
                     _init_sources_recursively(case_steps, sources)
 
 
+def _record_sources_recursively(
+    step: dict[str, Any],
+    source: str,
+    sources: dict[str, str],
+) -> None:
+    """Record *source* for a step and all its nested child steps.
+
+    Traverses ``then``, ``else``, ``steps``, ``default``, and ``cases.*``
+    so that ``workflow resolve`` attributes every step inside a composite
+    insert or replacement to the correct overlay layer.
+    """
+    step_id = step.get("id")
+    if isinstance(step_id, str):
+        sources[step_id] = source
+    for key in _NESTED_LIST_KEYS:
+        nested = step.get(key)
+        if isinstance(nested, list):
+            for child in nested:
+                if isinstance(child, dict):
+                    _record_sources_recursively(child, source, sources)
+    cases = step.get("cases")
+    if isinstance(cases, dict):
+        for case_steps in cases.values():
+            if isinstance(case_steps, list):
+                for child in case_steps:
+                    if isinstance(child, dict):
+                        _record_sources_recursively(child, source, sources)
+
+
+def _remove_sources_recursively(
+    step: dict[str, Any],
+    sources: dict[str, str],
+) -> None:
+    """Remove source entries for a step and all its nested child steps.
+
+    Traverses the same nesting keys as ``_record_sources_recursively``.
+    """
+    step_id = step.get("id")
+    if isinstance(step_id, str):
+        sources.pop(step_id, None)
+    for key in _NESTED_LIST_KEYS:
+        nested = step.get(key)
+        if isinstance(nested, list):
+            for child in nested:
+                if isinstance(child, dict):
+                    _remove_sources_recursively(child, sources)
+    cases = step.get("cases")
+    if isinstance(cases, dict):
+        for case_steps in cases.values():
+            if isinstance(case_steps, list):
+                for child in case_steps:
+                    if isinstance(child, dict):
+                        _remove_sources_recursively(child, sources)
+
+
 def apply_edit(
     steps: list[dict[str, Any]],
     edit: OverlayEdit,
@@ -207,30 +262,33 @@ def merge_steps(
             if location is not None:
                 parent_list, index = location
                 removed_step = parent_list[index]
-                removed_id = removed_step.get("id") if isinstance(removed_step, dict) else None
                 del parent_list[index]
-                if isinstance(removed_id, str):
+                if isinstance(removed_step, dict):
-                    sources.pop(removed_id, None)
+                    _remove_sources_recursively(removed_step, sources)
             continue
 
         # For replace/insert_*, the anchor survives. Only the highest-priority
         # replace is applied; lower-priority replaces on the same anchor are
         # skipped. Inserts are applied in merge order.
-        if winning_edit.operation == "replace":
+        #
-            winning_layer, _ = edits[-1]
+        # Inserts must be applied *before* the winning replace: if the replace
-            steps, composed, replaced_id = apply_edit(steps, winning_edit, winning_layer.source)
+        # changes the step ID, ``find_step`` can no longer locate the original
-            if isinstance(replaced_id, str):
+        # anchor and the inserts would raise.
-                sources.pop(replaced_id, None)
-            if composed is not None:
-                sources[composed.step_id] = composed.source
-
         for layer, edit in edits:
             if edit.operation in ("insert_after", "insert_before"):
                 steps, composed, replaced_id = apply_edit(steps, edit, layer.source)
                 if replaced_id is not None:
                     sources.pop(replaced_id, None)
                 if composed is not None:
-                    sources[composed.step_id] = composed.source
+                    _record_sources_recursively(edit.step, composed.source, sources)
+
+        if winning_edit.operation == "replace":
+            winning_layer, _ = edits[-1]
+            steps, composed, replaced_id = apply_edit(steps, winning_edit, winning_layer.source)
+            if isinstance(replaced_id, str):
+                sources.pop(replaced_id, None)
+            if composed is not None:
+                _record_sources_recursively(winning_edit.step, composed.source, sources)
 
     attribution = _build_attribution(steps, sources)
     return steps, attribution

tests/workflows/test_overlay_merge.py

@@ -364,6 +364,60 @@ class TestMergeSteps:
             for composed in attribution
         )
 
+    def test_merge_steps_insert_after_then_replace_same_anchor_id_change(self):
+        """Inserts must be applied before the winning replace so the anchor still exists.
+
+        Regression: when a replace changes the step ID, applying it before inserts
+        causes ``find_step`` to fail on the now-gone original anchor.
+        """
+        base = [_step("build")]
+        low = Overlay(
+            id="low",
+            extends="wf",
+            priority=5,
+            edits=[OverlayEdit("insert_after", "build", _step("test"))],
+        )
+        high = Overlay(
+            id="high",
+            extends="wf",
+            priority=10,
+            edits=[OverlayEdit("replace", "build", _step("compile"))],
+        )
+        steps, attribution = merge_steps(
+            base, [_layer(low, "project:low"), _layer(high, "project:high")]
+        )
+        # The insert should land after the original anchor position, then the
+        # anchor is replaced.  Final order: ["compile", "test"].
+        assert [s["id"] for s in steps] == ["compile", "test"]
+        assert attribution == [
+            ComposedStep("compile", "project:high"),
+            ComposedStep("test", "project:low"),
+        ]
+
+    def test_merge_steps_insert_before_then_replace_same_anchor_id_change(self):
+        """Same as above but with insert_before — anchor must still be findable."""
+        base = [_step("build")]
+        low = Overlay(
+            id="low",
+            extends="wf",
+            priority=5,
+            edits=[OverlayEdit("insert_before", "build", _step("lint"))],
+        )
+        high = Overlay(
+            id="high",
+            extends="wf",
+            priority=10,
+            edits=[OverlayEdit("replace", "build", _step("compile"))],
+        )
+        steps, attribution = merge_steps(
+            base, [_layer(low, "project:low"), _layer(high, "project:high")]
+        )
+        assert [s["id"] for s in steps] == ["lint", "compile"]
+        assert attribution == [
+            ComposedStep("lint", "project:low"),
+            ComposedStep("compile", "project:high"),
+        ]
+
     def test_merge_steps_unknown_anchor_still_raises(self):
         base = [_step("a")]
         overlay = Overlay(
@@ -375,6 +429,130 @@ class TestMergeSteps:
         with pytest.raises(ValueError, match="Anchor 'missing' not found"):
             merge_steps(base, [_layer(overlay, "project:ov")])
 
+    # ── composite step attribution ───────────────────────────────────────
+
+    def test_merge_insert_composite_if_attribution(self):
+        """Nested then/else children of an inserted 'if' step get the overlay source."""
+        base = [_step("a")]
+        composite = {
+            "id": "if-1",
+            "type": "if",
+            "condition": "true",
+            "then": [_step("then-a")],
+            "else": [_step("else-b")],
+        }
+        overlay = Overlay(
+            id="ov", extends="wf", priority=10,
+            edits=[OverlayEdit("insert_after", "a", composite)],
+        )
+        _steps, attribution = merge_steps(
+            base, [_layer(overlay, "project:ov")]
+        )
+        assert attribution == [
+            ComposedStep("a", "base"),
+            ComposedStep("if-1", "project:ov"),
+            ComposedStep("then-a", "project:ov"),
+            ComposedStep("else-b", "project:ov"),
+        ]
+
+    def test_merge_insert_composite_switch_attribution(self):
+        """Nested cases/default children of an inserted 'switch' step get the overlay source."""
+        base = [_step("a")]
+        composite = {
+            "id": "switch-1",
+            "type": "switch",
+            "expression": "{{inputs.x}}",
+            "cases": {"one": [_step("case-one")], "two": [_step("case-two")]},
+            "default": [_step("default-z")],
+        }
+        overlay = Overlay(
+            id="ov", extends="wf", priority=10,
+            edits=[OverlayEdit("insert_before", "a", composite)],
+        )
+        _steps, attribution = merge_steps(
+            base, [_layer(overlay, "project:ov")]
+        )
+        assert attribution == [
+            ComposedStep("switch-1", "project:ov"),
+            ComposedStep("default-z", "project:ov"),
+            ComposedStep("case-one", "project:ov"),
+            ComposedStep("case-two", "project:ov"),
+            ComposedStep("a", "base"),
+        ]
+
+    def test_merge_replace_flat_with_composite_attribution(self):
+        """Replacing a flat step with a composite step attributes all nested children."""
+        base = [_step("a")]
+        composite = {
+            "id": "if-1",
+            "type": "if",
+            "condition": "true",
+            "then": [_step("inner-x"), _step("inner-y")],
+        }
+        overlay = Overlay(
+            id="ov", extends="wf", priority=10,
+            edits=[OverlayEdit("replace", "a", composite)],
+        )
+        _steps, attribution = merge_steps(
+            base, [_layer(overlay, "project:ov")]
+        )
+        assert attribution == [
+            ComposedStep("if-1", "project:ov"),
+            ComposedStep("inner-x", "project:ov"),
+            ComposedStep("inner-y", "project:ov"),
+        ]
+
+    def test_merge_remove_composite_step_cleans_nested_sources(self):
+        """Removing a composite step also cleans its nested children from sources."""
+        base = [
+            {
+                "id": "if-1",
+                "type": "if",
+                "condition": "true",
+                "then": [_step("then-a")],
+                "else": [_step("else-b")],
+            },
+            _step("a"),
+        ]
+        overlay = Overlay(
+            id="ov", extends="wf", priority=10,
+            edits=[OverlayEdit("remove", "if-1")],
+        )
+        steps, attribution = merge_steps(
+            base, [_layer(overlay, "project:ov")]
+        )
+        assert [s["id"] for s in steps] == ["a"]
+        assert attribution == [ComposedStep("a", "base")]
+
+    def test_merge_insert_deeply_nested_composite_attribution(self):
+        """Deep nesting (if inside while) gets the overlay source at every level."""
+        base = [_step("a")]
+        inner_if = {
+            "id": "inner-if",
+            "type": "if",
+            "condition": "true",
+            "then": [_step("deep-x")],
+        }
+        composite = {
+            "id": "while-1",
+            "type": "while",
+            "condition": "true",
+            "steps": [inner_if],
+        }
+        overlay = Overlay(
+            id="ov", extends="wf", priority=10,
+            edits=[OverlayEdit("insert_after", "a", composite)],
+        )
+        _steps, attribution = merge_steps(
+            base, [_layer(overlay, "project:ov")]
+        )
+        assert attribution == [
+            ComposedStep("a", "base"),
+            ComposedStep("while-1", "project:ov"),
+            ComposedStep("inner-if", "project:ov"),
+            ComposedStep("deep-x", "project:ov"),
+        ]
+
 
 class TestValidateEdits:
     """Edit validation against known base step IDs."""

tests/workflows/test_resolver_integration.py

@@ -532,6 +532,46 @@ class TestWorkflowResolver:
         with pytest.raises(ValueError, match="Symlinked overlay directories are not allowed"):
             resolver.resolve("wf")
 
+    def test_resolve_attribution_for_inserted_composite_step(self, project_dir):
+        """Inserted composite steps must attribute nested children to the overlay source."""
+        data = {
+            "schema_version": "1.0",
+            "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"},
+            "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}],
+        }
+        _write_workflow(project_dir, "wf", data)
+        _write_overlay(
+            project_dir,
+            "wf",
+            "ov1",
+            {
+                "id": "ov1",
+                "extends": "wf",
+                "priority": 10,
+                "edits": [
+                    {
+                        "operation": "insert_after",
+                        "anchor": "a",
+                        "step": {
+                            "id": "if-1",
+                            "type": "if",
+                            "condition": "true",
+                            "then": [{"id": "then-x", "type": "command", "command": "echo"}],
+                            "else": [{"id": "else-y", "type": "command", "command": "echo"}],
+                        },
+                    }
+                ],
+            },
+        )
+
+        resolver = WorkflowResolver(project_dir)
+        _definition, _layers, attribution = resolver.resolve_with_layers("wf")
+        sources = {c.step_id: c.source for c in attribution}
+        assert sources["a"] == "base"
+        assert sources["if-1"] == "project:ov1"
+        assert sources["then-x"] == "project:ov1"
+        assert sources["else-y"] == "project:ov1"
+
     def test_engine_load_workflow_uses_resolver(self, project_dir):
         from specify_cli.workflows.engine import WorkflowEngine