mirror of
https://github.com/github/spec-kit.git
synced 2026-08-03 06:26:30 +08:00
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)
This commit is contained in:
Markus
@@ -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
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user