diff --git a/src/specify_cli/workflows/overlays/merge.py b/src/specify_cli/workflows/overlays/merge.py index 73a9c9f3e..3b11cd1e0 100644 --- a/src/specify_cli/workflows/overlays/merge.py +++ b/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): - sources.pop(removed_id, None) + if isinstance(removed_step, dict): + _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] - 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: - sources[composed.step_id] = composed.source - + # + # Inserts must be applied *before* the winning replace: if the replace + # changes the step ID, ``find_step`` can no longer locate the original + # anchor and the inserts would raise. 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 diff --git a/tests/workflows/test_overlay_merge.py b/tests/workflows/test_overlay_merge.py index 01be81bf4..9eb0db3b1 100644 --- a/tests/workflows/test_overlay_merge.py +++ b/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.""" diff --git a/tests/workflows/test_resolver_integration.py b/tests/workflows/test_resolver_integration.py index b7ee052ef..91f4b14e6 100644 --- a/tests/workflows/test_resolver_integration.py +++ b/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