github/github-spec-kit
1
0
Fork 0
mirror of https://github.com/github/spec-kit.git synced 2026-08-03 06:26:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(workflows): WorkflowResolver standalone (PR 1) (#3557)

Browse Source 
* feat(workflows): add standalone WorkflowResolver and overlay subsystem

Implement PR 1 of the workflow-overlays plan: a concrete, standalone
WorkflowResolver for downstream workflow extensibility without touching the
Preset subsystem.

- Add overlay manifest schema (Overlay, OverlayEdit, validate_overlay_yaml)
- Add pure-function merge engine (find_step, apply_edit, merge_steps,
  validate_edits) with recursive anchor search and higher-wins semantics
- Add StepListComposer and tiered layer sources (project, installed, base)
- Add WorkflowResolver facade with inline HIGHER_WINS priority sorting
- Add CLI verbs: workflow overlay add/set-priority/enable/disable/remove/list
  and workflow resolve <id>
- Wire WorkflowEngine.load_workflow through WorkflowResolver
- Extend workflow add to copy optional overlays/ subdirectory from local
  workflow directories
- Add comprehensive unit, integration, and security tests

Refs: discussion #3473 (https://github.com/github/spec-kit/discussions/3473)

Assisted-by: Kimi (model: opencode-go/kimi-k2.7-code, autonomous)

* fix(workflows): reject symlinked overlay directories in layer sources

Address PR #3557 review comments r3594064534 and r3594064563:

- ProjectOverlaySource.collect now rejects symlinked per-workflow overlay
  directories (.specify/workflows/overlays/<id>) before iterating
- InstalledOverlaySource.collect now rejects symlinked installed overlay
  directories (.specify/workflows/<id>/overlays) before iterating
- workflow_overlay_list catches ValueError from resolver and exits with
  code 1 instead of crashing on unhandled exceptions
- Added .specify/workflows/overlays to _reject_unsafe_workflow_storage
  chokepoint for defense-in-depth

These guards prevent symlinked overlay directories from redirecting
auto-loaded overlay YAML to attacker-controlled content outside the
project, which could inject executable shell steps into trusted workflows.

Refs: PR #3557 review comments r3594064534, r3594064563

Assisted-by: opencode-go/qwen3.7-max (autonomous)

* 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)

* 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)

* fix(workflows): harden overlay symlink handling

