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
73 changes: 66 additions & 7 deletions src/specify_cli/workflows/overlays/merge.py
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,51 @@ 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.

For the same reason the rescue applies only to an *unambiguous*
replace-plus-insert layer. If the winning layer also declared a ``remove``
on this anchor, the layer is asking for two incompatible fates and picking
one is that same separate question -- so the previous trailing-insert
outcome is preserved and such layers stay byte-identical to their
pre-rescue 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 not winning_layer:
continue
if edit.operation == "remove":
# Ambiguous layer (replace *and* remove on one anchor): leave the
# pre-existing trailing-insert fate untouched.
return edits[-1]
if 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 @@ -244,7 +289,10 @@ def _traverse_and_apply(
steps).

*edits* are expected to be in merge order (lowest priority first, highest
priority last); the winning edit for each anchor is ``edits[-1]``.
priority last). The winning edit for each anchor is chosen by
``_winning_fate_edit`` -- normally ``edits[-1]``, except that a trailing
``insert_*`` does not cancel a ``replace`` declared earlier by that same
overlay. Go through that helper rather than reading ``edits[-1]`` directly.
"""
result: list[dict[str, Any]] = []

Expand All @@ -255,7 +303,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)
Comment thread
jawwad-ali marked this conversation as resolved.
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 +323,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 +401,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
147 changes: 147 additions & 0 deletions tests/workflows/test_overlay_merge.py
Original file line number Diff line number Diff line change
Expand Up @@ -733,3 +733,150 @@ 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` 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`.

`remove` is explicitly out of scope: it destroys the anchor, so an insert
relative to it cannot also apply. Layers combining `remove` with a trailing
insert — with or without a `replace` alongside — keep their pre-existing
behaviour, as `test_remove_then_insert_after_same_overlay_is_unchanged` and
`test_replace_and_remove_with_trailing_insert_is_unchanged` pin.
"""

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"]

@pytest.mark.parametrize(
"order",
["replace_first", "remove_first"],
)
def test_replace_and_remove_with_trailing_insert_is_unchanged(self, order):
"""A layer asking for two incompatible fates keeps the old outcome.

The rescue applies only to an unambiguous replace-plus-insert layer. If
the winning layer also declared a `remove` on the anchor it is asking
for two incompatible fates, and choosing one is the separate question
this change deliberately does not answer — so the trailing-insert
outcome is preserved and the base step survives, exactly as it did
before the rescue existed. Pinned in both declaration orders so the
rescue cannot start honouring whichever of the two happens to come
first.
"""
base = [_step("implement"), _step("tail")]
replace_edit = OverlayEdit(
"replace", "implement", {**_step("implement"), "command": "custom.impl"}
)
remove_edit = OverlayEdit("remove", "implement")
fate_edits = (
[replace_edit, remove_edit]
if order == "replace_first"
else [remove_edit, replace_edit]
)
overlay = Overlay(
id="ov",
extends="wf",
priority=10,
edits=[
*fate_edits,
OverlayEdit("insert_after", "implement", _step("lint")),
],
)

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

assert [s["id"] for s in steps] == ["implement", "lint", "tail"]
# The base step, not the replacement: the ambiguous layer is left alone.
by_id = {s["id"]: s for s in steps}
assert by_id["implement"].get("command") != "custom.impl"

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