Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 49 additions & 6 deletions src/specify_cli/workflows/overlays/merge.py
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,38 @@ def _build_attribution(
return result


def _winning_fate_edit(
edits: list[tuple[OverlayLayer, OverlayEdit]],
) -> tuple[OverlayLayer, OverlayEdit] | None:
"""Return the edit that decides an anchor's fate.

Normally that is simply the last edit in merge order. The exception this
helper exists for: when the last edit is an ``insert_*``, the same overlay
may *also* have declared a ``replace`` on the anchor earlier in the file.
Declaration order inside one overlay is not a precedence signal -- priority
is a per-overlay property -- so the trailing insert must not cancel that
overlay's own replacement, which previously reverted the anchor to the base
step and discarded the replacement silently.

A ``replace`` leaves the anchor in place, so both edits can be honoured.
``remove`` is deliberately NOT rescued here: it destroys the anchor, so an
insert relative to it cannot also apply, and choosing between them is a
separate question. That combination keeps its existing behaviour.

Returns ``None`` only when there are no edits.
"""
if not edits:
return None
winning_layer, last_edit = edits[-1]
if last_edit.operation not in ("insert_after", "insert_before"):
return edits[-1]
replacement: tuple[OverlayLayer, OverlayEdit] | None = None
for layer, edit in edits:
if layer is winning_layer and edit.operation == "replace":
replacement = (layer, edit)
return replacement if replacement is not None else edits[-1]


def _traverse_and_apply(
steps: list[dict[str, Any]],
edits_by_anchor: dict[str, list[tuple[OverlayLayer, OverlayEdit]]],
Expand All @@ -255,7 +287,8 @@ def _traverse_and_apply(

step_id = step.get("id")
edits = edits_by_anchor.get(step_id, []) if isinstance(step_id, str) else []
winning_edit = edits[-1][1] if edits else None
fate = _winning_fate_edit(edits)
winning_edit = fate[1] if fate is not None else None

if winning_edit is not None and winning_edit.operation == "remove":
# Winning edit removes this step; ignore all other edits on this anchor.
Expand All @@ -274,7 +307,7 @@ def _traverse_and_apply(
result.append(new_step)

if winning_edit is not None and winning_edit.operation == "replace":
winning_layer = edits[-1][0]
winning_layer = fate[0]
new_step = copy.deepcopy(winning_edit.step)
_remove_sources_recursively(step, sources)
_record_sources_recursively(new_step, winning_layer.source, sources)
Expand Down Expand Up @@ -352,10 +385,20 @@ def merge_steps(
# the ancestor edit replaces or removes its subtree — those produce
# order-dependent results. Pure insert edits on an ancestor are safe because
# the ancestor step (and its descendants) remain intact.
anchor_winning_ops = {
anchor: anchor_edits[-1][1].operation
for anchor, anchor_edits in edits_by_anchor.items()
}
# Must agree with ``_traverse_and_apply``: use the same fate rule, or the
# conflict guard stops firing for a subtree that is in fact replaced.
# Every anchor stays in the mapping even when its fate is a pure insert --
# ``_check_anchor_conflicts`` reads the key set to find *descendant*
# anchors, so dropping insert-only anchors would stop conflicts being
# detected against them.
anchor_winning_ops = {}
for anchor, anchor_edits in edits_by_anchor.items():
anchor_fate = _winning_fate_edit(anchor_edits)
anchor_winning_ops[anchor] = (
anchor_fate[1].operation
if anchor_fate is not None
else anchor_edits[-1][1].operation
)
anchor_conflicts = _check_anchor_conflicts(anchor_winning_ops, base_steps)
if anchor_conflicts:
raise ValueError(
Expand Down
98 changes: 98 additions & 0 deletions tests/workflows/test_overlay_merge.py
Original file line number Diff line number Diff line change
Expand Up @@ -733,3 +733,101 @@ def test_replace_with_reused_id_does_not_affect_original(self):
assert sources.get("b") == "project:ov", (
f"expected 'project:ov' but got {sources.get('b')!r}"
)


class TestMergeStepsSameOverlayFateEdits:
"""An overlay's replace/remove must survive its own trailing insert.

`_traverse_and_apply` decided an anchor's fate with `edits[-1]`, which
treats declaration order *inside one overlay file* as a precedence signal.
Priority is a per-overlay property, so two edits from the same overlay have
no priority relation to break — yet a trailing `insert_after` reverted the
anchor to the base step and discarded that overlay's own `replace`.
"""

def test_replace_then_insert_after_same_overlay_keeps_replacement(self):
base = [_step("implement"), _step("tail")]
overlay = Overlay(
id="ov",
extends="wf",
priority=10,
edits=[
OverlayEdit(
"replace", "implement",
{**_step("implement"), "command": "custom.impl"},
),
OverlayEdit("insert_after", "implement", _step("lint")),
],
)

steps, _ = merge_steps(base, [_layer(overlay, "project:ov")])

by_id = {s["id"]: s for s in steps}
assert by_id["implement"]["command"] == "custom.impl"
assert "lint" in by_id

def test_replace_and_insert_order_inside_one_overlay_is_irrelevant(self):
"""Both declaration orders must produce the same result."""
base = [_step("implement"), _step("tail")]
replace_edit = OverlayEdit(
"replace", "implement", {**_step("implement"), "command": "custom.impl"}
)
insert_edit = OverlayEdit("insert_after", "implement", _step("lint"))

first, _ = merge_steps(
base,
[_layer(Overlay(id="ov", extends="wf", priority=10, edits=[replace_edit, insert_edit]), "project:ov")],
)
second, _ = merge_steps(
base,
[_layer(Overlay(id="ov", extends="wf", priority=10, edits=[insert_edit, replace_edit]), "project:ov")],
)

assert [(s["id"], s.get("command")) for s in first] == [
(s["id"], s.get("command")) for s in second
]

def test_remove_then_insert_after_same_overlay_is_unchanged(self):
"""`remove` is deliberately not rescued: it destroys the anchor, so an
insert relative to it cannot also apply. That combination keeps its
existing behaviour; only `replace` is rescued."""
base = [_step("implement"), _step("tail")]
overlay = Overlay(
id="ov",
extends="wf",
priority=10,
edits=[
OverlayEdit("remove", "implement"),
OverlayEdit("insert_after", "implement", _step("lint")),
],
)

steps, _ = merge_steps(base, [_layer(overlay, "project:ov")])

assert [s["id"] for s in steps] == ["implement", "lint", "tail"]

def test_higher_priority_insert_only_overlay_keeps_base_step(self):
"""A later layer that only inserts must NOT resurrect a lower layer's
replace — the fate still comes from the winning layer."""
base = [_step("implement")]
replacer = Overlay(
id="low", extends="wf", priority=5,
edits=[OverlayEdit(
"replace", "implement",
{**_step("implement"), "command": "low.impl"},
)],
)
inserter = Overlay(
id="high", extends="wf", priority=10,
edits=[OverlayEdit("insert_after", "implement", _step("lint"))],
)

steps, _ = merge_steps(
base,
[_layer(replacer, "project:low"), _layer(inserter, "project:high")],
)

by_id = {s["id"]: s for s in steps}
# The insert-only layer wins the anchor, so the base step survives.
assert by_id["implement"]["command"] == "speckit.specify"
assert "lint" in by_id