Assisted-by: GitHub Copilot (model: GPT-5.4, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* docs(workflows): remove stale installed-overlay references from workflows.md

The 2-tier refactor (cc28185) removed the installed-overlay tier entirely,
but docs/reference/workflows.md was not updated. This commit addresses all
four Cluster 2 findings from the PR review:

- workflow add: remove sentence about copying overlays/ subdirectory
- How Overlays Work: drop installed-overlay table row and precedence prose;
  rewrite to 2-tier model (project overlays only, source-order tie-break)
- overlay remove: drop trailing sentence about installed overlays
- Interaction with Bundles: rewrite to say workflow add installs only
  workflow.yml; remove installed-overlay discovery language

Fixes: r3596368791, r3596368831, r3596368873, r3596368919

Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(overlays): detect ancestor-conflict anchors in merge_steps

When two overlay edits target anchors that share a parent/descendant
relationship (e.g. remove an if-step + insert_after a nested child),
merge_steps processed them independently and in dict-insertion order,
making the outcome non-deterministic.

Add two private helpers to merge.py:
- _descendant_ids(step): returns all step IDs nested inside a step dict
  by delegating to the existing _all_base_step_ids helper on children.
- _check_anchor_conflicts(anchors, base_steps): for each targeted anchor
  finds its descendants and checks whether any other targeted anchor is
  among them; returns human-readable error strings.

Wire _check_anchor_conflicts into merge_steps immediately after
edits_by_anchor is built, before any tree mutation occurs. Raises
ValueError listing the conflicting anchor pair(s) so overlay authors
know exactly what to fix.

Add TestMergeStepsAncestorConflicts (6 cases):
- remove parent + insert_after child raises ValueError
- replace parent + remove child raises ValueError
- conflict across multiple overlays raises ValueError
- sibling anchors (not ancestor/descendant) pass
- single anchor passes
- parent targeted but child not targeted passes

Closes review comment r3596368746 (PR #3557, round 2, cluster 3).

Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(overlays): fix over-broad conflict detection and non-deterministic ID collision

Finding 1.1 — _check_anchor_conflicts was rejecting any ancestor/descendant
anchor pair, including insert-only edits that are perfectly safe. Only
replace/remove on an ancestor can destroy its subtree and make a descendant
anchor unresolvable. Change the signature to accept a dict[str, str]
(anchor → winning operation) and skip the check for insert_after/insert_before.

Finding 1.2 — merge_steps was calling find_step on the already-mutated tree,
so a replacement step that reused a base step ID could be accidentally targeted
by a later edit group (non-deterministic result depending on dict iteration
order). Replace the anchor-group loop with a single-pass _traverse_and_apply
that walks the original tree structure and applies edits as each step is
encountered. Anchors are never re-looked up in a mutated tree.

Design invariant enforced: overlays always apply to the original base tree and
cannot target steps introduced by other overlays. Non-remove edits on non-base
anchors now raise ValueError early.

Also removes apply_edit (no production callers, only tested in isolation) and
its test class — the new traversal inlines the same mechanics without the
find_step round-trip.

Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(overlays): reject ID trailing newlines and reuse existing .yaml path

Fix two input validation bugs in the overlay layer (Group 2 of copilot
review PR #3557):

1. _validate_safe_id in schema.py used re.match() which anchors only at
   the start of the string, so IDs like 'overlay\n' passed validation
   and could produce newline-containing file paths. Changed to fullmatch()
   so the entire string must satisfy the pattern.

2. workflow_overlay_add always wrote <id>.yml without checking whether
   <id>.yaml already existed. Since the resolver loads both extensions,
   this created two active layers whose edits applied twice. Now uses
   the existing _find_overlay_file() to detect a pre-existing file and
   reuse its path, falling back to .yml only for new overlays.

Tests added for both fixes.

Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(overlays): fix display order inversion and wrap file-read errors

Finding group 3 from copilot-review-v2.md:

3.1 — Precedence display inverted (overlays/__init__.py)
collect_all_layers used a single-pass sort by (-priority, source_asc),
which placed the *losing* equal-priority source first in the display
while claiming "highest first". Fix: two-pass stable sort — source
descending then priority descending — so the actual winner (last applied
by the composer) rises to the top of the display.

3.2 — Unwrapped file-read errors (overlays/layer_sources.py)
Only yaml.YAMLError was caught around path.read_text(), so an
unreadable or non-UTF-8 overlay produced a raw traceback. Fix: widen
the except clause to (yaml.YAMLError, OSError, UnicodeDecodeError),
matching the pattern used throughout catalog.py.

Tests:
- test_workflow_resolve_equal_priority_winner_shown_first: verifies
  project:zzz (the winner) appears before project:aaa in workflow resolve
  output when both overlays share the same priority.
- tests/workflows/test_overlay_layer_sources.py (new): OSError and
  non-UTF-8 bytes both produce OverlayLoadError, not raw tracebacks.

Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: rename misleading overlay test

Assisted-by: GitHub Copilot (model: gpt-5.3-codex, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: remove EOF blank line in overlay resolver

Assisted-by: GitHub Copilot (model: gpt-5.3-codex, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix: handle overlay read and enumeration errors

Assisted-by: GitHub Copilot (model: gpt-5.3-codex, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(overlays): validate resolver workflow IDs

Reject unsafe and reserved workflow IDs before overlay or base sources construct paths, preventing traversal through resolver and engine fallback paths.

Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix(overlays): drop _remove_sources_recursively from remove branch

In _traverse_and_apply, the remove branch called _remove_sources_recursively
to clean up attribution entries for the deleted step.  This was inherited from
the old apply_edit loop (c70a5d6) where it was needed because the sources dict
was queried exhaustively.

In the current single-pass design, _build_attribution only traverses the result
list, so stale sources entries for removed steps are never read.  The cleanup
call is therefore unnecessary — and actively harmful when another overlay has
replaced a different step with a new step that reuses the same ID: the pop
clobbers the replacement's attribution entry, causing workflow resolve to report
the surviving step as 'unknown'.

Fix: simply remove the _remove_sources_recursively call from the remove branch.
Add an attribution assertion to the existing reused-ID regression test to catch
this case.

Fixes: r3604242050 (Copilot review finding)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)

* Fix CLI overlay ID validation anchoring

Use fullmatch for CLI workflow/overlay ID validation so trailing newlines are rejected consistently with manifest validation.

Add regression coverage for newline-suffixed workflow and overlay IDs in overlay set-priority.

Assisted-by: GitHub Copilot (model: gpt-5.3-codex, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(workflows): validate workflow_id in layer sources before path construction

ProjectOverlaySource.collect() and BaseWorkflowSource.collect() joined
workflow_id directly onto storage paths without validation, enabling
path traversal (e.g. '../../outside') when called outside the
WorkflowResolver.

Add _validate_workflow_id() to layer_sources.py — mirrors the same
_SAFE_ID_PATTERN / _RESERVED_WORKFLOW_IDS check used by WorkflowResolver
in overlays/__init__.py — and call it at the top of both collect()
methods before any path is constructed.

Adds parametrised tests covering unsafe IDs and verifying no filesystem
access occurs for an invalid ID.

Closes review finding r3604772700.

Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(workflows): align layer source validation with _safe_workflow_id_dir

Plan §4.1 requires that Workflow-ID-Validierung, Symlink-/Containment-
Prüfungen and Fehlerübersetzung must not diverge between workflow
management and the overlay resolver.

My previous fix added ID pattern + reserved-name validation to both
collect() methods but was missing the containment step and the
BaseWorkflowSource directory/file checks that _safe_workflow_id_dir
performs.

Changes:
- Add _ensure_contained_dir(path, root) to layer_sources.py — pure
  domain mirror of overlays/_commands.py::_ensure_contained_dir that
  raises OverlayLoadError instead of typer.Exit
- ProjectOverlaySource.collect(): replace two inline symlink/dir checks
  with _ensure_contained_dir(workflow_overlay_dir, self.overlays_dir),
  adding the missing resolve().relative_to() containment step
- BaseWorkflowSource.collect(): add _ensure_contained_dir on the
  workflow directory, and add workflow.yml symlink check before is_file()

The same logic now lives in three places (workflow CLI, overlay CLI,
layer sources). The DRY extraction to workflows/_validation.py is
deferred to PR 3 per plan §4.1.

Tests: add containment and symlink tests for both sources.

Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix(overlays): resolve identity from manifest field, not filename

Align overlay identity resolution with the project-wide convention:
presets use preset.id, extensions use extension.id, workflows use
workflow.id, and workflow steps use step.type_key. Overlays must
derive identity from the manifest id field, not the filename.

Rewrite _find_overlay_file() to scan all YAML files in the overlay
directory and match on the manifest id field, fixing the bug where
enable/disable/remove/set-priority failed when filename != manifest id.

Closes: PR #3557 discussion r3605010197

Assisted-by: opencode-go/qwen3.7-max (autonomous)

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix(workflows): make validation behavior consistent across YAML loading paths

Address PR #3557 review finding r3607632921:

- Wrap yaml.YAMLError → ValueError in from_yaml() and from_string() so
  malformed YAML matches the documented exception contract
- Add except ValueError to workflow_info to handle composition errors
  cleanly instead of crashing with a raw traceback
- Remove validate_workflow() from compose() so the resolver path is
  parse-only like all other YAML loading mechanisms; callers validate
  explicitly via engine.validate()
- Update test to reflect new behavior: resolve() returns composed
  definition, caller validates separately

Assisted-by: opencode-go/qwen3.7-max (autonomous)

* fix(overlays): list disabled overlays in management view

Keep disabled overlays visible in workflow overlay list while leaving resolution behavior unchanged.

- add an include_disabled opt-in to overlay source/resolver collection
- use include_disabled=True for workflow overlay list
- add regression tests for list visibility and default filtering

Assisted-by: GitHub Copilot (model: GPT-5.4, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* docs: align overlay extends and resolver contract

Assisted-by: GitHub Copilot (model: GPT-5.3-Codex, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: use atomic write for overlay file updates to prevent hard-link attack

Replace in-place write_text() calls in workflow_overlay_add() and
_update_overlay_field() with the same mkstemp → write → os.replace()
pattern used by the workflow installer (_stage_workflow_file /
_commit_workflow_file / _discard_staged_workflow_file).

The prior code rejected symlinks and validated path containment, but a
hard-linked destination file passes both checks while sharing an inode
with an external file. write_text() would then truncate and overwrite
that external inode. The atomic staging approach never opens the
existing destination for writing, eliminating the hard-link vector.

Fixes findings r3608669512 and r3608669517 on PR #3557.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, supervised)

* fix(composer): preserve invalid base definition instead of coercing steps to []

When 'steps' is not a list, returning early with the unmodified
WorkflowDefinition lets validate_workflow surface the proper error
("'steps' must be a list.") to the caller. The previous silent
coercion to [] masked the validation error entirely.

Fixes: https://github.com/github/spec-kit/pull/3557#discussion_r3608669506

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, supervised)

* fix: align workflow overlay priority semantics

Assisted-by: GitHub Copilot (model: GPT-5.6 Terra, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: validate overlay priority presentation

Assisted-by: GitHub Copilot (model: GPT-5.6 Terra, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: catch OverflowError in normalize_priority for float infinity values

YAML values like `priority: .inf` parse to float('inf'), causing
int() to raise OverflowError. This broke validate_overlay_yaml()'s
'validation never raises' contract. Adding OverflowError to the
except clause makes it fall back to the default priority (10),
consistent with other invalid value handling.

Assisted-by: GitHub Copilot (model: claude-sonnet-4.6, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Markus <markus@example.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Markus Wondrak
2026-07-21 15:35:25 +02:00
committed by GitHub
parent 8a5bcc21a5
commit d6fa0460ed
19 changed files with 4938 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
src/specify_cli/workflows/_commands.py
View File

@@ -49,6 +49,13 @@ workflow_step_catalog_app = typer.Typer(
)
workflow_step_app.add_typer(workflow_step_catalog_app, name="catalog")
workflow_overlay_app = typer.Typer(
name="overlay",
help="Manage workflow overlays",
add_completion=False,
)
workflow_app.add_typer(workflow_overlay_app, name="overlay")
def _error_console(json_output: bool):
"""Console for error text: stderr under ``--json`` so the JSON stdout
@@ -192,6 +199,10 @@ def _reject_unsafe_workflow_storage(project_root: Path) -> None:
project_root / ".specify" / "workflows" / "runs",
".specify/workflows/runs",
)
_reject_unsafe_dir(
project_root / ".specify" / "workflows" / "overlays",
".specify/workflows/overlays",
)
def _scan_for_workflow_owner(parts: tuple[str, ...]) -> int | None:
@@ -366,7 +377,7 @@ def _resolve_installed_workflow_ownership(
_WORKFLOW_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
_RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"runs", "steps"})
_RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"overlays", "runs", "steps"})
def _reject_insecure_download_redirect(old_url: str, new_url: str) -> None:
@@ -2386,6 +2397,9 @@ def workflow_info(
# Local workflow definition not found on disk; fall back to
# catalog/registry lookup below.
pass
except ValueError as exc:
console.print(f"[red]Error:[/red] Invalid workflow: {_escape_markup(str(exc))}")
raise typer.Exit(1)
if definition:
console.print(f"\n[bold cyan]{definition.name}[/bold cyan] ({definition.id})")
@@ -3152,6 +3166,102 @@ def workflow_step_catalog_remove(
console.print(f"[green]✓[/green] Step catalog source '{removed_name}' removed")
@workflow_overlay_app.command("add")
def workflow_overlay_add_cmd(
source: Path = typer.Argument(..., help="Path to overlay YAML file"),
priority: int = typer.Option(
10,
"--priority",
help="Resolution priority (lower = higher precedence, default 10)",
),
):
"""Add a project-local overlay for a workflow."""
from .overlays._commands import workflow_overlay_add
project_root = _require_specify_project()
if workflow_overlay_add(project_root, source, priority) is None:
raise typer.Exit(1)
@workflow_overlay_app.command("set-priority")
def workflow_overlay_set_priority_cmd(
workflow_id: str = typer.Argument(..., help="Workflow ID the overlay extends"),
overlay_id: str = typer.Argument(..., help="Overlay ID"),
priority: int = typer.Argument(
..., help="New priority (lower = higher precedence)"
),
):
"""Set the priority of a project-local overlay."""
from .overlays._commands import workflow_overlay_set_priority
project_root = _require_specify_project()
if not workflow_overlay_set_priority(project_root, workflow_id, overlay_id, priority):
raise typer.Exit(1)
@workflow_overlay_app.command("enable")
def workflow_overlay_enable_cmd(
workflow_id: str = typer.Argument(..., help="Workflow ID the overlay extends"),
overlay_id: str = typer.Argument(..., help="Overlay ID"),
):
"""Enable a project-local overlay."""
from .overlays._commands import workflow_overlay_enable
project_root = _require_specify_project()
if not workflow_overlay_enable(project_root, workflow_id, overlay_id):
raise typer.Exit(1)
@workflow_overlay_app.command("disable")
def workflow_overlay_disable_cmd(
workflow_id: str = typer.Argument(..., help="Workflow ID the overlay extends"),
overlay_id: str = typer.Argument(..., help="Overlay ID"),
):
"""Disable a project-local overlay."""
from .overlays._commands import workflow_overlay_disable
project_root = _require_specify_project()
if not workflow_overlay_disable(project_root, workflow_id, overlay_id):
raise typer.Exit(1)
@workflow_overlay_app.command("remove")
def workflow_overlay_remove_cmd(
workflow_id: str = typer.Argument(..., help="Workflow ID the overlay extends"),
overlay_id: str = typer.Argument(..., help="Overlay ID"),
):
"""Remove a project-local overlay."""
from .overlays._commands import workflow_overlay_remove
project_root = _require_specify_project()
if not workflow_overlay_remove(project_root, workflow_id, overlay_id):
raise typer.Exit(1)
@workflow_overlay_app.command("list")
def workflow_overlay_list_cmd(
workflow_id: str = typer.Argument(..., help="Workflow ID"),
):
"""List overlays for a workflow."""
from .overlays._commands import workflow_overlay_list
project_root = _require_specify_project()
if workflow_overlay_list(project_root, workflow_id) is None:
raise typer.Exit(1)
@workflow_app.command("resolve")
def workflow_resolve_cmd(
workflow_id: str = typer.Argument(..., help="Workflow ID to resolve"),
):
"""Show layer attribution for a resolved workflow."""
from .overlays._commands import workflow_resolve
project_root = _require_specify_project()
if workflow_resolve(project_root, workflow_id) is None:
raise typer.Exit(1)
def register(app: typer.Typer) -> None:
"""Attach the workflow command group to the root Typer app."""
app.add_typer(workflow_app, name="workflow")

25
src/specify_cli/workflows/engine.py
View File

@@ -79,7 +79,11 @@ class WorkflowDefinition:
def from_yaml(cls, path: Path) -> WorkflowDefinition:
"""Load a workflow definition from a YAML file."""
with open(path, encoding="utf-8") as f:
data = yaml.safe_load(f)
try:
data = yaml.safe_load(f)
except yaml.YAMLError as exc:
msg = f"Invalid YAML in {path}: {exc}"
raise ValueError(msg) from exc
if not isinstance(data, dict):
msg = f"Workflow YAML must be a mapping, got {type(data).__name__}."
raise ValueError(msg)
@@ -88,7 +92,11 @@ class WorkflowDefinition:
@classmethod
def from_string(cls, content: str) -> WorkflowDefinition:
"""Load a workflow definition from a YAML string."""
data = yaml.safe_load(content)
try:
data = yaml.safe_load(content)
except yaml.YAMLError as exc:
msg = f"Invalid YAML: {exc}"
raise ValueError(msg) from exc
if not isinstance(data, dict):
msg = f"Workflow YAML must be a mapping, got {type(data).__name__}."
raise ValueError(msg)
@@ -727,13 +735,24 @@ class WorkflowEngine:
ValueError:
If the workflow YAML is invalid.
"""
from .overlays import WorkflowResolver
path = Path(source).expanduser()
# Try as a direct file path first
if path.suffix.lower() in (".yml", ".yaml") and path.is_file():
return WorkflowDefinition.from_yaml(path)
# Try as an installed workflow ID
# Try as an installed workflow ID, resolving any overlays.
resolver = WorkflowResolver(self.project_root)
try:
return resolver.resolve(str(source))
except FileNotFoundError:
# Fall back to the direct workflow.yml path so callers still get
# the original error when the workflow id is not installed.
pass
# Legacy direct path check for workflows installed without registry entries.
installed_path = (
self.project_root
/ ".specify"

95
src/specify_cli/workflows/overlays/__init__.py Normal file
View File

@@ -0,0 +1,95 @@
"""Workflow overlay resolver — composes installed workflows from layers."""
from __future__ import annotations
from pathlib import Path
from ..engine import WorkflowDefinition
from .composer import StepListComposer
from .layer_sources import (
BaseWorkflowSource,
Layer,
ProjectOverlaySource,
)
from .merge import ComposedStep
from .schema import _RESERVED_WORKFLOW_IDS, _SAFE_ID_PATTERN
def _validate_workflow_id(workflow_id: str) -> None:
"""Reject workflow IDs that are unsafe as installed-storage path segments."""
if (
not isinstance(workflow_id, str)
or not _SAFE_ID_PATTERN.fullmatch(workflow_id)
or workflow_id in _RESERVED_WORKFLOW_IDS
):
raise ValueError(f"Invalid workflow ID: {workflow_id!r}")
class WorkflowResolver:
"""Resolves a workflow ID to its composed ``WorkflowDefinition``.
Collects layers from two tiers:
- project-local overlays (``.specify/workflows/overlays/<id>/*.yml``)
- the base workflow itself (``.specify/workflows/<id>/workflow.yml``)
Resolution is lower-wins: overlays with lower priority numbers are applied
later and override earlier edits on the same anchors.
"""
def __init__(self, project_root: Path) -> None:
self.project_root = project_root
self._sources = [
ProjectOverlaySource(project_root),
BaseWorkflowSource(project_root),
]
self._composer = StepListComposer()
def collect_all_layers(
self, workflow_id: str, *, include_disabled: bool = False
) -> list[Layer]:
"""Collect overlays sorted by precedence, followed by the base layer.
Lower priority numbers win. Ties are sorted alphabetically by source,
matching ``PresetRegistry.list_by_priority()``. The base workflow is a
foundation rather than a precedence candidate, so it is kept separate.
"""
_validate_workflow_id(workflow_id)
all_layers: list[Layer] = []
for source in self._sources:
all_layers.extend(
source.collect(workflow_id, include_disabled=include_disabled)
)
overlays = [layer for layer in all_layers if layer.tier != "base"]
base_layers = [layer for layer in all_layers if layer.tier == "base"]
return (
sorted(overlays, key=lambda layer: (layer.priority, layer.source))
+ base_layers
)
def resolve(self, workflow_id: str) -> WorkflowDefinition:
"""Resolve a workflow ID to its composed definition.
This method composes layers but does not validate workflow semantics;
callers should validate the returned definition when needed.
Raises:
FileNotFoundError: if the workflow cannot be found.
ValueError: if layer collection/composition fails.
"""
layers = self.collect_all_layers(workflow_id)
definition, _ = self._composer.compose(layers)
if definition is None:
raise FileNotFoundError(f"Workflow not found: {workflow_id}")
return definition
def resolve_with_layers(
self, workflow_id: str
) -> tuple[WorkflowDefinition, list[Layer], list[ComposedStep]]:
"""Resolve a workflow and return its definition plus layer attribution."""
layers = self.collect_all_layers(workflow_id)
definition, attribution = self._composer.compose(layers)
if definition is None:
raise FileNotFoundError(f"Workflow not found: {workflow_id}")
return definition, layers, attribution

442
src/specify_cli/workflows/overlays/_commands.py Normal file
View File

@@ -0,0 +1,442 @@
"""CLI handlers for ``specify workflow overlay *`` and ``specify workflow resolve``."""
from __future__ import annotations
from pathlib import Path
from typing import Any
import typer
import yaml
from ..._console import console, err_console
from ...extensions import normalize_priority
from .._commands import (
_commit_workflow_file,
_discard_committed_backup_file,
_reject_unsafe_dir,
_reject_unsafe_workflow_storage,
_safe_discard_staged_workflow_file,
_stage_workflow_file,
)
from . import WorkflowResolver
from .schema import _RESERVED_WORKFLOW_IDS, _SAFE_ID_PATTERN, validate_overlay_yaml
def _validate_overlay_id_or_exit(id_value: str, label: str) -> None:
"""Validate a single-segment overlay/workflow id from CLI arguments."""
if not isinstance(id_value, str) or not id_value:
err_console.print(f"[red]Error:[/red] {label} is required and must be a non-empty string.")
raise typer.Exit(1)
if not _SAFE_ID_PATTERN.fullmatch(id_value):
err_console.print(
f"[red]Error:[/red] Invalid {label} {id_value!r}: "
"only lowercase letters, digits, and hyphens are allowed."
)
raise typer.Exit(1)
def _validate_workflow_id_or_exit(workflow_id: str) -> None:
"""Validate a workflow id, treating the overlay root as reserved."""
_validate_overlay_id_or_exit(workflow_id, "workflow ID")
if workflow_id in _RESERVED_WORKFLOW_IDS:
err_console.print(
f"[red]Error:[/red] Invalid workflow ID {workflow_id!r}: "
"reserved name."
)
raise typer.Exit(1)
def _overlay_root(project_root: Path) -> Path:
"""Return the project-local overlay root after rejecting unsafe ancestors."""
_reject_unsafe_workflow_storage(project_root)
root = project_root / ".specify" / "workflows" / "overlays"
_reject_unsafe_dir(root, ".specify/workflows/overlays")
return root
def _project_overlay_dir(project_root: Path, workflow_id: str) -> Path:
"""Return the project-local overlay directory for a workflow id.
Raises typer.Exit if the resolved path escapes the overlay root.
"""
_validate_workflow_id_or_exit(workflow_id)
root = _overlay_root(project_root)
target = root / workflow_id
return _ensure_contained_dir(target, root)
def _ensure_contained_dir(path: Path, root: Path) -> Path:
"""Ensure *path* resolves inside *root* and is not a symlink.
Returns *path* if safe. Raises typer.Exit on traversal or symlink.
"""
_reject_unsafe_dir(root, ".specify/workflows/overlays")
if path.is_symlink():
err_console.print(
f"[red]Error:[/red] Refusing to use symlinked path {path}."
)
raise typer.Exit(1)
if path.exists() and not path.is_dir():
err_console.print(
f"[red]Error:[/red] Overlay directory path is not a directory: {path}."
)
raise typer.Exit(1)
try:
resolved = path.resolve()
root_resolved = root.resolve()
resolved.relative_to(root_resolved)
except ValueError:
err_console.print(
f"[red]Error:[/red] Path traversal detected: {path} is outside the allowed directory."
)
raise typer.Exit(1)
return path
def _find_overlay_file(project_root: Path, workflow_id: str, overlay_id: str) -> Path | None:
"""Locate a project-local overlay file by its manifest ID, not filename.
Scans all YAML files in the overlay directory and matches on the ``id``
field inside each manifest. This aligns with ``ProjectOverlaySource.collect()``
which also derives identity from the manifest, not the filename.
"""
_validate_workflow_id_or_exit(workflow_id)
_validate_overlay_id_or_exit(overlay_id, "overlay ID")
overlay_dir = _project_overlay_dir(project_root, workflow_id)
if not overlay_dir.is_dir():
return None
try:
entries = sorted(overlay_dir.iterdir())
except OSError:
return None
matches: list[Path] = []
for path in entries:
if not path.is_file() or path.suffix not in (".yml", ".yaml"):
continue
if path.is_symlink():
continue
data, _ = _read_overlay(path)
if data is None:
continue
if data.get("id") == overlay_id:
matches.append(path)
if len(matches) > 1:
paths = ", ".join(str(path) for path in matches)
err_console.print(
f"[red]Error:[/red] Duplicate overlay ID '{overlay_id}' in {paths}. "
"Resolve the duplicate manifest IDs before continuing."
)
raise typer.Exit(1)
return matches[0] if matches else None
def _ensure_contained_path(path: Path, root: Path) -> Path:
"""Return *path* only if it resolves inside *root*; otherwise raise typer.Exit."""
_reject_unsafe_dir(root, ".specify/workflows/overlays")
if path.is_symlink():
err_console.print(
f"[red]Error:[/red] Refusing to use symlinked path {path}."
)
raise typer.Exit(1)
try:
resolved = path.resolve()
root_resolved = root.resolve()
resolved.relative_to(root_resolved)
except ValueError:
err_console.print(
f"[red]Error:[/red] Path traversal detected: {path} is outside the allowed directory."
)
raise typer.Exit(1)
return path
def _read_overlay(path: Path) -> tuple[dict[str, Any] | None, list[str]]:
"""Read and parse an overlay YAML file, returning (data, errors)."""
try:
content = path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as exc:
return None, [f"Failed to read {path}: {exc}"]
try:
data = yaml.safe_load(content)
except yaml.YAMLError as exc:
return None, [f"Invalid YAML in {path}: {exc}"]
if not isinstance(data, dict):
return None, [f"Overlay {path} must be a YAML mapping."]
return data, []
def workflow_overlay_add(
project_root: Path,
source: Path,
priority: int | None = None,
) -> Path | None:
"""Add a project-local overlay from a YAML file.
Returns the path of the installed overlay file, or None on failure.
"""
_reject_unsafe_workflow_storage(project_root)
data, errors = _read_overlay(source)
if data is None:
for err in errors:
err_console.print(f"[red]Error:[/red] {err}")
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:
if isinstance(priority, bool) or not isinstance(priority, int) or priority < 1:
err_console.print("[red]Error:[/red] Priority must be >= 1.")
return None
data["priority"] = normalize_priority(priority)
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
data["priority"] = overlay.priority
target_dir = _project_overlay_dir(project_root, overlay.extends)
# Reuse an existing .yaml file so we don't create a duplicate .yml layer.
existing = _find_overlay_file(project_root, overlay.extends, overlay.id)
if existing is not None:
target_path = existing
else:
target_path = _ensure_contained_path(
target_dir / f"{overlay.id}.yml", _overlay_root(project_root)
)
backup: Path | None = None
try:
target_dir.mkdir(parents=True, exist_ok=True)
existed_before = target_path.exists()
staged = _stage_workflow_file(target_path.parent)
try:
staged.write_bytes(yaml.safe_dump(data, sort_keys=False).encode("utf-8"))
backup = _commit_workflow_file(staged, target_path, existed_before)
except BaseException:
_safe_discard_staged_workflow_file(
staged, target_path.parent, existed_before
)
raise
except OSError as exc:
err_console.print(f"[red]Error:[/red] Failed to write overlay: {exc}")
return None
_discard_committed_backup_file(backup)
console.print(
f"[green]\u2713[/green] Overlay '{overlay.id}' added for workflow '{overlay.extends}'"
)
return target_path
def _update_overlay_field(
project_root: Path,
workflow_id: str,
overlay_id: str,
field: str,
value: Any,
) -> bool:
"""Update a single field in a project-local overlay file."""
_reject_unsafe_workflow_storage(project_root)
path = _find_overlay_file(project_root, workflow_id, overlay_id)
if path is None:
err_console.print(
f"[red]Error:[/red] Overlay '{overlay_id}' not found for workflow '{workflow_id}'"
)
return False
data, errors = _read_overlay(path)
if data is None:
for err in errors:
err_console.print(f"[red]Error:[/red] {err}")
return False
data[field] = value
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 False
backup: Path | None = None
try:
existed_before = path.exists()
staged = _stage_workflow_file(path.parent)
try:
staged.write_bytes(yaml.safe_dump(data, sort_keys=False).encode("utf-8"))
backup = _commit_workflow_file(staged, path, existed_before)
except BaseException:
_safe_discard_staged_workflow_file(staged, path.parent, existed_before)
raise
except OSError as exc:
err_console.print(f"[red]Error:[/red] Failed to write overlay: {exc}")
return False
_discard_committed_backup_file(backup)
return True
def workflow_overlay_set_priority(
project_root: Path,
workflow_id: str,
overlay_id: str,
priority: int,
) -> bool:
"""Set the priority of a project-local overlay."""
if isinstance(priority, bool) or not isinstance(priority, int) or priority < 1:
err_console.print("[red]Error:[/red] Priority must be >= 1.")
raise typer.Exit(1)
normalized_priority = normalize_priority(priority)
if _update_overlay_field(
project_root, workflow_id, overlay_id, "priority", normalized_priority
):
console.print(
f"[green]\u2713[/green] Priority of overlay '{overlay_id}' set to {normalized_priority}"
)
return True
return False
def workflow_overlay_enable(
project_root: Path,
workflow_id: str,
overlay_id: str,
) -> bool:
"""Enable a project-local overlay."""
if _update_overlay_field(project_root, workflow_id, overlay_id, "enabled", True):
console.print(f"[green]\u2713[/green] Overlay '{overlay_id}' enabled")
return True
return False
def workflow_overlay_disable(
project_root: Path,
workflow_id: str,
overlay_id: str,
) -> bool:
"""Disable a project-local overlay."""
if _update_overlay_field(project_root, workflow_id, overlay_id, "enabled", False):
console.print(f"[green]\u2713[/green] Overlay '{overlay_id}' disabled")
return True
return False
def workflow_overlay_remove(
project_root: Path,
workflow_id: str,
overlay_id: str,
) -> bool:
"""Remove a project-local overlay file."""
_reject_unsafe_workflow_storage(project_root)
path = _find_overlay_file(project_root, workflow_id, overlay_id)
if path is None:
err_console.print(
f"[red]Error:[/red] Overlay '{overlay_id}' not found for workflow '{workflow_id}'"
)
return False
try:
path.unlink()
except OSError as exc:
err_console.print(f"[red]Error:[/red] Failed to remove overlay: {exc}")
return False
console.print(f"[green]\u2713[/green] Overlay '{overlay_id}' removed")
return True
def workflow_overlay_list(project_root: Path, workflow_id: str) -> list[dict[str, Any]] | None:
"""List all overlays for a workflow and print a summary table.
Returns the raw list data for machine-readable callers, or None on error.
"""
_reject_unsafe_workflow_storage(project_root)
_validate_workflow_id_or_exit(workflow_id)
resolver = WorkflowResolver(project_root)
try:
layers = resolver.collect_all_layers(workflow_id, include_disabled=True)
except ValueError as exc:
err_console.print(f"[red]Error:[/red] {exc}")
return None
overlays = [layer for layer in layers if layer.tier != "base"]
if not overlays:
console.print(f"[yellow]No overlays found for workflow '{workflow_id}'.[/yellow]")
return []
console.print(f"Overlays for workflow '{workflow_id}':")
rows: list[dict[str, Any]] = []
for layer in overlays:
overlay = layer.content
rows.append({
"id": overlay.id,
"source": layer.source,
"tier": layer.tier,
"priority": normalize_priority(overlay.priority),
"enabled": overlay.enabled,
"path": str(layer.path) if layer.path else None,
})
enabled_marker = "enabled" if overlay.enabled else "disabled"
console.print(
f" \u2022 {overlay.id} (priority={normalize_priority(overlay.priority)}, "
f"source={layer.source}, {enabled_marker})"
)
return rows
def workflow_resolve(project_root: Path, workflow_id: str) -> dict[str, Any] | None:
"""Print layer attribution for a resolved workflow.
Returns a serializable attribution payload.
"""
_reject_unsafe_workflow_storage(project_root)
_validate_workflow_id_or_exit(workflow_id)
resolver = WorkflowResolver(project_root)
try:
definition, layers, attribution = resolver.resolve_with_layers(workflow_id)
except FileNotFoundError:
err_console.print(
f"[red]Error:[/red] Workflow '{workflow_id}' not found"
)
return None
except ValueError as exc:
err_console.print(f"[red]Error:[/red] {exc}")
return None
console.print(f"Resolved workflow '{workflow_id}':")
console.print("Layers (highest precedence first):")
for layer in layers:
priority = (
"n/a" if layer.tier == "base" else str(normalize_priority(layer.priority))
)
console.print(
f" \u2022 [{layer.tier}] {layer.source} "
f"(priority={priority})"
)
console.print("Step attribution:")
for composed in attribution:
console.print(f" \u2022 {composed.step_id}: {composed.source}")
return {
"workflow_id": workflow_id,
"layers": [
{
"source": layer.source,
"tier": layer.tier,
"priority": (
None
if layer.tier == "base"
else normalize_priority(layer.priority)
),
}
for layer in layers
],
"attribution": [
{"step_id": composed.step_id, "source": composed.source}
for composed in attribution
],
}

97
src/specify_cli/workflows/overlays/composer.py Normal file
View File

@@ -0,0 +1,97 @@
"""Workflow overlay composer — builds a WorkflowDefinition from layers."""
from __future__ import annotations
from typing import Any
from ..engine import WorkflowDefinition
from .layer_sources import Layer
from .merge import OverlayLayer, merge_steps, validate_edits
class StepListComposer:
"""Compose a workflow from a base layer and overlay layers.
- The base layer (tier="base") provides the full step list.
- Overlay layers provide edit operations.
- Overlays are applied in merge order: highest priority number first,
lowest last, so lower priority numbers win. Ties are applied by overlay
ID, with the alphabetically last ID winning.
- Returns a parsed WorkflowDefinition; callers must validate separately.
"""
def compose(
self, layers: list[Layer]
) -> tuple[WorkflowDefinition | None, list]:
"""Compose a ``WorkflowDefinition`` from the given layers.
Returns ``(None, [])`` when no base layer is present.
"""
base_layer: Layer | None = None
overlay_layers: list[Layer] = []
for layer in layers:
if layer.tier == "base":
base_layer = layer
else:
overlay_layers.append(layer)
if base_layer is None or base_layer.path is None:
return None, []
# Read the base workflow definition from disk.
base_definition = WorkflowDefinition.from_yaml(base_layer.path)
base_steps = base_definition.data.get("steps", [])
if not isinstance(base_steps, list):
# Preserve the invalid definition intact so validate_workflow can
# report "'steps' must be a list." to the caller; coercing to []
# here would mask that error.
return base_definition, []
# Last applied wins, so apply lower priority numbers last.
merge_order = sorted(
overlay_layers,
key=lambda layer: (-layer.priority, layer.content.id),
)
# Validate edits against base anchors before mutation.
base_step_ids = self._collect_base_step_ids(base_steps)
for layer in merge_order:
edit_errors = validate_edits(layer.content.edits, base_step_ids)
if edit_errors:
raise ValueError(
f"Overlay '{layer.content.id}' has invalid edits:\n - "
+ "\n - ".join(edit_errors)
)
composed_steps, attribution = merge_steps(
base_steps,
[OverlayLayer(layer.content, layer.source) for layer in merge_order],
)
# Build composed data while preserving all non-step fields from base.
composed_data: dict[str, Any] = dict(base_definition.data)
composed_data["steps"] = composed_steps
composed_definition = WorkflowDefinition(composed_data, source_path=base_layer.path)
return composed_definition, attribution
def _collect_base_step_ids(self, steps: list[dict[str, Any]]) -> set[str]:
"""Collect all base step IDs reachable in the step tree."""
ids: set[str] = set()
for step in steps:
if not isinstance(step, dict):
continue
step_id = step.get("id")
if isinstance(step_id, str):
ids.add(step_id)
for key in ("then", "else", "steps", "default"):
nested = step.get(key)
if isinstance(nested, list):
ids.update(self._collect_base_step_ids(nested))
cases = step.get("cases")
if isinstance(cases, dict):
for case_steps in cases.values():
if isinstance(case_steps, list):
ids.update(self._collect_base_step_ids(case_steps))
return ids

234
src/specify_cli/workflows/overlays/layer_sources.py Normal file
View File

@@ -0,0 +1,234 @@
"""Workflow overlay layer sources."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
import yaml
from .schema import Overlay, _RESERVED_WORKFLOW_IDS, _SAFE_ID_PATTERN, validate_overlay_yaml
@dataclass
class Layer:
"""A single layer in the workflow overlay stack."""
content: Overlay
source: str
tier: str
priority: int
path: Path | None = None
class OverlayLoadError(ValueError):
"""Raised when an overlay file cannot be loaded or validated."""
def __init__(self, path: Path, errors: list[str]) -> None:
self.path = path
self.errors = errors
super().__init__(f"Invalid overlay {path}:\n - " + "\n - ".join(errors))
def _validate_workflow_id(workflow_id: str, context_path: Path) -> None:
"""Raise OverlayLoadError if workflow_id is not a safe path-segment identifier.
Mirrors the same check performed by WorkflowResolver so layer sources are
safe to call directly, without going through the resolver.
"""
if (
not isinstance(workflow_id, str)
or not _SAFE_ID_PATTERN.fullmatch(workflow_id)
or workflow_id in _RESERVED_WORKFLOW_IDS
):
raise OverlayLoadError(
context_path,
[f"Invalid workflow ID: {workflow_id!r}"],
)
def _ensure_contained_dir(path: Path, root: Path) -> None:
"""Raise OverlayLoadError if *path* is a symlink, a non-directory, or escapes *root*.
Mirrors the logic of ``_ensure_contained_dir`` in ``overlays/_commands.py``
but raises ``OverlayLoadError`` instead of ``typer.Exit`` so layer sources
can enforce the same invariants without a CLI dependency.
The caller is responsible for ensuring *root* itself is already validated
(e.g. via ``_resolve_project_overlay_root``).
"""
if path.is_symlink():
raise OverlayLoadError(path, ["Symlinked overlay directories are not allowed"])
if path.exists() and not path.is_dir():
raise OverlayLoadError(path, ["Overlay directory path is not a directory"])
try:
path.resolve().relative_to(root.resolve())
except ValueError:
raise OverlayLoadError(
path, ["Path traversal detected: directory escapes allowed root"]
) from None
def _resolve_workflows_root(project_root: Path) -> Path:
"""Return the workflow storage root after rejecting unsafe ancestors."""
project_root_resolved = project_root.resolve()
workflows_root = project_root / ".specify" / "workflows"
current = project_root
for part in (".specify", "workflows"):
current = current / part
if current.is_symlink():
raise OverlayLoadError(
current,
[f"Symlinked workflow directories are not allowed ({current})"],
)
if current.exists() and not current.is_dir():
raise OverlayLoadError(
current,
[f"Workflow directory path is not a directory ({current})"],
)
try:
workflows_root.resolve().relative_to(project_root_resolved)
except ValueError:
raise OverlayLoadError(
workflows_root,
["Workflow directory escapes the project root"],
) from None
return workflows_root
def _resolve_project_overlay_root(project_root: Path) -> Path:
"""Return the unresolved overlay root after rejecting unsafe ancestors."""
workflows_root = _resolve_workflows_root(project_root)
overlays_root = workflows_root / "overlays"
if overlays_root.is_symlink():
raise OverlayLoadError(
overlays_root,
[f"Symlinked overlay directories are not allowed ({overlays_root})"],
)
if overlays_root.exists() and not overlays_root.is_dir():
raise OverlayLoadError(
overlays_root,
[f"Overlay directory path is not a directory ({overlays_root})"],
)
return overlays_root
class ProjectOverlaySource:
"""Project-local overlays: ``.specify/workflows/overlays/<id>/*.yml``."""
tier = "project-overlay"
def __init__(self, project_root: Path) -> None:
self.project_root = project_root
self.overlays_dir = project_root / ".specify" / "workflows" / "overlays"
def collect(self, workflow_id: str, *, include_disabled: bool = False) -> list[Layer]:
"""Collect project-local overlays for the given workflow id.
Args:
workflow_id: Workflow identifier whose overlay directory to scan.
include_disabled: When True, return disabled overlays for
management/list views. Resolution paths keep the default False.
"""
self.overlays_dir = _resolve_project_overlay_root(self.project_root)
_validate_workflow_id(workflow_id, self.overlays_dir)
workflow_overlay_dir = self.overlays_dir / workflow_id
_ensure_contained_dir(workflow_overlay_dir, self.overlays_dir)
if not workflow_overlay_dir.is_dir():
return []
layers: list[Layer] = []
overlay_paths_by_id: dict[str, Path] = {}
try:
entries = sorted(workflow_overlay_dir.iterdir())
except OSError as exc:
raise OverlayLoadError(
workflow_overlay_dir, [f"Cannot enumerate overlays: {exc}"]
) from exc
for path in entries:
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"])
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
except yaml.YAMLError as exc:
raise OverlayLoadError(path, [f"Invalid YAML: {exc}"]) from exc
except (OSError, UnicodeDecodeError) as exc:
raise OverlayLoadError(path, [f"Cannot load overlay: {exc}"]) from exc
if (
not include_disabled
and isinstance(data, dict)
and data.get("enabled", True) is False
):
continue
overlay, errors = validate_overlay_yaml(data)
if overlay is None or errors:
raise OverlayLoadError(path, errors)
if overlay.extends != workflow_id:
raise OverlayLoadError(
path,
[
f"Overlay extends {overlay.extends!r}, but is stored under "
f"workflow {workflow_id!r}."
],
)
first_path = overlay_paths_by_id.get(overlay.id)
if first_path is not None:
raise OverlayLoadError(
path,
[
f"Duplicate overlay id {overlay.id!r}; also declared in "
f"{first_path}."
],
)
overlay_paths_by_id[overlay.id] = path
layers.append(
Layer(
content=overlay,
source=f"project:{overlay.id}",
tier=self.tier,
priority=overlay.priority,
path=path,
)
)
return layers
class BaseWorkflowSource:
"""Base workflow layer: ``.specify/workflows/<id>/workflow.yml``."""
tier = "base"
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, *, include_disabled: bool = False) -> list[Layer]:
"""Return the base workflow as a single layer if it exists."""
self.workflows_dir = _resolve_workflows_root(self.project_root)
_validate_workflow_id(workflow_id, self.workflows_dir)
workflow_dir = self.workflows_dir / workflow_id
_ensure_contained_dir(workflow_dir, self.workflows_dir)
path = workflow_dir / "workflow.yml"
if path.is_symlink():
raise OverlayLoadError(path, ["Symlinked workflow files are not allowed"])
if not path.is_file():
return []
# The base layer is represented by an Overlay with empty edits.
overlay = Overlay(
id=workflow_id,
extends=workflow_id,
priority=0,
edits=[],
)
return [
Layer(
content=overlay,
source="base",
tier=self.tier,
priority=0,
path=path,
)
]

395
src/specify_cli/workflows/overlays/merge.py Normal file
View File

@@ -0,0 +1,395 @@
"""Pure-function merge engine for workflow step lists."""
from __future__ import annotations
import copy
from dataclasses import dataclass
from typing import Any
from .schema import VALID_OPERATIONS, Overlay, OverlayEdit
@dataclass(frozen=True)
class ComposedStep:
"""Attribution tracking for a single composed step."""
step_id: str
source: str
@dataclass(frozen=True)
class OverlayLayer:
"""An overlay together with its layer source for attribution."""
overlay: Overlay
source: str
# Nested step keys that may contain a list of steps.
_NESTED_LIST_KEYS = ("then", "else", "steps", "default")
def find_step(
steps: list[dict[str, Any]], step_id: str
) -> tuple[list[dict[str, Any]], int] | None:
"""Recursively locate a step by ID and return its (parent_list, index).
Searches flat lists and nested lists inside ``then``, ``else``, ``steps``,
``default``, and ``cases.*``. Does *not* descend into ``fan-out`` template
steps because those are runtime-multiplied stamps, not uniquely-addressable
nodes.
"""
for i, step in enumerate(steps):
if not isinstance(step, dict):
continue
if step.get("id") == step_id:
return (steps, i)
for key in _NESTED_LIST_KEYS:
nested = step.get(key)
if isinstance(nested, list):
result = find_step(nested, step_id)
if result is not None:
return result
cases = step.get("cases")
if isinstance(cases, dict):
for case_steps in cases.values():
if isinstance(case_steps, list):
result = find_step(case_steps, step_id)
if result is not None:
return result
return None
def _all_base_step_ids(steps: list[dict[str, Any]]) -> set[str]:
"""Collect all step IDs reachable in a step tree (excluding fan-out templates)."""
ids: set[str] = set()
for step in steps:
if not isinstance(step, dict):
continue
step_id = step.get("id")
if isinstance(step_id, str):
ids.add(step_id)
for key in _NESTED_LIST_KEYS:
nested = step.get(key)
if isinstance(nested, list):
ids.update(_all_base_step_ids(nested))
cases = step.get("cases")
if isinstance(cases, dict):
for case_steps in cases.values():
if isinstance(case_steps, list):
ids.update(_all_base_step_ids(case_steps))
return ids
def _descendant_ids(step: dict[str, Any]) -> set[str]:
"""Return all step IDs nested inside *step* (not including *step* itself)."""
ids: set[str] = set()
for key in _NESTED_LIST_KEYS:
nested = step.get(key)
if isinstance(nested, list):
ids.update(_all_base_step_ids(nested))
cases = step.get("cases")
if isinstance(cases, dict):
for case_steps in cases.values():
if isinstance(case_steps, list):
ids.update(_all_base_step_ids(case_steps))
return ids
def _check_anchor_conflicts(
anchor_operations: dict[str, str],
base_steps: list[dict[str, Any]],
) -> list[str]:
"""Return error messages for anchor pairs where one is an ancestor of the other.
Only flags conflicts where the ancestor's winning edit is ``replace`` or
``remove`` — operations that destroy the subtree and make any descendant
anchor unresolvable. Pure insert operations on an ancestor leave it intact,
so its descendants remain reachable regardless of processing order.
Callers should raise on any returned errors before mutating the step tree.
"""
errors: list[str] = []
for anchor, operation in sorted(anchor_operations.items()):
if operation in ("insert_after", "insert_before"):
# Inserts leave the ancestor step intact; descendants are unaffected.
continue
location = find_step(base_steps, anchor)
if location is None:
continue # missing anchors are reported by validate_edits
parent_list, idx = location
step = parent_list[idx]
conflicting = set(anchor_operations.keys()) & _descendant_ids(step)
for child_anchor in sorted(conflicting):
errors.append(
f"Anchor conflict: '{anchor}' is an ancestor of '{child_anchor}'. "
"Targeting both anchors in the same overlay set produces "
"order-dependent results; restructure edits to avoid nesting."
)
return errors
def _init_sources_recursively(
steps: list[dict[str, Any]], sources: dict[str, str]
) -> None:
"""Initialize attribution sources for all base steps, recursively."""
for step in steps:
if not isinstance(step, dict):
continue
step_id = step.get("id")
if isinstance(step_id, str):
sources[step_id] = "base"
for key in _NESTED_LIST_KEYS:
nested = step.get(key)
if isinstance(nested, list):
_init_sources_recursively(nested, sources)
cases = step.get("cases")
if isinstance(cases, dict):
for case_steps in cases.values():
if isinstance(case_steps, list):
_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) and sources.get(step_id) == "base":
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 _build_attribution(
steps: list[dict[str, Any]],
sources: dict[str, str],
) -> list[ComposedStep]:
"""Build an ordered attribution list from the composed step tree."""
result: list[ComposedStep] = []
for step in steps:
if not isinstance(step, dict):
continue
step_id = step.get("id")
if isinstance(step_id, str):
result.append(ComposedStep(step_id, sources.get(step_id, "unknown")))
for key in _NESTED_LIST_KEYS:
nested = step.get(key)
if isinstance(nested, list):
result.extend(_build_attribution(nested, sources))
cases = step.get("cases")
if isinstance(cases, dict):
for case_steps in cases.values():
if isinstance(case_steps, list):
result.extend(_build_attribution(case_steps, sources))
return result
def _traverse_and_apply(
steps: list[dict[str, Any]],
edits_by_anchor: dict[str, list[tuple[OverlayLayer, OverlayEdit]]],
sources: dict[str, str],
) -> list[dict[str, Any]]:
"""Walk the original step tree and apply overlay edits as each step is encountered.
Edits are always resolved against the *original* structure — this function
traverses the unmodified list passed in, so a replacement step's new ID can
never be mistaken for a base anchor. Nested lists (``then``, ``else``, etc.)
are recursed into only for steps that survive the edit (not for replaced
steps).
*edits* are expected to be in merge order (lowest priority first, highest
priority last); the winning edit for each anchor is ``edits[-1]``.
"""
result: list[dict[str, Any]] = []
for step in steps:
if not isinstance(step, dict):
result.append(step)
continue
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
if winning_edit is not None and winning_edit.operation == "remove":
# Winning edit removes this step; ignore all other edits on this anchor.
# Do NOT call _remove_sources_recursively here: _build_attribution only
# traverses the result list, so stale sources entries for removed steps
# are never read. Calling it would incorrectly pop the attribution of a
# *surviving* step that reuses the same ID (e.g. a replacement step
# introduced by a higher-priority overlay targeting a different anchor).
continue
# Insert before (in merge order).
for layer, edit in edits:
if edit.operation == "insert_before":
new_step = copy.deepcopy(edit.step)
_record_sources_recursively(new_step, layer.source, sources)
result.append(new_step)
if winning_edit is not None and winning_edit.operation == "replace":
winning_layer = edits[-1][0]
new_step = copy.deepcopy(winning_edit.step)
_remove_sources_recursively(step, sources)
_record_sources_recursively(new_step, winning_layer.source, sources)
result.append(new_step)
else:
# No replacement: keep this step and recurse into its nested lists.
for key in _NESTED_LIST_KEYS:
nested = step.get(key)
if isinstance(nested, list):
step[key] = _traverse_and_apply(nested, edits_by_anchor, sources)
cases = step.get("cases")
if isinstance(cases, dict):
for case_key, case_steps in cases.items():
if isinstance(case_steps, list):
cases[case_key] = _traverse_and_apply(case_steps, edits_by_anchor, sources)
result.append(step)
# Insert after (highest priority closest to anchor — reversed merge order).
for layer, edit in reversed(edits):
if edit.operation == "insert_after":
new_step = copy.deepcopy(edit.step)
_record_sources_recursively(new_step, layer.source, sources)
result.append(new_step)
return result
def merge_steps(
base_steps: list[dict[str, Any]],
overlays: list[OverlayLayer],
) -> tuple[list[dict[str, Any]], list[ComposedStep]]:
"""Apply overlays to base steps in merge order and return composed steps.
*overlays* is expected to be sorted by merge order (lowest priority first,
highest priority last). The returned step list is a deep copy of the base;
base_steps is never mutated.
Higher-wins semantics are enforced for edits that target the same base
anchor: the highest-priority edit (last in *overlays*) decides the fate of
the anchor. A lower-priority ``remove`` cannot prevent a higher-priority
``replace`` or ``insert_*`` on the same anchor.
"""
steps = copy.deepcopy(base_steps)
sources: dict[str, str] = {}
_init_sources_recursively(steps, sources)
# Group edits by anchor, preserving merge order.
edits_by_anchor: dict[str, list[tuple[OverlayLayer, OverlayEdit]]] = {}
for layer in overlays:
for edit in layer.overlay.edits:
edits_by_anchor.setdefault(edit.anchor, []).append((layer, edit))
# Raise early for non-remove edits that target anchors not present in the base.
# Overlays always apply to the original tree; they cannot target steps introduced
# by other overlays.
base_ids = _all_base_step_ids(base_steps)
for anchor, anchor_edits in edits_by_anchor.items():
winning_op = anchor_edits[-1][1].operation
if winning_op != "remove" and anchor not in base_ids:
raise ValueError(f"Anchor '{anchor}' not found in workflow steps.")
# Reject edits that target anchors with a parent/descendant relationship when
# 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()
}
anchor_conflicts = _check_anchor_conflicts(anchor_winning_ops, base_steps)
if anchor_conflicts:
raise ValueError(
"Overlay anchor conflict(s) detected:\n - " + "\n - ".join(anchor_conflicts)
)
# Apply all overlay edits via a single-pass traversal of the original tree.
# Each edit is resolved against the original step structure, so a replacement
# step's new ID can never be mistaken for a base anchor in a later edit group.
result = _traverse_and_apply(steps, edits_by_anchor, sources)
attribution = _build_attribution(result, sources)
return result, attribution
def validate_edits(
edits: list[OverlayEdit],
base_step_ids: set[str],
) -> list[str]:
"""Validate overlay edits against a set of known base step IDs.
Returns a list of human-readable error messages. Does not raise.
"""
errors: list[str] = []
for idx, edit in enumerate(edits):
if edit.operation not in VALID_OPERATIONS:
errors.append(f"Edit {idx}: invalid operation {edit.operation!r}.")
continue
if edit.anchor not in base_step_ids:
errors.append(
f"Edit {idx}: anchor '{edit.anchor}' does not match any base step id."
)
if edit.operation == "remove":
if edit.step is not None:
errors.append(f"Edit {idx}: 'remove' must not include a step.")
continue
if not isinstance(edit.step, dict):
errors.append(f"Edit {idx}: '{edit.operation}' requires a step mapping.")
continue
step_id = edit.step.get("id")
if not isinstance(step_id, str) or not step_id:
errors.append(f"Edit {idx}: step is missing required 'id'.")
continue
if ":" in step_id:
errors.append(
f"Edit {idx}: step id {step_id!r} contains ':' which is reserved "
"for engine-generated nested IDs."
)
return errors

176
src/specify_cli/workflows/overlays/schema.py Normal file
View File

@@ -0,0 +1,176 @@
"""Workflow overlay schema — dataclasses and validation for overlay manifests."""
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Any, Literal
from ...extensions import normalize_priority
# Safe single-segment identifiers: no path separators, no traversal, no dots.
_SAFE_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
_RESERVED_OVERLAY_WORKFLOW_IDS: frozenset[str] = frozenset({"overlays"})
_RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"overlays", "runs", "steps"})
VALID_OPERATIONS = frozenset({"insert_after", "insert_before", "replace", "remove"})
# Map shorthand keys to operation names.
_SHORTHAND_OPERATION_KEYS: frozenset[str] = VALID_OPERATIONS
@dataclass(frozen=True)
class OverlayEdit:
"""A single edit operation on a workflow step list."""
operation: Literal["insert_after", "insert_before", "replace", "remove"]
anchor: str
step: dict[str, Any] | None = None
@dataclass
class Overlay:
"""A declared overlay (one YAML file)."""
id: str
extends: str
edits: list[OverlayEdit]
priority: int = 10
enabled: bool = True
def _validate_safe_id(
value: str,
field_name: str,
allow_reserved: bool = False,
reserved_ids: frozenset[str] = _RESERVED_OVERLAY_WORKFLOW_IDS,
) -> str | None:
"""Return an error message if *value* is not a safe path segment ID."""
if not isinstance(value, str) or not value:
return f"Overlay '{field_name}' is required and must be a non-empty string."
if not _SAFE_ID_PATTERN.fullmatch(value):
return (
f"Overlay '{field_name}' {value!r} contains invalid characters; "
"only lowercase letters, digits, and hyphens are allowed."
)
if not allow_reserved and value in reserved_ids:
return f"Overlay '{field_name}' {value!r} is reserved."
return None
def _parse_edit(edit_raw: dict[str, Any], idx: int) -> tuple[OverlayEdit | None, str | None]:
"""Parse a single edit dict into an OverlayEdit or an error string."""
shorthand_keys = [key for key in _SHORTHAND_OPERATION_KEYS if key in edit_raw]
has_operation = "operation" in edit_raw
operation: str | None = None
anchor: Any = None
if shorthand_keys and has_operation:
return None, (
f"Edit at index {idx} mixes shorthand operation key "
f"({shorthand_keys[0]!r}) with explicit 'operation' field."
)
if len(shorthand_keys) > 1:
return None, (
f"Edit at index {idx} has multiple operation keys: "
f"{', '.join(repr(k) for k in shorthand_keys)}."
)
if shorthand_keys:
operation = shorthand_keys[0]
anchor = edit_raw[operation]
elif has_operation:
operation = edit_raw.get("operation")
anchor = edit_raw.get("anchor")
else:
return None, f"Edit at index {idx} has no operation; expected one of {sorted(VALID_OPERATIONS)}."
if operation not in VALID_OPERATIONS:
return None, f"Edit at index {idx} has invalid operation {operation!r}."
if not isinstance(anchor, str) or not anchor:
return None, f"Edit at index {idx} has invalid 'anchor'."
step = edit_raw.get("step")
if operation == "remove":
if step is not None:
return None, f"Edit at index {idx} ('remove') must not include 'step'."
return OverlayEdit(operation=operation, anchor=anchor), None
if not isinstance(step, dict):
return None, f"Edit at index {idx} ('{operation}') requires 'step' mapping."
step_id = step.get("id")
if not isinstance(step_id, str) or not step_id:
return None, f"Edit at index {idx} step is missing required 'id'."
if ":" in step_id:
return None, (
f"Edit at index {idx} step id {step_id!r} contains ':' "
"which is reserved for engine-generated nested IDs."
)
return OverlayEdit(operation=operation, anchor=anchor, step=step), None
def validate_overlay_yaml(data: dict[str, Any]) -> tuple[Overlay | None, list[str]]:
"""Validate an overlay manifest dict and return (Overlay, errors).
Errors are returned as a list of strings; validation never raises.
"""
errors: list[str] = []
if not isinstance(data, dict):
return None, ["Overlay manifest must be a mapping."]
overlay_id = data.get("id")
if err := _validate_safe_id(overlay_id, "id"):
errors.append(err)
overlay_id = ""
extends = data.get("extends")
if err := _validate_safe_id(
extends,
"extends",
reserved_ids=_RESERVED_WORKFLOW_IDS,
):
errors.append(err)
extends = ""
priority = normalize_priority(data.get("priority", 10))
edits_raw = data.get("edits")
edits: list[OverlayEdit] = []
if not isinstance(edits_raw, list):
errors.append("Overlay 'edits' is required and must be a list.")
elif not edits_raw:
errors.append("Overlay 'edits' must be a non-empty list.")
else:
for idx, edit_raw in enumerate(edits_raw):
if not isinstance(edit_raw, dict):
errors.append(f"Edit at index {idx} must be a mapping.")
continue
edit, err = _parse_edit(edit_raw, idx)
if err:
errors.append(err)
continue
if edit is not None:
edits.append(edit)
enabled = data.get("enabled", True)
if not isinstance(enabled, bool):
errors.append("Overlay 'enabled' must be a boolean.")
enabled = bool(enabled)
if errors:
return None, errors
return (
Overlay(
id=overlay_id,
extends=extends,
priority=priority,
edits=edits,
enabled=enabled,
),
[],
)