refactor(workflows): simplify overlay architecture to 2-tier

Remove installed overlays tier to enforce clean separation of concerns:
- workflow add installs workflows only (no overlay copying)
- workflow overlay add installs overlays only (project-local)

Changes:
- Remove InstalledOverlaySource class and all references
- Remove overlay-copying logic from _validate_and_install_local()
- Update WorkflowResolver to 2-tier: project overlays + base workflow
- Fix --priority override timing: apply before validation, not after
- Remove tests for installed overlays (no longer applicable)

Rationale: If upstream controls both base workflow and shipped overlays,
and both get overwritten on bundle update, there's no reason to ship
overlays separately. Overlays only make sense when someone other than
the base author adds them.

Resolves all three review findings from PR #3557:
- r3594064677: workflow add no longer copies overlays from all call sites
- r3594064705: --priority override now applied before validation
- r3594064726: no stale installed overlays (tier removed entirely)

Assisted-by: Claude (model: claude-opus-4-7, autonomous)
This commit is contained in:
Markus
2026-07-16 16:24:00 +02:00
parent c70a5d61cf
commit cc281856d6
8 changed files with 67 additions and 321 deletions

View File

@@ -637,16 +637,6 @@ def workflow_add(
import shutil import shutil
shutil.copy2(yaml_path, dest_dir / "workflow.yml") shutil.copy2(yaml_path, dest_dir / "workflow.yml")
# If the source is a directory that also contains an overlays/ subdirectory,
# copy it alongside the workflow.yml so installed overlays are preserved.
source_dir = yaml_path.parent
source_overlays = source_dir / "overlays"
if source_overlays.is_dir() and not source_overlays.is_symlink():
dest_overlays = dest_dir / "overlays"
if dest_overlays.exists():
shutil.rmtree(dest_overlays)
shutil.copytree(source_overlays, dest_overlays)
registry.add(definition.id, { registry.add(definition.id, {
"name": definition.name, "name": definition.name,
"version": definition.version, "version": definition.version,

View File

@@ -8,7 +8,6 @@ from ..engine import WorkflowDefinition
from .composer import StepListComposer from .composer import StepListComposer
from .layer_sources import ( from .layer_sources import (
BaseWorkflowSource, BaseWorkflowSource,
InstalledOverlaySource,
Layer, Layer,
ProjectOverlaySource, ProjectOverlaySource,
) )
@@ -18,9 +17,8 @@ from .merge import ComposedStep
class WorkflowResolver: class WorkflowResolver:
"""Resolves a workflow ID to its composed ``WorkflowDefinition``. """Resolves a workflow ID to its composed ``WorkflowDefinition``.
Collects layers from three tiers: Collects layers from two tiers:
- project-local overlays (``.specify/workflows/overlays/<id>/*.yml``) - project-local overlays (``.specify/workflows/overlays/<id>/*.yml``)
- installed overlays shipped with a workflow (``.specify/workflows/<id>/overlays/*.yml``)
- the base workflow itself (``.specify/workflows/<id>/workflow.yml``) - the base workflow itself (``.specify/workflows/<id>/workflow.yml``)
Resolution is higher-wins: overlays with higher priority are applied Resolution is higher-wins: overlays with higher priority are applied
@@ -31,7 +29,6 @@ class WorkflowResolver:
self.project_root = project_root self.project_root = project_root
self._sources = [ self._sources = [
ProjectOverlaySource(project_root), ProjectOverlaySource(project_root),
InstalledOverlaySource(project_root),
BaseWorkflowSource(project_root), BaseWorkflowSource(project_root),
] ]
self._composer = StepListComposer() self._composer = StepListComposer()
@@ -39,8 +36,7 @@ class WorkflowResolver:
def collect_all_layers(self, workflow_id: str) -> list[Layer]: def collect_all_layers(self, workflow_id: str) -> list[Layer]:
"""Collect all layers sorted by resolver precedence. """Collect all layers sorted by resolver precedence.
Higher priority wins. Ties are broken so project overlays rank above Higher priority wins. Ties are broken alphabetically by source.
installed overlays (they are listed first in higher-wins order).
""" """
all_layers: list[Layer] = [] all_layers: list[Layer] = []
for source in self._sources: for source in self._sources:
@@ -48,11 +44,7 @@ class WorkflowResolver:
return sorted( return sorted(
all_layers, all_layers,
key=lambda layer: ( key=lambda layer: (-layer.priority, layer.source),
-layer.priority,
0 if layer.tier == "project-overlay" else 1,
layer.source,
),
) )
def resolve(self, workflow_id: str) -> WorkflowDefinition: def resolve(self, workflow_id: str) -> WorkflowDefinition:

View File

@@ -147,6 +147,12 @@ def workflow_overlay_add(
err_console.print(f"[red]Error:[/red] {err}") err_console.print(f"[red]Error:[/red] {err}")
return None return None
# Apply --priority override before validation so a valid CLI priority
# can fix a missing or invalid priority in the file.
if priority is not None:
_validate_priority(priority)
data["priority"] = priority
overlay, validation_errors = validate_overlay_yaml(data) overlay, validation_errors = validate_overlay_yaml(data)
if overlay is None: if overlay is None:
err_console.print("[red]Error:[/red] Overlay validation failed:") err_console.print("[red]Error:[/red] Overlay validation failed:")
@@ -154,17 +160,6 @@ def workflow_overlay_add(
err_console.print(f" \u2022 {err}") err_console.print(f" \u2022 {err}")
return None return None
if priority is not None:
_validate_priority(priority)
data["priority"] = priority
# Re-validate after mutation.
overlay, validation_errors = validate_overlay_yaml(data)
if overlay is None:
err_console.print("[red]Error:[/red] Overlay validation failed:")
for err in validation_errors:
err_console.print(f" \u2022 {err}")
return None
target_dir = _project_overlay_dir(project_root, overlay.extends) target_dir = _project_overlay_dir(project_root, overlay.extends)
target_dir.mkdir(parents=True, exist_ok=True) target_dir.mkdir(parents=True, exist_ok=True)
target_path = _ensure_contained_path( target_path = _ensure_contained_path(

View File

@@ -75,51 +75,6 @@ class ProjectOverlaySource:
return layers return layers
class InstalledOverlaySource:
"""Installed overlays: ``.specify/workflows/<id>/overlays/*.yml``."""
tier = "installed-overlay"
def __init__(self, project_root: Path) -> None:
self.project_root = project_root
self.workflows_dir = project_root / ".specify" / "workflows"
def collect(self, workflow_id: str) -> list[Layer]:
"""Collect all installed overlays shipped with the given workflow."""
installed_overlay_dir = self.workflows_dir / workflow_id / "overlays"
if installed_overlay_dir.is_symlink():
raise OverlayLoadError(
installed_overlay_dir,
["Symlinked overlay directories are not allowed"],
)
if not installed_overlay_dir.is_dir():
return []
layers: list[Layer] = []
for path in sorted(installed_overlay_dir.iterdir()):
if not path.is_file() or path.suffix not in (".yml", ".yaml"):
continue
if path.is_symlink():
raise OverlayLoadError(path, ["Symlinked overlay files are not allowed"])
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
overlay, errors = validate_overlay_yaml(data)
if overlay is None or errors:
raise OverlayLoadError(path, errors)
if not overlay.enabled:
continue
if overlay.extends != workflow_id:
continue
layers.append(
Layer(
content=overlay,
source=f"installed:{overlay.id}",
tier=self.tier,
priority=overlay.priority,
path=path,
)
)
return layers
class BaseWorkflowSource: class BaseWorkflowSource:
"""Base workflow layer: ``.specify/workflows/<id>/workflow.yml``.""" """Base workflow layer: ``.specify/workflows/<id>/workflow.yml``."""

View File

@@ -82,6 +82,48 @@ class TestOverlayCli:
data = yaml.safe_load(installed.read_text(encoding="utf-8")) data = yaml.safe_load(installed.read_text(encoding="utf-8"))
assert data["priority"] == 5 assert data["priority"] == 5
def test_overlay_add_with_priority_override_missing_in_file(self, project_dir, monkeypatch):
"""--priority must fix a missing priority in the overlay file."""
monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir)
_write_workflow(
project_dir,
"wf",
{
"schema_version": "1.0",
"workflow": {"id": "wf", "name": "WF", "version": "1.0.0"},
"steps": [{"id": "a", "type": "command", "command": "echo"}],
},
)
# Overlay file has NO priority field
overlay_file = project_dir / "overlay.yml"
overlay_file.write_text(
yaml.safe_dump(
{
"id": "ov1",
"extends": "wf",
"edits": [
{
"operation": "insert_after",
"anchor": "a",
"step": {"id": "new", "type": "command", "command": "echo"},
}
],
}
),
encoding="utf-8",
)
result = runner.invoke(
app, ["workflow", "overlay", "add", str(overlay_file), "--priority", "5"]
)
assert result.exit_code == 0, result.output
assert "Overlay 'ov1' added" in result.output
installed = project_dir / ".specify" / "workflows" / "overlays" / "wf" / "ov1.yml"
assert installed.is_file()
data = yaml.safe_load(installed.read_text(encoding="utf-8"))
assert data["priority"] == 5
def test_overlay_set_priority(self, project_dir, monkeypatch): def test_overlay_set_priority(self, project_dir, monkeypatch):
monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir)
_write_workflow( _write_workflow(
@@ -308,7 +350,9 @@ class TestOverlayCli:
result = runner.invoke(app, ["workflow", "add", str(source_dir)]) result = runner.invoke(app, ["workflow", "add", str(source_dir)])
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
# Overlays in the source directory should NOT be copied — workflow add
# only installs the workflow.yml, not sibling overlays.
installed_overlay = ( installed_overlay = (
project_dir / ".specify" / "workflows" / "wf" / "overlays" / "ov1.yml" project_dir / ".specify" / "workflows" / "wf" / "overlays" / "ov1.yml"
) )
assert installed_overlay.is_file() assert not installed_overlay.exists()

View File

@@ -336,31 +336,32 @@ class TestMergeSteps:
ComposedStep("high-after", "project:high"), ComposedStep("high-after", "project:high"),
] ]
def test_merge_steps_project_wins_tie_over_installed_same_anchor(self): def test_merge_steps_later_overlay_wins_tie_same_anchor(self):
"""When two overlays have the same priority, the one applied later wins."""
base = [_step("a")] base = [_step("a")]
installed = Overlay( first = Overlay(
id="same", id="first",
extends="wf", extends="wf",
priority=10, priority=10,
edits=[OverlayEdit("replace", "a", _step("installed-replace"))], edits=[OverlayEdit("replace", "a", _step("first-replace"))],
) )
project = Overlay( second = Overlay(
id="same", id="second",
extends="wf", extends="wf",
priority=10, priority=10,
edits=[OverlayEdit("replace", "a", _step("project-replace"))], edits=[OverlayEdit("replace", "a", _step("second-replace"))],
) )
# Merge order: installed first, then project (project wins tie). # Merge order: first applied, then second wins tie.
steps, attribution = merge_steps( steps, attribution = merge_steps(
base, base,
[ [
_layer(installed, "installed:same"), _layer(first, "overlay:first"),
_layer(project, "project:same"), _layer(second, "overlay:second"),
], ],
) )
assert [s["id"] for s in steps] == ["project-replace"] assert [s["id"] for s in steps] == ["second-replace"]
assert any( assert any(
composed.step_id == "project-replace" and composed.source == "project:same" composed.step_id == "second-replace" and composed.source == "overlay:second"
for composed in attribution for composed in attribution
) )

View File

@@ -251,38 +251,3 @@ class TestOverlayPathTraversal:
result = runner.invoke(app, ["workflow", "overlay", "list", "wf"]) result = runner.invoke(app, ["workflow", "overlay", "list", "wf"])
assert result.exit_code != 0, result.output assert result.exit_code != 0, result.output
assert "symlink" in result.output.lower() assert "symlink" in result.output.lower()
def test_overlay_list_rejects_symlinked_installed_overlays_dir(self, project_dir, monkeypatch, tmp_path):
"""Overlay list must reject a symlinked installed overlays directory."""
monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir)
# Create a real overlays directory outside the project.
outside_dir = tmp_path / "outside_installed"
outside_dir.mkdir()
outside_dir.joinpath("evil.yml").write_text(
yaml.safe_dump(
{
"id": "evil",
"extends": "wf",
"priority": 100,
"edits": [
{
"operation": "insert_after",
"anchor": "a",
"step": {"id": "evil-step", "type": "command", "command": "echo"},
}
],
}
),
encoding="utf-8",
)
# Create a workflow directory and symlink its overlays/ to the outside.
wf_dir = project_dir / ".specify" / "workflows" / "wf"
wf_dir.mkdir(parents=True, exist_ok=True)
symlink_dir = wf_dir / "overlays"
symlink_dir.symlink_to(outside_dir)
result = runner.invoke(app, ["workflow", "overlay", "list", "wf"])
assert result.exit_code != 0, result.output
assert "symlink" in result.output.lower()

View File

@@ -77,37 +77,6 @@ class TestWorkflowResolver:
definition = resolver.resolve("wf") definition = resolver.resolve("wf")
assert [s["id"] for s in definition.steps] == ["a", "new", "b"] assert [s["id"] for s in definition.steps] == ["a", "new", "b"]
def test_resolve_with_installed_overlay(self, project_dir):
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)
installed_dir = project_dir / ".specify" / "workflows" / "wf" / "overlays"
installed_dir.mkdir(parents=True, exist_ok=True)
installed_dir.joinpath("shipped.yml").write_text(
yaml.safe_dump(
{
"id": "shipped",
"extends": "wf",
"priority": 5,
"edits": [
{
"operation": "insert_after",
"anchor": "a",
"step": {"id": "shipped-new", "type": "command", "command": "echo"},
}
],
}
),
encoding="utf-8",
)
resolver = WorkflowResolver(project_dir)
definition = resolver.resolve("wf")
assert [s["id"] for s in definition.steps] == ["a", "shipped-new"]
def test_resolve_higher_priority_wins(self, project_dir): def test_resolve_higher_priority_wins(self, project_dir):
data = { data = {
"schema_version": "1.0", "schema_version": "1.0",
@@ -156,56 +125,6 @@ class TestWorkflowResolver:
# ends up closer to the anchor and wins the conflict. # ends up closer to the anchor and wins the conflict.
assert [s["id"] for s in definition.steps] == ["a", "high-step", "low-step"] assert [s["id"] for s in definition.steps] == ["a", "high-step", "low-step"]
def test_resolve_project_wins_tie_with_installed(self, project_dir):
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)
installed_dir = project_dir / ".specify" / "workflows" / "wf" / "overlays"
installed_dir.mkdir(parents=True, exist_ok=True)
installed_dir.joinpath("shipped.yml").write_text(
yaml.safe_dump(
{
"id": "shipped",
"extends": "wf",
"priority": 10,
"edits": [
{
"operation": "insert_after",
"anchor": "a",
"step": {"id": "installed-step", "type": "command", "command": "echo"},
}
],
}
),
encoding="utf-8",
)
_write_overlay(
project_dir,
"wf",
"project",
{
"id": "project",
"extends": "wf",
"priority": 10,
"edits": [
{
"operation": "insert_after",
"anchor": "a",
"step": {"id": "project-step", "type": "command", "command": "echo"},
}
],
},
)
resolver = WorkflowResolver(project_dir)
definition = resolver.resolve("wf")
# Same priority: installed applied first, then project wins tie by
# ending up closer to the anchor.
assert [s["id"] for s in definition.steps] == ["a", "project-step", "installed-step"]
def test_resolve_with_layers_returns_attribution(self, project_dir): def test_resolve_with_layers_returns_attribution(self, project_dir):
data = { data = {
"schema_version": "1.0", "schema_version": "1.0",
@@ -237,57 +156,6 @@ class TestWorkflowResolver:
assert any(layer.tier == "base" for layer in layers) assert any(layer.tier == "base" for layer in layers)
assert attribution == [ComposedStep("a", "base"), ComposedStep("new", "project:ov1")] assert attribution == [ComposedStep("a", "base"), ComposedStep("new", "project:ov1")]
def test_resolve_attribution_distinguishes_project_and_installed_same_id(self, project_dir):
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)
installed_dir = project_dir / ".specify" / "workflows" / "wf" / "overlays"
installed_dir.mkdir(parents=True, exist_ok=True)
installed_dir.joinpath("same.yml").write_text(
yaml.safe_dump(
{
"id": "same",
"extends": "wf",
"priority": 10,
"edits": [
{
"operation": "insert_after",
"anchor": "a",
"step": {"id": "installed-new", "type": "command", "command": "echo"},
}
],
}
),
encoding="utf-8",
)
_write_overlay(
project_dir,
"wf",
"same",
{
"id": "same",
"extends": "wf",
"priority": 10,
"edits": [
{
"operation": "insert_after",
"anchor": "a",
"step": {"id": "project-new", "type": "command", "command": "echo"},
}
],
},
)
resolver = WorkflowResolver(project_dir)
definition, _layers, attribution = resolver.resolve_with_layers("wf")
assert [s["id"] for s in definition.steps] == ["a", "project-new", "installed-new"]
sources = {c.step_id: c.source for c in attribution}
assert sources["project-new"] == "project:same"
assert sources["installed-new"] == "installed:same"
def test_resolve_attribution_for_nested_base_steps(self, project_dir): def test_resolve_attribution_for_nested_base_steps(self, project_dir):
data = { data = {
"schema_version": "1.0", "schema_version": "1.0",
@@ -335,31 +203,6 @@ class TestWorkflowResolver:
with pytest.raises(ValueError): with pytest.raises(ValueError):
resolver.resolve("wf") resolver.resolve("wf")
def test_resolve_invalid_installed_overlay_fails(self, project_dir):
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)
installed_dir = project_dir / ".specify" / "workflows" / "wf" / "overlays"
installed_dir.mkdir(parents=True, exist_ok=True)
installed_dir.joinpath("broken.yml").write_text(
yaml.safe_dump(
{
"id": "broken",
"extends": "wf",
"priority": 10,
"edits": "not-a-list",
}
),
encoding="utf-8",
)
resolver = WorkflowResolver(project_dir)
with pytest.raises(ValueError):
resolver.resolve("wf")
def test_resolve_disabled_overlay_is_skipped(self, project_dir): def test_resolve_disabled_overlay_is_skipped(self, project_dir):
data = { data = {
"schema_version": "1.0", "schema_version": "1.0",
@@ -493,45 +336,6 @@ class TestWorkflowResolver:
with pytest.raises(ValueError, match="Symlinked overlay directories are not allowed"): with pytest.raises(ValueError, match="Symlinked overlay directories are not allowed"):
resolver.resolve("wf") resolver.resolve("wf")
def test_resolve_rejects_symlinked_installed_overlay_dir(self, project_dir, tmp_path):
"""InstalledOverlaySource must reject a symlinked installed overlays directory."""
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)
# Create a real overlays directory outside the project with a malicious overlay.
outside_dir = tmp_path / "outside_installed"
outside_dir.mkdir(parents=True, exist_ok=True)
outside_dir.joinpath("evil.yml").write_text(
yaml.safe_dump(
{
"id": "evil",
"extends": "wf",
"priority": 100,
"edits": [
{
"operation": "insert_after",
"anchor": "a",
"step": {"id": "evil-step", "type": "command", "command": "rm -rf /"},
}
],
}
),
encoding="utf-8",
)
# Symlink the installed overlays directory to the outside location.
wf_dir = project_dir / ".specify" / "workflows" / "wf"
symlink_dir = wf_dir / "overlays"
symlink_dir.symlink_to(outside_dir)
resolver = WorkflowResolver(project_dir)
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): def test_resolve_attribution_for_inserted_composite_step(self, project_dir):
"""Inserted composite steps must attribute nested children to the overlay source.""" """Inserted composite steps must attribute nested children to the overlay source."""
data = { data = {