Compare commits

..
1 Commits
Author SHA1 Message Date
github-actions[bot] 489a3d51d1 chore: bump version to 0.15.1 2026-07-31 13:38:08 +00:00
34 changed files with 187 additions and 2384 deletions
-25
View File
@@ -2,31 +2,6 @@
<!-- insert new changelog below this comment -->
## [0.15.2] - 2026-08-03
### Changed
- fix(presets): restore core skills instead of deleting them on preset remove (#3929)
- fix(manifests): reject non-string metadata instead of crashing on it (#3943)
- fix: narrow bare except Exception in invoke separator resolution (#3856)
- fix(workflows): keep the init step's documented ignore_agent_tools default on an explicit null (#3889)
- fix(kimi): preserve non-UTF-8 user skills (#3895)
- fix(presets): tolerate non-UTF-8 legacy commands (#3896)
- feat: allow overriding default init integration via SPECKIT_INTEGRATION_DEFAULT (#3952)
- Add adrkit extension to community catalog (#3947)
- feat(extensions): scaffold config templates on extension add/enable (#2000)
- fix(events): skip non-UTF-8 extension manifests (#3900)
- fix(workflows): fail a gate whose on_reject is not abort/skip/retry (#3888)
- fix(presets): validate required manifest mappings (#3898)
- fix: eliminate TOCTOU race in zip packaging (#3855)
- fix(workflows): fail a fan-in step whose output is not a mapping (#3887)
- fix(workflows): refetch non-UTF-8 catalog caches (#3901)
- fix(bundler): wrap local catalog decode failures (#3902)
- Add `--extension` flag to `specify init` for opting into extensions at init time (#3914)
- fix: bound response reads in extension catalog and download (#3775)
- fix(workflows): reject a retry gate whose verdict enum forbids the reset value (#3912)
- chore: release 0.15.1, begin 0.15.2.dev0 development (#3913)
## [0.15.1] - 2026-07-31
### Changed
-1
View File
@@ -25,7 +25,6 @@ The following community-contributed extensions are available in [`catalog.commun
| Extension | Purpose | Category | Effect | URL |
|-----------|---------|----------|--------|-----|
| adrkit — decision memory for spec-driven development | Pulls the decisions governing this work into agent context, checks produced plans against them, and drafts an ADR from a plan artifact | `process` | Read+Write | [adrkit](https://github.com/mbeacom/adrkit) |
| Agent Assign | Assign specialized Claude Code agents to spec-kit tasks for targeted execution | `process` | Read+Write | [spec-kit-agent-assign](https://github.com/xymelon/spec-kit-agent-assign) |
| Agent Governance | Generate agent-platform repository governance files from Spec Kit metadata | `process` | Read+Write | [spec-kit-agent-governance](https://github.com/bigsmartben/spec-kit-agent-governance) |
| AI-Driven Engineering (AIDE) | A structured 7-step workflow for building new projects from scratch with AI assistants — from vision through implementation | `process` | Read+Write | [aide](https://github.com/mnriem/spec-kit-extensions/tree/main/aide) |
+1 -2
View File
@@ -25,7 +25,7 @@ Creates a new Spec Kit project with the necessary directory structure, templates
Use `<project_name>` to create a new directory, or `--here` (or `.`) to initialize in the current directory. If the directory already has files, use `--force` to merge without confirmation.
When `--integration` is omitted, interactive terminals prompt you to choose an integration. Non-interactive sessions, such as CI or piped runs, default to GitHub Copilot; pass `--integration <key>` to choose a different integration explicitly, or set `SPECKIT_INTEGRATION_DEFAULT` to change the fallback (see [Environment Variables](#environment-variables)).
When `--integration` is omitted, interactive terminals prompt you to choose an integration. Non-interactive sessions, such as CI or piped runs, default to GitHub Copilot; pass `--integration <key>` to choose a different integration explicitly.
### Examples
@@ -50,7 +50,6 @@ specify init my-project --integration copilot --preset compliance
| Variable | Description |
| ----------------- | ------------------------------------------------------------------------ |
| `SPECKIT_INTEGRATION_DEFAULT` | Override the fallback integration used by `specify init` when `--integration` is omitted (interactive prompt default and non-interactive fallback). Set it to any registered integration key (e.g. `gemini`, `claude`). An unrecognized value is ignored with a warning and the built-in default (`copilot`) is used. An explicit `--integration <key>` always takes precedence. |
| `SPECIFY_INIT_DIR` | Target a member project from outside its directory (e.g. a monorepo root) without `cd`, for non-interactive / CI use. Set it to the **project root** — the directory *containing* `.specify/` (relative paths resolve against the current directory). The path must exist and contain `.specify/`, otherwise the command errors and does **not** fall back to the current directory. Resolved once in the core root helper (`get_repo_root` in Bash, `Get-RepoRoot` in PowerShell), so it is honored by the core feature scripts (`/speckit.plan`, `/speckit.tasks`, …) and the Git extension's feature-branch creation, which inherit it. The `specify` CLI applies the **same** validation rules to every project-scoped subcommand (`specify integration …`, `specify extension …`, `specify workflow …`, `specify preset …`, and the rest that operate on a `.specify/` project), so those can target a member project too. When unset, Bash/PowerShell helpers keep their existing upward search; the `specify` CLI keeps its project-scoped resolver cwd-only unless a command explicitly defines broader detection (for example, bundle commands). |
| `SPECIFY_FEATURE_DIRECTORY` | Override the active feature directory *within* the resolved project (takes precedence over `.specify/feature.json`). Relative paths resolve under the project root. Combine with `SPECIFY_INIT_DIR` to pick both the project and the feature non-interactively. |
| `SPECIFY_FEATURE` | Override feature detection for non-Git repositories. Set to the feature directory name (e.g., `001-photo-albums`) to work on a specific feature when not using Git branches. Must be set in the context of the agent prior to using `/speckit.plan` or follow-up commands. |
-13
View File
@@ -623,19 +623,6 @@ pauses: the named stored input is reset to `""`. A later resume therefore
prompts or pauses again until another verdict is supplied. Approve, abort, and
skip outcomes leave the input unchanged.
Because of that reset, a verdict input used with `on_reject: retry` must accept
`""`. If it declares an `enum`, include the empty string — otherwise the reset
value violates the input's own `enum` and the run can no longer be resumed with
any input. `specify workflow add` reports this as a validation error.
```yaml
inputs:
spec_verdict:
type: string
enum: ["", approve, reject]
default: ""
```
## FAQ
### What happens when a workflow hits a gate step?
+1 -36
View File
@@ -1,43 +1,8 @@
{
"schema_version": "1.0",
"updated_at": "2026-08-03T00:00:00Z",
"updated_at": "2026-07-29T00:00:00Z",
"catalog_url": "https://raw.githubusercontent.com/github/spec-kit/main/extensions/catalog.community.json",
"extensions": {
"adrkit": {
"name": "adrkit — decision memory for spec-driven development",
"id": "adrkit",
"description": "Pulls the decisions governing this work into agent context, checks produced plans against them, and drafts an ADR from a plan artifact.",
"author": "Mark Beacom (@mbeacom)",
"version": "0.1.2",
"download_url": "https://github.com/mbeacom/adrkit/releases/download/spec-kit-v0.1.2/adrkit.zip",
"repository": "https://github.com/mbeacom/adrkit",
"homepage": "https://adrkit.dev",
"documentation": "https://github.com/mbeacom/adrkit/blob/main/packages/adapters/spec-kit/README.md",
"changelog": "https://github.com/mbeacom/adrkit/blob/main/CHANGELOG.md",
"license": "Apache-2.0",
"category": "process",
"effect": "read-write",
"requires": {
"speckit_version": ">=0.13.0,<0.16.0",
"tools": [{ "name": "adr", "version": ">=0.3.0", "required": true }]
},
"provides": {
"commands": 3,
"hooks": 1
},
"tags": [
"adr",
"governance",
"decision-records",
"architecture",
"compliance"
],
"verified": false,
"downloads": 0,
"stars": 0,
"created_at": "2026-08-03T00:00:00Z",
"updated_at": "2026-08-03T00:00:00Z"
},
"aide": {
"name": "AI-Driven Engineering (AIDE)",
"id": "aide",
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "specify-cli"
version = "0.15.2"
version = "0.15.1"
description = "Specify CLI, part of GitHub Spec Kit. A tool to bootstrap your projects for Spec-Driven Development (SDD)."
readme = "README.md"
requires-python = ">=3.11"
-2
View File
@@ -77,9 +77,7 @@ from ._version import (
from ._agent_config import (
AGENT_CONFIG as AGENT_CONFIG,
DEFAULT_INIT_INTEGRATION as DEFAULT_INIT_INTEGRATION,
DEFAULT_INIT_INTEGRATION_ENV_VAR as DEFAULT_INIT_INTEGRATION_ENV_VAR,
SCRIPT_TYPE_CHOICES as SCRIPT_TYPE_CHOICES,
resolve_default_init_integration as resolve_default_init_integration,
)
from ._init_options import (
INIT_OPTIONS_FILE as INIT_OPTIONS_FILE,
-32
View File
@@ -1,8 +1,6 @@
"""Agent configuration constants derived from the integration registry."""
from __future__ import annotations
import os
import sys
from typing import Any
@@ -19,36 +17,6 @@ AGENT_CONFIG: dict[str, dict[str, Any]] = _build_agent_config()
DEFAULT_INIT_INTEGRATION = "copilot"
#: Environment variable used to override the fallback integration that
#: ``specify init`` selects in non-interactive sessions. Follows the existing
#: ``SPECKIT_INTEGRATION_*`` namespace (see ``SPECKIT_INTEGRATION_<KEY>_EXECUTABLE``
#: and ``SPECKIT_INTEGRATION_CATALOG_URL``).
DEFAULT_INIT_INTEGRATION_ENV_VAR = "SPECKIT_INTEGRATION_DEFAULT"
def resolve_default_init_integration() -> str:
"""Return the default init integration, honoring an env-var override.
Reads :data:`DEFAULT_INIT_INTEGRATION_ENV_VAR`
(``SPECKIT_INTEGRATION_DEFAULT``). When it names a registered integration
key, that key is returned; otherwise the hardcoded
:data:`DEFAULT_INIT_INTEGRATION` (``"copilot"``) is used. An invalid value
emits a warning to stderr rather than silently falling back, so operators
can tell a typo from an intentional default.
"""
override = (os.environ.get(DEFAULT_INIT_INTEGRATION_ENV_VAR) or "").strip()
if not override:
return DEFAULT_INIT_INTEGRATION
if override in AGENT_CONFIG:
return override
print(
f"Warning: {DEFAULT_INIT_INTEGRATION_ENV_VAR}='{override}' is not a "
f"recognized integration; falling back to '{DEFAULT_INIT_INTEGRATION}'. "
f"Choose from: {', '.join(sorted(AGENT_CONFIG.keys()))}.",
file=sys.stderr,
)
return DEFAULT_INIT_INTEGRATION
SCRIPT_TYPE_CHOICES: dict[str, str] = {
"sh": "POSIX Shell (bash/zsh)",
"ps": "PowerShell",
+1 -1
View File
@@ -679,7 +679,7 @@ class CommandRegistrar:
_integ = get_integration(agent_name)
if _integ is not None:
_sep = _integ.invoke_separator_for_mode(registrar_writes_skills)
except (ImportError, ValueError, KeyError):
except Exception:
pass
_prefix = get_invocation_prefix(agent_name, registrar_writes_skills)
+3 -3
View File
@@ -18,7 +18,7 @@ from urllib.request import url2pathname
from ..._assets import _locate_core_pack, _repo_root
from ..._download_security import MAX_JSON_CATALOG_BYTES, read_response_limited
from .. import BundlerError
from ..lib.yamlio import load_json, loads_json
from ..lib.yamlio import loads_json
from ..models.catalog import CatalogSource
from ..models.manifest import ComponentRef
@@ -145,13 +145,13 @@ def make_catalog_fetcher(*, allow_network: bool = True):
path = _file_url_to_path(parsed)
if not path.exists():
raise BundlerError(f"Catalog file not found: {path}")
return load_json(path)
return loads_json(path.read_text(encoding="utf-8"), origin=str(path))
if scheme == "" or _is_windows_drive_path(url):
path = Path(url)
if not path.exists():
raise BundlerError(f"Catalog file not found: {path}")
return load_json(path)
return loads_json(path.read_text(encoding="utf-8"), origin=str(path))
if scheme in ("http", "https"):
if not allow_network:
+3 -5
View File
@@ -93,11 +93,9 @@ def build_bundle(
# extraction, but collapse to two canonical modes (0755 when any
# execute bit is set on the source, otherwise 0644) so identical
# inputs yield a byte-for-byte identical artifact.
with file_path.open("rb") as fh:
st = os.fstat(fh.fileno())
mode = 0o755 if st.st_mode & 0o111 else 0o644
info.external_attr = mode << 16
archive.writestr(info, fh.read())
mode = 0o755 if file_path.stat().st_mode & 0o111 else 0o644
info.external_attr = mode << 16
archive.writestr(info, file_path.read_bytes())
return BuildResult(artifact_path=artifact_path, file_count=len(files))
+2 -4
View File
@@ -119,8 +119,6 @@ def _run_init(integration: str, *, script_type: str, offline: bool = False) -> N
preset=None,
integration=integration,
integration_options=None,
extensions=None,
trust_extension_urls=False,
)
except typer.Exit as exc:
if exc.exit_code:
@@ -131,13 +129,13 @@ def _run_init(integration: str, *, script_type: str, offline: bool = False) -> N
def _resolve_init_integration(override: str | None, manifest) -> str:
"""Precedence (FR-013): explicit override → bundle-declared → default."""
from ..._agent_config import resolve_default_init_integration
from ..._agent_config import DEFAULT_INIT_INTEGRATION
if override:
return override
if manifest is not None and manifest.integration is not None:
return manifest.integration.id
return resolve_default_init_integration()
return DEFAULT_INIT_INTEGRATION
# ===== Consume =====
+5 -220
View File
@@ -14,8 +14,8 @@ from rich.panel import Panel
from .._agent_config import (
AGENT_CONFIG,
DEFAULT_INIT_INTEGRATION,
SCRIPT_TYPE_CHOICES,
resolve_default_init_integration,
)
from .._assets import (
_locate_bundled_preset,
@@ -30,145 +30,6 @@ def _stdin_is_interactive() -> bool:
return sys.stdin.isatty()
def _ext_spec_is_url(ext_spec: str) -> bool:
"""Return True when *ext_spec* is an http(s) URL rather than a name/path."""
from urllib.parse import urlparse
try:
return urlparse(ext_spec).scheme in ("http", "https")
except ValueError:
return False
def _confirm_extension_url_trust(
url_specs: list[str], *, trust_override: bool
) -> dict[str, bool]:
"""Resolve trust for each URL-based extension before the Live display.
URL installs pull an arbitrary external extension, so they get the same
default-deny confirmation as ``extension add --from``. Returns a mapping of
``url_spec -> approved``. With *trust_override* every URL is pre-approved.
In a non-interactive session without the override, every URL is denied
(the prompt cannot be answered), mirroring the default-deny posture.
"""
from rich.markup import escape as _escape_markup
from rich.panel import Panel
approvals: dict[str, bool] = {}
interactive = _stdin_is_interactive()
for spec in url_specs:
if trust_override:
approvals[spec] = True
continue
if not interactive:
approvals[spec] = False
continue
console.print()
console.print(
Panel(
"[bold]You are installing an extension from an external URL that is not\n"
"listed in any of your configured extension catalogs.[/bold]\n\n"
f"URL: {_escape_markup(spec)}\n\n"
"Only install extensions from sources you trust.",
title="[bold yellow]⚠ Untrusted Source[/bold yellow]",
border_style="yellow",
padding=(1, 2),
)
)
console.print()
approvals[spec] = typer.confirm(
f"Install extension from {spec}?", default=False
)
return approvals
def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_version: str) -> str:
"""Install a single extension during ``specify init``.
Handles bundled extension names, local directory paths, and HTTPS URLs.
Returns a short status message on success.
Raises ``ValueError`` on failure so the caller can convert it to a
tracker error without aborting the entire init.
"""
from urllib.parse import urlparse
from .._assets import _locate_bundled_extension
from ..extensions import ExtensionCatalog, ExtensionError, ExtensionManager
from ..extensions._commands import (
_resolve_catalog_extension,
install_extension_from_url,
)
manager = ExtensionManager(project_path)
# --- URL ---
parsed = urlparse(ext_spec)
if parsed.scheme in ("http", "https"):
try:
manifest = install_extension_from_url(
manager, project_path, ext_spec, speckit_version
)
except ExtensionError as exc:
raise ValueError(str(exc)) from exc
return f"{manifest.name} v{manifest.version} installed"
# --- Local path ---
if ext_spec.startswith(("./", "../", "/", "~/", ".\\", "..\\")) or Path(ext_spec).is_absolute():
source_path = Path(ext_spec).expanduser().resolve()
if not source_path.exists():
raise ValueError(f"Directory not found: {source_path}")
if not (source_path / "extension.yml").exists():
raise ValueError(f"No extension.yml found in {source_path}")
manifest = manager.install_from_directory(source_path, speckit_version)
return f"{manifest.name} v{manifest.version} installed"
# --- Bundled extension name or catalog ID ---
bundled_path = _locate_bundled_extension(ext_spec)
if bundled_path is not None:
if manager.registry.is_installed(ext_spec):
return "already installed"
manifest = manager.install_from_directory(bundled_path, speckit_version)
return f"{manifest.name} v{manifest.version} installed"
# Fall back to catalog
catalog = ExtensionCatalog(project_path)
ext_info, catalog_error = _resolve_catalog_extension(ext_spec, catalog, "add")
if catalog_error:
raise ValueError(f"Could not query extension catalog: {catalog_error}")
if not ext_info:
raise ValueError(f"Extension '{ext_spec}' not found in bundled extensions or catalog")
resolved_id = ext_info["id"]
if resolved_id != ext_spec:
bundled_path = _locate_bundled_extension(resolved_id)
if bundled_path is not None:
if manager.registry.is_installed(resolved_id):
return "already installed"
manifest = manager.install_from_directory(bundled_path, speckit_version)
return f"{manifest.name} v{manifest.version} installed"
if ext_info.get("bundled") and not ext_info.get("download_url"):
from ..extensions import REINSTALL_COMMAND
raise ValueError(
f"Extension '{resolved_id}' is bundled with spec-kit but not found in the installed package. "
f"Try reinstalling spec-kit: {REINSTALL_COMMAND}"
)
if not ext_info.get("_install_allowed", True):
catalog_name = ext_info.get("_catalog_name", "community")
raise ValueError(
f"Extension '{ext_spec}' is in the '{catalog_name}' catalog but installation is not allowed from that catalog"
)
zip_path = catalog.download_extension(resolved_id)
try:
manifest = manager.install_from_zip(zip_path, speckit_version)
finally:
zip_path.unlink(missing_ok=True)
return f"{manifest.name} v{manifest.version} installed"
def ensure_constitution_from_template(
project_path: Path, tracker: StepTracker | None = None
) -> None:
@@ -281,16 +142,6 @@ def register(app: typer.Typer) -> None:
"--integration-options",
help='Options for the integration (e.g. --integration-options="--commands-dir .myagent/cmds")',
),
extensions: list[str] | None = typer.Option(
None,
"--extension",
help="Install an extension during initialization (bundled name, local path, or HTTPS URL). Repeatable.",
),
trust_extension_urls: bool = typer.Option(
False,
"--trust-extension-urls",
help="Pre-authorize installing extensions from external URLs without the interactive trust prompt (required for non-interactive URL installs).",
),
):
"""
Initialize a new Specify project.
@@ -323,10 +174,6 @@ def register(app: typer.Typer) -> None:
specify init --here --integration gemini
specify init my-project --integration generic --integration-options="--commands-dir .myagent/commands/" # Bring your own agent; requires --commands-dir
specify init my-project --integration claude --preset healthcare-compliance # With preset
specify init my-project --integration copilot --extension git # With bundled extension
specify init my-project --extension git --extension selftest # Multiple extensions
specify init my-project --extension ./my-extensions/custom-ext # Local path extension
specify init my-project --extension https://example.com/extensions/my-ext.zip --trust-extension-urls # URL extension (non-interactive)
"""
# Lazy imports to avoid circular dependency — __init__.py imports this module
from .. import (
@@ -466,18 +313,17 @@ def register(app: typer.Typer) -> None:
raise typer.Exit(1)
selected_ai = integration
elif not _stdin_is_interactive():
default_integration = resolve_default_init_integration()
console.print(
f"[dim]Non-interactive session detected: defaulting to '{default_integration}'. "
f"[dim]Non-interactive session detected: defaulting to '{DEFAULT_INIT_INTEGRATION}'. "
"Use --integration to choose a different agent.[/dim]"
)
selected_ai = default_integration
selected_ai = DEFAULT_INIT_INTEGRATION
else:
ai_choices = {key: config["name"] for key, config in AGENT_CONFIG.items()}
selected_ai = select_with_arrows(
ai_choices,
"Choose your coding agent integration:",
resolve_default_init_integration(),
DEFAULT_INIT_INTEGRATION,
)
if not integration:
@@ -567,31 +413,10 @@ def register(app: typer.Typer) -> None:
("chmod", "Ensure scripts executable"),
("constitution", "Constitution setup"),
("workflow", "Install bundled workflow"),
("final", "Finalize"),
]:
tracker.add(key, label)
if extensions:
from rich.markup import escape as _escape_markup
for i, ext_spec in enumerate(extensions):
tracker.add(
f"extension-{i}", f"Install extension: {_escape_markup(ext_spec)}"
)
tracker.add("final", "Finalize")
# Resolve trust for URL-based extensions BEFORE entering the Live
# display: the confirmation prompt cannot be shown/answered underneath
# the Rich Live spinner. URL installs are default-deny unless the user
# confirms interactively or passes --trust-extension-urls.
extension_url_approvals: dict[str, bool] = {}
if extensions:
url_specs = [e for e in extensions if _ext_spec_is_url(e)]
if url_specs:
extension_url_approvals = _confirm_extension_url_trust(
url_specs, trust_override=trust_extension_urls
)
# Disable transient mode on Windows: PowerShell 5.1's legacy console
# hangs when Rich tries to restore cursor state via VT escape sequences.
_transient = sys.platform != "win32"
@@ -801,46 +626,6 @@ def register(app: typer.Typer) -> None:
continuing="Continuing without the optional preset.",
)
# Install extensions specified via --extension
if extensions:
from rich.markup import escape as _escape_markup
from ..extensions._commands import _refresh_events_and_warn
speckit_ver = get_speckit_version()
any_extension_installed = False
for i, ext_spec in enumerate(extensions):
tracker.start(f"extension-{i}")
# Skip URL extensions the user did not confirm as trusted
# (default-deny; resolved before the Live display).
if _ext_spec_is_url(ext_spec) and not extension_url_approvals.get(
ext_spec, False
):
tracker.error(
f"extension-{i}",
"skipped: untrusted URL not confirmed "
"(use --trust-extension-urls)",
)
continue
try:
status_msg = _install_extension_during_init(
project_path, ext_spec, speckit_ver
)
tracker.complete(f"extension-{i}", status_msg)
any_extension_installed = True
except Exception as ext_err:
sanitized_ext = str(ext_err).replace("\n", " ").strip()
tracker.error(
f"extension-{i}",
f"failed: {_escape_markup(sanitized_ext[:120])}",
)
# Refresh native event configuration once after the batch so
# that an extension declaring ``events:`` has its hooks
# activated, mirroring the ``extension add`` path.
if any_extension_installed:
_refresh_events_and_warn(project_path)
# Seed the constitution AFTER preset installation so that a
# preset-provided constitution-template (resolved via the
# priority stack) wins over the core template.
+1 -1
View File
@@ -895,7 +895,7 @@ def collect_extension_events(project_root: Path) -> ResolvedEvents:
continue
try:
data = yaml.safe_load(ext_yml.read_text(encoding="utf-8")) or {}
except (UnicodeDecodeError, yaml.YAMLError):
except yaml.YAMLError:
continue
if not isinstance(data, dict):
continue
-177
View File
@@ -285,25 +285,9 @@ class ExtensionManifest:
raise ValidationError(
f"Invalid extension: expected a mapping, got {type(ext).__name__}"
)
# Check presence AND type: the format/version checks below feed these
# values straight to ``re.match`` and ``packaging.Version``, both of
# which raise a bare TypeError on a non-string. YAML makes that an easy
# authoring slip -- unquoted ``version: 1.0`` parses as a float and
# ``id: 2`` as an int -- and TypeError is not a ValidationError, so it
# escapes every caller that already handles a malformed manifest (see
# list_installed()'s "Corrupted extension" fallback, which catches
# ValidationError only, making one bad extension exit ``specify
# extension list`` with a raw traceback and hide the healthy ones).
# Mirrors the sibling IntegrationDescriptor, which already type-checks
# the same four fields.
for field in ["id", "name", "version", "description"]:
if field not in ext:
raise ValidationError(f"Missing extension.{field}")
if not isinstance(ext[field], str):
raise ValidationError(
f"Invalid extension.{field}: expected a string, "
f"got {type(ext[field]).__name__}"
)
# Validate extension ID format
if not re.match(r"^[a-z0-9-]+$", ext["id"]):
@@ -407,16 +391,6 @@ class ExtensionManifest:
)
if "name" not in cmd or "file" not in cmd:
raise ValidationError("Command missing 'name' or 'file'")
# The pattern match below would raise a bare TypeError on a
# non-string name (``name: 2``), escaping the ValidationError
# contract. The 'file' field needs no check here:
# relative_extension_path_violation() below already rejects a
# non-string value.
if not isinstance(cmd["name"], str):
raise ValidationError(
f"Invalid command name: expected a string, "
f"got {type(cmd['name']).__name__}"
)
# Validate the 'file' field at manifest-load time using the single
# shared policy in relative_extension_path_violation(), so manifest
@@ -588,14 +562,6 @@ class ExtensionManifest:
"""Get list of provided commands."""
return self.data.get("provides", {}).get("commands", [])
@property
def config(self) -> List[Dict[str, Any]]:
"""Get list of provided config templates, normalized to dictionaries."""
raw = self.data.get("provides", {}).get("config", [])
if not isinstance(raw, list) or not all(isinstance(entry, dict) for entry in raw):
return []
return raw
@property
def hooks(self) -> Dict[str, Any]:
"""Get hook definitions."""
@@ -2523,149 +2489,6 @@ class ExtensionManager:
extension_dir, speckit_version, priority=priority, force=force
)
def _config_root_is_contained(self, specify_dir: Path) -> bool:
"""Report whether `.specify` is a real directory inside the project.
Checked component by component so a symlink anywhere on the path is
rejected before it becomes the containment root. A missing `.specify`
is fine: scaffolding creates it under the project root.
"""
try:
root = self.project_root.resolve()
except OSError:
return False
current = self.project_root
for part in specify_dir.relative_to(self.project_root).parts:
current = current / part
if current.is_symlink():
return False
if not current.exists():
return True
try:
if current.resolve().relative_to(root) is None:
return False
except (OSError, ValueError):
return False
return current.is_dir()
@staticmethod
def _target_follows_preserved_convention(target_name: str) -> bool:
"""True when a scaffold target survives remove/backup/restore.
Those paths only handle top-level ``*-config.yml`` and
``*-config.local.yml`` files, so anything nested or otherwise named is
not preserved across an update.
"""
if "/" in target_name or "\\" in target_name:
return False
return target_name.endswith("-config.yml") or target_name.endswith(
"-config.local.yml"
)
def scaffold_config(self, extension_id: str) -> tuple[List[str], List[str], List[str]]:
"""Deploy config templates from an installed extension to the project.
Reads the extension's manifest provides.config section and copies
each config template to the project's .specify/ directory. Existing
config files are never overwritten (user customizations are preserved).
Args:
extension_id: ID of the installed extension
Returns:
Tuple of (deployed, skipped_existing, failed) where each is a list
of config file names.
"""
ext_dir = self.extensions_dir / extension_id
manifest_path = ext_dir / "extension.yml"
if not manifest_path.exists():
return [], [], []
manifest = ExtensionManifest(manifest_path)
deployed = []
skipped_existing = []
failed = []
provides = manifest.data.get("provides", {})
raw_config = provides.get("config", [])
config_is_malformed = (
"config" in provides
and (
not isinstance(raw_config, list)
or not all(isinstance(entry, dict) for entry in raw_config)
)
)
if config_is_malformed:
return deployed, skipped_existing, ["provides.config"]
ext_dir_resolved = ext_dir.resolve()
# Config is deployed beneath the extension's own directory because that
# is where it is read from: ConfigManager._get_project_config() loads
# `.specify/extensions/<id>/<id>-config.yml`, and the bundled scripts
# and READMEs use the same location. Writing to `.specify/<name>` put
# the file somewhere nothing ever looks.
config_dir = self.project_root / ".specify" / "extensions" / extension_id
# Resolving that directory and trusting the result as the containment
# root lets a symlinked component point outside the project: every
# target would then satisfy relative_to and copy2 would write
# externally. Refuse a symlinked component up front, matching the
# project safe-write path in shared_infra.
if not self._config_root_is_contained(config_dir):
return deployed, skipped_existing, ["provides.config"]
config_dir_resolved = config_dir.resolve()
for config_entry in manifest.config:
template_name = config_entry.get("template", "")
target_name = config_entry.get("name", template_name)
failure_name = target_name if isinstance(target_name, str) and target_name else "provides.config"
if not isinstance(template_name, str) or not template_name:
failed.append(failure_name)
continue
if not isinstance(target_name, str) or not target_name:
failed.append(failure_name)
continue
# Only scaffold what removal actually preserves. remove(keep_config)
# keeps top-level files ending in -config.yml / -config.local.yml and
# rmtree's every subdirectory; the backup path globs the same
# top-level pattern. A nested or differently-named target would be
# silently destroyed by `extension add --force` and replaced with the
# template default, losing the user's customization.
if not self._target_follows_preserved_convention(target_name):
failed.append(failure_name)
continue
template_candidate = ext_dir / template_name
template_path = template_candidate.resolve()
target_path = (config_dir / target_name).resolve()
try:
template_path.relative_to(ext_dir_resolved)
target_path.relative_to(config_dir_resolved)
except ValueError:
failed.append(failure_name)
continue
if template_candidate.is_symlink() or not template_path.is_file():
failed.append(failure_name)
continue
if target_path.exists():
skipped_existing.append(target_name)
continue
try:
# mkdir belongs inside the handler: a nested target like
# foo/config.yml must land in `failed` when `.specify/foo` is a
# file or cannot be created, not raise out of scaffolding after
# `extension add` has already installed the extension.
target_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(template_path, target_path)
except OSError:
failed.append(target_name)
continue
deployed.append(target_name)
return deployed, skipped_existing, failed
def install_from_zip(
self,
zip_path: Path,
+129 -195
View File
@@ -95,141 +95,6 @@ def _refresh_events_and_warn(project_root: Path) -> None:
console.print(f" {key}: {_escape_markup(detail)}")
def install_extension_from_url(
manager,
project_root: Path,
url: str,
speckit_version: str,
*,
priority: int = 10,
force: bool = False,
):
"""Download an archive from *url* and install it, reusing the hardened path.
Shares the same download hardening as ``extension add --from``:
HTTPS enforcement, the catalog's authenticated + redirect-guarded
``_open_url`` fetch, a bounded (50 MiB) response read, archive-format
detection (ZIP or tar.gz/tgz), and a TOCTOU-safe transient download file
consumed directly by ``install_from_zip``.
Returns the installed manifest. Raises ``ExtensionError`` on any failure so
callers can present a uniform message without a second downloader.
"""
import urllib.error
from . import ExtensionCatalog, ExtensionError
if not is_https_or_localhost_http(url):
raise ExtensionError(
"URL must use HTTPS (HTTP is only allowed for localhost)"
)
download_dir = _validate_safe_cache_dir(project_root)
archive_filename = f"extension-url-download-{uuid4().hex}.archive"
# Only used for diagnostic messages: the real archive is a transient inode
# (unlinked on POSIX, O_TEMPORARY on Windows) consumed via ``archive_file``
# below, so this path is never opened again.
archive_path = download_dir / archive_filename
try:
dl_catalog = ExtensionCatalog(project_root)
download_url = url
extra_headers = None
resolved_url = dl_catalog._resolve_github_release_asset_api_url(download_url)
if resolved_url:
download_url = resolved_url
extra_headers = {"Accept": "application/octet-stream"}
with dl_catalog._open_url(
download_url, timeout=60, extra_headers=extra_headers
) as response:
archive_data = read_response_limited(
response,
error_type=ExtensionError,
label=f"extension {url}",
)
final_url = (
response.geturl() if hasattr(response, "geturl") else download_url
)
content_type = (
response.getheader("Content-Type")
if hasattr(response, "getheader")
else None
)
except urllib.error.URLError as exc:
raise ExtensionError(f"Failed to download from {url}: {exc}") from exc
download_fd = -1
download_file = None
try:
try:
download_fd = _safe_open_download_zip(
project_root, download_dir, archive_filename
)
except OSError as exc:
raise ExtensionError(
f"Could not safely create download file: {exc}"
) from exc
try:
download_file = os.fdopen(download_fd, "w+b")
download_fd = -1
download_file.write(archive_data)
download_file.flush()
download_file.seek(0)
except OSError as exc:
raise ExtensionError(
f"Could not safely write download file: {exc}"
) from exc
format_source = (
final_url
if archive_format_from_name(final_url) is not None
else url
)
try:
detect_archive_format(
archive_path,
archive_file=download_file,
source_name=format_source,
content_type=content_type,
error_type=ExtensionError,
)
except ExtensionError as exc:
raise ExtensionError(
f"{url} did not return a ZIP archive or tar.gz/tgz archive "
f"(got {len(archive_data)} bytes). This usually means the request "
"was not authenticated and a login/HTML page was returned. "
"Verify the URL and configured credentials."
) from exc
# Consume the transient inode reserved above rather than reopening the
# cache pathname during extraction.
try:
return manager.install_from_zip(
archive_path,
speckit_version,
priority=priority,
force=force,
archive_file=download_file,
)
except OSError as exc:
raise ExtensionError(
f"Could not install extension from downloaded archive: {exc}"
) from exc
finally:
if download_file is not None:
try:
download_file.close()
except OSError:
pass
elif download_fd >= 0:
try:
os.close(download_fd)
except OSError:
pass
def _load_catalog_command_config(project_root: Path, config_path: Path) -> dict:
"""Load extension catalog CLI config with user-facing shape errors."""
try:
@@ -945,20 +810,134 @@ def extension_add(
)
elif from_url:
# Install from URL archive via the shared hardened downloader
# (HTTPS enforcement, authenticated redirect-guarded fetch,
# bounded read, archive-format detection, TOCTOU-safe transient
# archive). Same path used by ``specify init --extension <url>``.
console.print(f"Downloading from {safe_url}...")
manifest = install_extension_from_url(
manager,
project_root,
from_url,
speckit_version,
priority=priority,
force=force,
)
# Install from an archive URL.
import urllib.error
console.print(f"Downloading from {safe_url}...")
download_dir = _validate_safe_cache_dir(project_root)
archive_filename = f"extension-url-download-{uuid4().hex}.archive"
# Only used for diagnostic messages: the real archive is a
# transient inode (unlinked on POSIX, O_TEMPORARY on Windows)
# consumed via ``archive_file`` below, so this path is never
# opened again.
archive_path = download_dir / archive_filename
try:
# Use the catalog's authenticated fetch so configured
# credentials (incl. GitHub Enterprise Server) are applied
# and GHES release-asset URLs resolve via /api/v3 — keeping
# --from consistent with catalog-based installs.
dl_catalog = ExtensionCatalog(project_root)
download_url = from_url
extra_headers = None
resolved_url = dl_catalog._resolve_github_release_asset_api_url(download_url)
if resolved_url:
download_url = resolved_url
extra_headers = {"Accept": "application/octet-stream"}
with dl_catalog._open_url(
download_url, timeout=60, extra_headers=extra_headers
) as response:
archive_data = read_response_limited(
response,
error_type=ExtensionError,
label=f"extension {from_url}",
)
final_url = (
response.geturl()
if hasattr(response, "geturl")
else download_url
)
content_type = (
response.getheader("Content-Type")
if hasattr(response, "getheader")
else None
)
download_fd = -1
download_file = None
try:
try:
download_fd = _safe_open_download_zip(
project_root, download_dir, archive_filename
)
except OSError as exc:
console.print(
"[red]Error:[/red] Could not safely create download file: "
f"{_escape_markup(str(exc))}"
)
raise typer.Exit(1)
try:
download_file = os.fdopen(download_fd, "w+b")
download_fd = -1
download_file.write(archive_data)
download_file.flush()
download_file.seek(0)
except OSError as exc:
console.print(
"[red]Error:[/red] Could not safely write download file: "
f"{_escape_markup(str(exc))}"
)
raise typer.Exit(1)
format_source = (
final_url
if archive_format_from_name(final_url) is not None
else from_url
)
try:
detect_archive_format(
archive_path,
archive_file=download_file,
source_name=format_source,
content_type=content_type,
error_type=ExtensionError,
)
except ExtensionError:
console.print(
f"[red]Error:[/red] {safe_url} did not return a ZIP archive "
"or tar.gz/tgz archive "
f"(got {len(archive_data)} bytes). This usually means "
"the request was not authenticated and a login/HTML page was "
"returned. Verify the URL and configured credentials."
)
raise typer.Exit(1)
# Consume the transient inode reserved above rather
# than reopening the cache pathname during extraction.
try:
manifest = manager.install_from_zip(
archive_path,
speckit_version,
priority=priority,
force=force,
archive_file=download_file,
)
except OSError as exc:
console.print(
"[red]Error:[/red] Could not install extension from downloaded archive: "
f"{_escape_markup(str(exc))}"
)
raise typer.Exit(1)
finally:
if download_file is not None:
try:
download_file.close()
except OSError:
pass
elif download_fd >= 0:
try:
os.close(download_fd)
except OSError:
pass
except urllib.error.URLError as e:
console.print(
f"[red]Error:[/red] Failed to download from {safe_url}: "
f"{_escape_markup(str(e))}"
)
raise typer.Exit(1)
else:
# Try bundled extensions first (shipped with spec-kit)
bundled_path = _locate_bundled_extension(extension)
@@ -1071,29 +1050,8 @@ def extension_add(
if reg_skills:
console.print(f"\n[green]✓[/green] {len(reg_skills)} agent skill(s) auto-registered")
# Scaffold config templates automatically
deployed, skipped, failed = manager.scaffold_config(manifest.id)
config_home = f".specify/extensions/{_escape_markup(str(manifest.id))}"
if deployed:
console.print("\n[bold cyan]Config scaffolded:[/bold cyan]")
for cfg in deployed:
console.print(f" • {config_home}/{_escape_markup(str(cfg))}")
if skipped:
console.print(f"\n[dim]Config files already exist (preserved): {_escape_markup(', '.join(skipped))}[/dim]")
if failed:
console.print(
f"\n[yellow]Warning:[/yellow] Config templates not scaffolded: "
f"{_escape_markup(', '.join(failed))}. "
"Verify the extension manifest and template files."
)
# Only warn when configuration is actually unresolved. Scaffolding that
# deployed or preserved every template has already answered this, and an
# extension without provides.config has nothing to configure; the blanket
# warning contradicted the output directly above it.
if failed or not (deployed or skipped):
console.print("\n[yellow]⚠[/yellow] Configuration may be required")
console.print(f" Check: {config_home}/")
console.print("\n[yellow]⚠[/yellow] Configuration may be required")
console.print(f" Check: .specify/extensions/{_escape_markup(str(manifest.id))}/")
except ValidationError as e:
console.print(f"\n[red]Validation Error:[/red] {_escape_markup(str(e))}")
@@ -2573,30 +2531,6 @@ def extension_enable(
# are re-emitted in installed integrations.
_refresh_events_and_warn(project_root)
# Scaffold config templates on enable
try:
deployed, skipped, failed = manager.scaffold_config(extension_id)
except Exception as exc:
console.print(
f"\n[yellow]Warning:[/yellow] Failed to scaffold config for extension "
f"'{_escape_markup(str(display_name))}'."
)
console.print(f"[dim]Details: {_escape_markup(str(exc))}[/dim]")
deployed, skipped, failed = [], [], []
config_home = f".specify/extensions/{_escape_markup(str(extension_id))}"
if deployed:
console.print("\n[bold cyan]Config scaffolded:[/bold cyan]")
for cfg in deployed:
console.print(f" • {config_home}/{_escape_markup(str(cfg))}")
if skipped:
console.print(f"\n[dim]Config files already exist (preserved): {_escape_markup(', '.join(skipped))}[/dim]")
if failed:
console.print(
f"\n[yellow]Warning:[/yellow] Config templates not scaffolded: "
f"{_escape_markup(', '.join(failed))}. "
"Verify the extension manifest and template files."
)
@extension_app.command("disable")
def extension_disable(
@@ -317,7 +317,7 @@ def _is_speckit_generated_skill(skill_dir: Path) -> bool:
try:
content = skill_file.read_text(encoding="utf-8")
except (OSError, UnicodeError):
except OSError:
return False
if not content.startswith("---"):
+7 -88
View File
@@ -299,33 +299,11 @@ class PresetManifest:
f"(expected {self.SCHEMA_VERSION})"
)
for section in ("preset", "requires", "provides"):
if not isinstance(self.data[section], dict):
raise PresetValidationError(
f"Invalid {section}: expected a mapping"
)
# Validate preset metadata
pack = self.data["preset"]
# Check presence AND type: the format/version checks below feed these
# values straight to ``re.match`` and ``packaging.Version``, both of
# which raise a bare TypeError on a non-string. YAML makes that an easy
# authoring slip -- unquoted ``version: 1.0`` parses as a float and
# ``id: 2`` as an int -- and TypeError is not a PresetValidationError,
# so it escapes every caller that already handles a malformed manifest
# (see list_installed()'s "Corrupted preset" fallback, which catches
# PresetValidationError only, making one bad preset exit ``specify
# preset list`` with a raw traceback and hide the healthy ones).
# Mirrors the sibling IntegrationDescriptor, which already type-checks
# the same four fields.
for field in ["id", "name", "version", "description"]:
if field not in pack:
raise PresetValidationError(f"Missing preset.{field}")
if not isinstance(pack[field], str):
raise PresetValidationError(
f"Invalid preset.{field}: expected a string, "
f"got {type(pack[field]).__name__}"
)
# Validate pack ID format
if not re.match(r'^[a-z0-9-]+$', pack["id"]):
@@ -383,19 +361,6 @@ class PresetManifest:
"Template missing 'type', 'name', or 'file'"
)
# 'name' feeds re.match and 'file' feeds os.path.normpath below;
# both raise a bare TypeError on a non-string, which is not a
# PresetValidationError and so escapes the callers that handle a
# malformed manifest. The sibling extension manifest already
# rejects a non-string command 'file' via
# relative_extension_path_violation().
for field in ("type", "name", "file"):
if not isinstance(tmpl[field], str):
raise PresetValidationError(
f"Invalid template {field}: expected a string, "
f"got {type(tmpl[field]).__name__}"
)
if tmpl["type"] not in VALID_PRESET_TEMPLATE_TYPES:
raise PresetValidationError(
f"Invalid template type '{tmpl['type']}': "
@@ -2186,10 +2151,7 @@ class PresetManager:
]
if dir_core_ext_names:
self._unregister_skills_in_dir(
dir_core_ext_names,
skills_dir,
dir_agent,
restore_from_bundled_core=True,
dir_core_ext_names, skills_dir, dir_agent
)
for _skill_name, cmd_name, top_layer in override_skills:
@@ -3024,7 +2986,6 @@ class PresetManager:
preset_dir: Union[Path, str],
*,
additional_owned_sources: Optional[Dict[str, str]] = None,
restore_from_bundled_core: bool = False,
) -> Dict[Path, tuple[Optional[str], List[str]]]:
"""Restore original SKILL.md files after a preset is removed.
@@ -3032,17 +2993,6 @@ class PresetManager:
regenerate the skill from the core command template. If no core
template exists, the skill directory is removed.
Args:
restore_from_bundled_core: When True, a missing project-local
core template (the common case — ``specify init`` never
populates ``.specify/templates/commands``) falls back to
the bundled core_pack/repo-root templates so the skill is
restored instead of deleted (#3928). Callers that are
retiring a skill because its command now renders elsewhere
(a command file superseding it) must leave this False so
the skill is removed rather than resurrected with core
content that would duplicate the winning command.
``registered_skills`` records exactly which agent directories this
preset actually wrote to (see :meth:`_register_skills`), so removal
restores precisely those directories rather than guessing at every
@@ -3116,7 +3066,6 @@ class PresetManager:
renderer_agent,
pack_id=pack_id,
additional_owned_sources=additional_owned_sources,
restore_from_bundled_core=restore_from_bundled_core,
)
if mutated_names:
restored[skills_dir] = (
@@ -3149,7 +3098,6 @@ class PresetManager:
selected_ai,
pack_id=pack_id,
additional_owned_sources=additional_owned_sources,
restore_from_bundled_core=restore_from_bundled_core,
)
return (
{skills_dir: (selected_ai, mutated_names)}
@@ -3223,7 +3171,6 @@ class PresetManager:
*,
pack_id: Optional[str] = None,
additional_owned_sources: Optional[Dict[str, str]] = None,
restore_from_bundled_core: bool = False,
) -> List[str]:
"""Restore original SKILL.md files within a single skills directory.
@@ -3234,7 +3181,6 @@ class PresetManager:
placeholder resolution and argument-hint formatting.
additional_owned_sources: Generated non-preset source markers
accepted as owned for specific skill names.
restore_from_bundled_core: See ``_unregister_skills``.
Returns:
Skill names whose files were restored or removed.
@@ -3310,34 +3256,9 @@ class PresetManager:
if current_source not in owned_sources:
continue
extension_restore = extension_restore_index.get(skill_name)
# Try to find the core command template. Project-local overrides
# in core_templates_dir take precedence, but that directory is
# rarely populated — the real core commands ship in the bundled
# core_pack (wheel install) or the repo-root templates/ tree
# (source checkout). Callers that want a genuine restore (a
# preset was removed outright, not superseded by another
# renderer) opt into that fallback via restore_from_bundled_core
# so the skill is restored instead of deleted (#3928). An
# installed extension providing a core-named command resolves
# ahead of bundled core elsewhere, so skip the bundled fallback
# when an extension restore exists — otherwise it would win
# over the higher-priority extension layer below.
core_file = core_templates_dir / f"{short_name}.md"
if (
not core_file.exists()
and restore_from_bundled_core
and extension_restore is None
):
from .. import _locate_core_pack, _repo_root
_core_pack = _locate_core_pack()
if _core_pack is not None:
core_file = _core_pack / "commands" / f"{short_name}.md"
else:
core_file = _repo_root() / "templates" / "commands" / f"{short_name}.md"
if not core_file.exists():
# Try to find the core command template
core_file = core_templates_dir / f"{short_name}.md" if core_templates_dir.exists() else None
if core_file and not core_file.exists():
core_file = None
if core_file:
@@ -3382,6 +3303,7 @@ class PresetManager:
mutated_names.append(skill_name)
continue
extension_restore = extension_restore_index.get(skill_name)
if extension_restore:
content = extension_restore["source_file"].read_text(encoding="utf-8")
frontmatter, body = registrar.parse_frontmatter(content)
@@ -3520,9 +3442,7 @@ class PresetManager:
"registered_skills", registered_skills
)
if persisted_skills:
self._unregister_skills(
persisted_skills, dest_dir, restore_from_bundled_core=True
)
self._unregister_skills(persisted_skills, dest_dir)
try:
if dest_dir.exists():
shutil.rmtree(dest_dir)
@@ -3860,7 +3780,6 @@ class PresetManager:
restorable_skills,
pack_dir,
additional_owned_sources=override_sources,
restore_from_bundled_core=True,
)
try:
from ..agents import CommandRegistrar
@@ -5372,7 +5291,7 @@ class PresetResolver:
fm_strategy = fm_data.get("strategy")
if isinstance(fm_strategy, str) and fm_strategy.lower() in VALID_PRESET_STRATEGIES:
strategy = fm_strategy.lower()
except (UnicodeDecodeError, yaml.YAMLError, OSError):
except (yaml.YAMLError, OSError):
# Best-effort legacy frontmatter parsing: keep default
# strategy ("replace") when content is unreadable/invalid.
pass
+2 -2
View File
@@ -514,7 +514,7 @@ class WorkflowCatalog:
cached = json.load(f)
if isinstance(cached, dict):
return cached
except (UnicodeDecodeError, json.JSONDecodeError, OSError):
except (json.JSONDecodeError, OSError):
# Ignore invalid/unreadable cache and fall back to fetching from source.
pass
@@ -1210,7 +1210,7 @@ class StepCatalog:
cached = json.load(f)
if isinstance(cached, dict):
return cached
except (UnicodeDecodeError, json.JSONDecodeError, OSError):
except (json.JSONDecodeError, OSError):
# Ignore invalid/unreadable cache and fall back to fetching from source.
pass
+18 -47
View File
@@ -308,16 +308,15 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
errors.append("Workflow has no steps defined.")
seen_ids: set[str] = set()
# ``input_defs`` maps declared workflow input names to their definitions —
# used by ``_validate_steps`` to cross-reference gate ``verdict_input``
# bindings (both that the name exists and that its ``enum`` permits the
# reset sentinel). ``None`` means the inputs block itself is malformed
# (already reported above); the cross-check is then disabled so one
# authoring mistake does not cascade into N spurious "undeclared" errors.
input_defs: dict[str, Any] | None = (
dict(definition.inputs) if isinstance(definition.inputs, dict) else None
# ``input_names`` is the set of declared workflow input names — used by
# ``_validate_steps`` to cross-reference gate ``verdict_input`` bindings.
# ``None`` means the inputs block itself is malformed (already reported
# above); the cross-check is then disabled so one authoring mistake does
# not cascade into N spurious "undeclared" errors.
input_names: set[str] | None = (
set(definition.inputs) if isinstance(definition.inputs, dict) else None
)
_validate_steps(definition.steps, seen_ids, errors, input_defs)
_validate_steps(definition.steps, seen_ids, errors, input_names)
return errors
@@ -326,15 +325,15 @@ def _validate_steps(
steps: list[dict[str, Any]],
seen_ids: set[str],
errors: list[str],
input_defs: dict[str, Any] | None = None,
input_names: set[str] | None = None,
inside_fan_out: bool = False,
) -> None:
"""Recursively validate a list of steps.
``input_defs`` maps declared workflow input names to their definitions (or
is ``None`` when the inputs block is malformed). ``inside_fan_out`` is
threaded through nested control-flow steps so gate verdict bindings can be
rejected anywhere inside a fan-out template.
``input_names`` is the set of declared workflow input names (or ``None``
when the inputs block is malformed). ``inside_fan_out`` is threaded
through nested control-flow steps so gate verdict bindings can be rejected
anywhere inside a fan-out template.
"""
from . import STEP_REGISTRY
@@ -441,39 +440,11 @@ def _validate_steps(
f"Gate step {step_id!r}: 'verdict_input' is not "
"supported inside fan-out templates."
)
elif input_defs is not None and verdict_input not in input_defs:
elif input_names is not None and verdict_input not in input_names:
errors.append(
f"Gate step {step_id!r}: 'verdict_input' references "
f"undeclared input {verdict_input!r}."
)
elif input_defs is not None:
# ``on_reject: retry`` resets the bound input to "" before
# pausing, and every later resume re-resolves the persisted
# inputs through ``_coerce_input``. If the input declares an
# ``enum`` that omits "", that reset value is instantly
# illegal: the run pauses fine, but the next resume that
# supplies any input raises "value '' not in allowed
# values", and no verdict can be routed through the gate
# again. Require the enum to admit the sentinel so the
# retry cycle the field advertises is actually reachable.
verdict_def = input_defs.get(verdict_input)
enum_values = (
verdict_def.get("enum")
if isinstance(verdict_def, dict)
else None
)
if (
step_config.get("on_reject") == "retry"
and isinstance(enum_values, list)
and "" not in enum_values
):
errors.append(
f"Gate step {step_id!r}: on_reject='retry' resets "
f"verdict input {verdict_input!r} to '' when the "
f"gate is rejected, but that input's 'enum' does "
f"not allow ''. Add '' to the enum or use "
f"on_reject='abort'/'skip'."
)
# Recursively validate nested steps
for nested_key in ("then", "else", "steps"):
@@ -483,7 +454,7 @@ def _validate_steps(
nested,
seen_ids,
errors,
input_defs,
input_names,
inside_fan_out=inside_fan_out,
)
@@ -496,7 +467,7 @@ def _validate_steps(
case_steps,
seen_ids,
errors,
input_defs,
input_names,
inside_fan_out=inside_fan_out,
)
@@ -507,7 +478,7 @@ def _validate_steps(
default,
seen_ids,
errors,
input_defs,
input_names,
inside_fan_out=inside_fan_out,
)
@@ -520,7 +491,7 @@ def _validate_steps(
[fan_step],
set(),
fan_errors,
input_defs,
input_names,
inside_fan_out=True,
)
errors.extend(fan_errors)
@@ -20,31 +20,9 @@ class FanInStep(StepBase):
def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
wait_for = config.get("wait_for", [])
output_config = config.get("output")
if output_config is None:
output_config = config.get("output") or {}
if not isinstance(output_config, dict):
output_config = {}
elif not isinstance(output_config, dict):
# ``validate`` rejects a non-mapping ``output`` and its comment says
# why: "execute() silently coerces a non-mapping output to {}, so the
# author's declared aggregation keys would vanish with no error."
# The engine does not auto-validate before ``execute``, so on an
# unvalidated run that is exactly what happened -- and ``x or {}``
# masked the falsy shapes ([], false, 0, '') before the isinstance
# check even ran. Every declared key vanished while the step still
# reported COMPLETED, so downstream ``steps.<id>.output.<key>``
# resolved to None and interpolated as "": the same "silent empty
# result + COMPLETED" wiring bug the ``wait_for`` guard below
# rejects. Fail loudly with validate()'s own message instead. An
# explicit ``output:`` (YAML null) stays valid, matching validate.
return StepResult(
status=StepStatus.FAILED,
error=(
f"Fan-in step {config.get('id', '?')!r}: 'output' must be a "
f"mapping of key -> expression, got "
f"{type(output_config).__name__}."
),
output={"results": []},
)
# The engine does not auto-validate step config, so an unvalidated run
# with a non-list ``wait_for`` reaches here raw. Iterating it then
@@ -75,31 +75,6 @@ class GateStep(StepBase):
},
)
# ``validate`` rejects an ``on_reject`` outside abort/skip/retry, but the
# engine does not auto-validate before ``execute``. The reject branch
# below handles only "abort" and "retry" and then falls through to its
# ``on_reject == "skip"`` case, so on an unvalidated run any other value
# makes a REJECTED gate report COMPLETED and the run walks straight past
# the review the gate exists to enforce. Reachable by a capitalisation
# slip ("Abort"), a guessed verb ("fail", "stop"), a non-string, or the
# ``None`` that a bare ``on_reject:`` yields -- note ``config.get(k,
# default)`` does NOT substitute the default for an explicit null. Fail
# loudly instead, mirroring the ``options``/``verdict_input`` guards here.
if on_reject not in ("abort", "skip", "retry"):
return StepResult(
status=StepStatus.FAILED,
error=(
f"Gate step {config.get('id', '?')!r}: 'on_reject' must be "
f"'abort', 'skip', or 'retry', got {on_reject!r}."
),
output={
"message": message,
"options": options,
"on_reject": on_reject,
"choice": None,
},
)
if has_verdict_input and (
not isinstance(verdict_input, str) or not verdict_input
):
@@ -11,10 +11,7 @@ from __future__ import annotations
import os
from typing import Any
from specify_cli._agent_config import (
SCRIPT_TYPE_CHOICES,
resolve_default_init_integration,
)
from specify_cli._agent_config import DEFAULT_INIT_INTEGRATION, SCRIPT_TYPE_CHOICES
from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
from specify_cli.workflows.expressions import evaluate_expression
@@ -57,8 +54,7 @@ class InitStep(StepBase):
Initialize in the target directory instead of creating a new one.
``integration``
Integration key (e.g. ``copilot``). Defaults to the workflow's
default integration, then to the resolved default init integration
(``SPECKIT_INTEGRATION_DEFAULT`` env var, else ``copilot``).
default integration, then to ``DEFAULT_INIT_INTEGRATION``.
``integration_options``
Extra options for the integration (e.g. ``"--skills"`` or
``"--commands-dir .myagent/cmds"``).
@@ -85,7 +81,7 @@ class InitStep(StepBase):
# Apply the same default that specify init uses in non-interactive mode
# so that output.integration reflects the actual integration used.
if not integration:
integration = resolve_default_init_integration()
integration = DEFAULT_INIT_INTEGRATION
integration_options = self._resolve(
config.get("integration_options"), context
@@ -95,17 +91,9 @@ class InitStep(StepBase):
force = self._resolve_bool(config.get("force"), context)
# Workflows run unattended; skip the agent CLI presence check by default.
# ``config.get(key, True)`` applies that default only when the key is
# ABSENT: a bare ``ignore_agent_tools:`` in YAML parses to None, which
# ``_resolve_bool`` then turns into False -- flipping the documented
# default and re-enabling the agent-CLI presence check this step promises
# to skip, so an unattended run fails with "Agent Detection Error" for
# any integration whose CLI is not installed. Normalize an explicit null
# to the default, mirroring the while/do-while ``max_iterations`` handling.
raw_ignore_agent_tools = config.get("ignore_agent_tools")
if raw_ignore_agent_tools is None:
raw_ignore_agent_tools = True
ignore_agent_tools = self._resolve_bool(raw_ignore_agent_tools, context)
ignore_agent_tools = self._resolve_bool(
config.get("ignore_agent_tools", True), context
)
argv: list[str] = ["init"]
if here:
@@ -44,20 +44,6 @@ def test_precedence_default_when_unspecified():
assert _resolve_init_integration(None, None) == "copilot"
def test_precedence_default_honors_env_var(monkeypatch):
monkeypatch.setenv("SPECKIT_INTEGRATION_DEFAULT", "gemini")
# With no override and no bundle-declared integration, the env-var default
# applies instead of the hardcoded "copilot".
assert _resolve_init_integration(None, None) == "gemini"
assert _resolve_init_integration(None, _manifest()) == "gemini"
# Explicit override and bundle-declared integration still take precedence.
assert _resolve_init_integration("claude", None) == "claude"
assert (
_resolve_init_integration(None, _manifest(integration={"id": "claude"}))
== "claude"
)
def _build_mini(tmp_path: Path) -> Path:
bundle = tmp_path / "mini"
bundle.mkdir()
-338
View File
@@ -116,68 +116,6 @@ class TestInitIntegrationFlag:
data = json.loads((project / ".specify" / "integration.json").read_text(encoding="utf-8"))
assert data["integration"] == specify_cli.DEFAULT_INIT_INTEGRATION
def test_noninteractive_init_honors_default_integration_env_var(
self, tmp_path, monkeypatch
):
from typer.testing import CliRunner
from specify_cli import app
import specify_cli
def fail_select(*_args, **_kwargs):
raise AssertionError("non-interactive init should not open the integration picker")
monkeypatch.setattr(specify_cli, "select_with_arrows", fail_select)
monkeypatch.setenv(
specify_cli.DEFAULT_INIT_INTEGRATION_ENV_VAR, "gemini"
)
runner = CliRunner()
project = tmp_path / "noninteractive_env"
result = runner.invoke(app, [
"init", str(project), "--script", "sh", "--ignore-agent-tools",
], catch_exceptions=False)
assert result.exit_code == 0, result.output
assert "defaulting to 'gemini'" in result.output
data = json.loads((project / ".specify" / "integration.json").read_text(encoding="utf-8"))
assert data["integration"] == "gemini"
def test_interactive_init_picker_default_honors_env_var(
self, tmp_path, monkeypatch
):
# The interactive integration picker must receive the resolved
# SPECKIT_INTEGRATION_DEFAULT value as its default_key, not the
# hardcoded constant (guards the picker wiring against regression).
from typer.testing import CliRunner
from specify_cli import app
import specify_cli.commands.init as init_mod
monkeypatch.setattr(init_mod, "_stdin_is_interactive", lambda: True)
monkeypatch.setenv("SPECKIT_INTEGRATION_DEFAULT", "gemini")
captured = {}
def fake_select(options, prompt_text=None, default_key=None):
# Only capture the integration picker (not the script picker).
if "Choose your coding agent integration" in (prompt_text or ""):
captured["default_key"] = default_key
return default_key
monkeypatch.setattr(init_mod, "select_with_arrows", fake_select)
runner = CliRunner()
project = tmp_path / "interactive_env"
result = runner.invoke(app, [
"init", str(project), "--script", "sh", "--ignore-agent-tools",
], catch_exceptions=False)
assert result.exit_code == 0, result.output
assert captured.get("default_key") == "gemini"
data = json.loads((project / ".specify" / "integration.json").read_text(encoding="utf-8"))
assert data["integration"] == "gemini"
def test_init_here_nonempty_noninteractive_errors_with_force_guidance(self, tmp_path):
"""`init --here` on a non-empty directory with no confirmation input (empty
stdin) must fail fast with guidance to use --force, instead of the bare
@@ -2434,279 +2372,3 @@ def test_refresh_shared_templates_preserves_recovered_user_file(tmp_path):
# Recovered user content must survive (fail-before: replaced by bundled body).
assert user_file.read_text(encoding="utf-8") == "# USER CUSTOM CONTENT\n"
class TestExtensionFlag:
"""Tests for the --extension flag on specify init."""
def _run_init(self, tmp_path, args, project_name="ext-test"):
from unittest.mock import patch
from typer.testing import CliRunner
from specify_cli import app
project = tmp_path / project_name
project.mkdir(exist_ok=True)
old_cwd = os.getcwd()
try:
os.chdir(project)
runner = CliRunner()
# Patch get_speckit_version to return a stable (non-dev) version so that
# the extension compatibility check (SpecifierSet(">=0.2.0")) passes.
with patch(
"specify_cli.commands.init.get_speckit_version",
return_value="0.8.2",
):
result = runner.invoke(app, [
"init", "--here",
"--integration", "copilot",
"--script", "sh",
"--ignore-agent-tools",
] + args, catch_exceptions=False)
finally:
os.chdir(old_cwd)
return project, result
def test_bundled_extension_installed(self, tmp_path):
"""--extension git installs the bundled git extension."""
project, result = self._run_init(tmp_path, ["--extension", "git"], project_name="ext-bundled")
assert result.exit_code == 0, f"init failed:\n{result.output}"
ext_dir = project / ".specify" / "extensions" / "git"
assert ext_dir.exists(), "git extension directory not found"
assert (ext_dir / "extension.yml").exists(), "extension.yml not found"
# Tracker should show extension step as done
normalized = _normalize_cli_output(result.output)
assert "Install extension: git" in normalized
def test_multiple_extensions_installed(self, tmp_path):
"""--extension can be specified multiple times."""
project, result = self._run_init(
tmp_path,
["--extension", "git", "--extension", "selftest"],
project_name="ext-multi",
)
assert result.exit_code == 0, f"init failed:\n{result.output}"
ext_dir_git = project / ".specify" / "extensions" / "git"
ext_dir_selftest = project / ".specify" / "extensions" / "selftest"
assert ext_dir_git.exists(), "git extension not installed"
assert ext_dir_selftest.exists(), "selftest extension not installed"
def test_local_path_extension_installed(self, tmp_path):
"""--extension /abs/path installs from a local absolute directory path."""
from specify_cli import _locate_bundled_extension
# Use the bundled git extension directory as our "local" extension source
bundled_git = _locate_bundled_extension("git")
assert bundled_git is not None, "bundled git extension not found; cannot run test"
# Pass the absolute path directly (starts with "/")
project, result = self._run_init(
tmp_path,
["--extension", str(bundled_git)],
project_name="ext-local",
)
assert result.exit_code == 0, f"init failed:\n{result.output}"
ext_dir = project / ".specify" / "extensions" / "git"
assert ext_dir.exists(), "extension from local path not installed"
def test_unknown_extension_shows_error_in_tracker(self, tmp_path):
"""An unknown extension name records a tracker error but does not abort init."""
project, result = self._run_init(
tmp_path,
["--extension", "nonexistent-xyz-ext"],
project_name="ext-unknown",
)
assert result.exit_code == 0, "init should not abort on unknown extension"
normalized = _normalize_cli_output(result.output)
assert "failed" in normalized.lower(), "expected 'failed' for unknown extension"
def test_extension_flag_works_with_preset(self, tmp_path):
"""--extension and --preset can be combined."""
project, result = self._run_init(
tmp_path,
["--extension", "git", "--preset", "lean"],
project_name="ext-preset",
)
assert result.exit_code == 0, f"init failed:\n{result.output}"
ext_dir = project / ".specify" / "extensions" / "git"
assert ext_dir.exists(), "git extension not installed alongside preset"
@staticmethod
def _zip_bytes_from_dir(source_dir):
"""Build in-memory ZIP bytes from an extension directory (yml at root)."""
import io
import zipfile
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
for path in sorted(source_dir.rglob("*")):
if path.is_file():
zf.write(path, arcname=str(path.relative_to(source_dir)))
return buf.getvalue()
def test_url_extension_rejects_non_https(self, tmp_path):
"""A non-HTTPS URL is rejected before any download; init is not aborted."""
project, result = self._run_init(
tmp_path,
["--extension", "http://example.com/ext.zip", "--trust-extension-urls"],
project_name="ext-http",
)
assert result.exit_code == 0, "init should not abort on a rejected URL"
normalized = _normalize_cli_output(result.output)
assert "failed" in normalized.lower()
# No extension directory should have been created for the bad URL.
assert not (project / ".specify" / "extensions" / "ext").exists()
def test_url_extension_skipped_without_trust(self, tmp_path):
"""Non-interactive URL install without --trust-extension-urls is denied."""
from unittest.mock import patch
with patch(
"specify_cli.commands.init._stdin_is_interactive", return_value=False
), patch("specify_cli.authentication.http.open_url") as mock_open:
project, result = self._run_init(
tmp_path,
["--extension", "https://example.com/git.zip"],
project_name="ext-url-denied",
)
assert result.exit_code == 0, f"init failed:\n{result.output}"
# Default-deny: no download attempted, nothing installed.
mock_open.assert_not_called()
normalized = _normalize_cli_output(result.output)
assert "untrusted url" in normalized.lower()
assert not (project / ".specify" / "extensions" / "git").exists()
def test_url_extension_interactive_confirm_installs(self, tmp_path):
"""An interactive 'yes' to the trust prompt allows the URL install."""
import io
from unittest.mock import patch
from specify_cli import _locate_bundled_extension
bundled_git = _locate_bundled_extension("git")
assert bundled_git is not None, "bundled git extension not found"
zip_bytes = self._zip_bytes_from_dir(bundled_git)
class FakeResponse(io.BytesIO):
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def _cache_dir_stand_in(project_root):
d = project_root / ".specify" / "extensions" / ".cache" / "downloads"
d.mkdir(parents=True, exist_ok=True)
return d
def _open_download_zip(project_root, download_dir, zip_filename):
target = download_dir / zip_filename
o_temporary = getattr(os, "O_TEMPORARY", 0)
if o_temporary:
return os.open(
target, os.O_RDWR | os.O_CREAT | os.O_EXCL | o_temporary, 0o600
)
fd = os.open(target, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600)
try:
os.unlink(target)
except OSError:
os.close(fd)
raise
return fd
with patch(
"specify_cli.commands.init._stdin_is_interactive", return_value=True
), patch("typer.confirm", return_value=True), patch(
"specify_cli.authentication.http.open_url",
return_value=FakeResponse(zip_bytes),
), patch(
"specify_cli.extensions._commands._validate_safe_cache_dir",
side_effect=_cache_dir_stand_in,
), patch(
"specify_cli.extensions._commands._safe_open_download_zip",
side_effect=_open_download_zip,
):
project, result = self._run_init(
tmp_path,
["--extension", "https://example.com/git.zip"],
project_name="ext-url-confirm",
)
assert result.exit_code == 0, f"init failed:\n{result.output}"
assert (project / ".specify" / "extensions" / "git").exists()
def test_url_extension_installs_zip(self, tmp_path):
"""A successful HTTPS ZIP download installs via the shared hardened path."""
import io
from unittest.mock import patch
from specify_cli import _locate_bundled_extension
bundled_git = _locate_bundled_extension("git")
assert bundled_git is not None, "bundled git extension not found"
zip_bytes = self._zip_bytes_from_dir(bundled_git)
class FakeResponse(io.BytesIO):
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
def _cache_dir_stand_in(project_root):
d = project_root / ".specify" / "extensions" / ".cache" / "downloads"
d.mkdir(parents=True, exist_ok=True)
return d
def _open_download_zip(project_root, download_dir, zip_filename):
target = download_dir / zip_filename
o_temporary = getattr(os, "O_TEMPORARY", 0)
if o_temporary:
return os.open(
target, os.O_RDWR | os.O_CREAT | os.O_EXCL | o_temporary, 0o600
)
fd = os.open(target, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600)
try:
os.unlink(target)
except OSError:
os.close(fd)
raise
return fd
with patch(
"specify_cli.authentication.http.open_url",
return_value=FakeResponse(zip_bytes),
), patch(
"specify_cli.extensions._commands._validate_safe_cache_dir",
side_effect=_cache_dir_stand_in,
), patch(
"specify_cli.extensions._commands._safe_open_download_zip",
side_effect=_open_download_zip,
):
project, result = self._run_init(
tmp_path,
["--extension", "https://example.com/git.zip", "--trust-extension-urls"],
project_name="ext-url",
)
assert result.exit_code == 0, f"init failed:\n{result.output}"
ext_dir = project / ".specify" / "extensions" / "git"
assert ext_dir.exists(), "extension from URL not installed"
assert (ext_dir / "extension.yml").exists()
# Transient download archive must not linger in the cache.
cache_dir = project / ".specify" / "extensions" / ".cache" / "downloads"
leftover = list(cache_dir.glob("*.zip")) if cache_dir.exists() else []
assert not leftover, f"download cache not cleaned: {leftover}"
-7
View File
@@ -164,13 +164,6 @@ class TestCollectExtensionEvents:
(ext_dir / "extension.yml").write_text("invalid: - - -", encoding="utf-8")
assert collect_extension_events(tmp_path) == {}
def test_non_utf8_manifest_skipped(self, tmp_path):
ext_dir = tmp_path / ".specify" / "extensions" / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "extension.yml").write_bytes(b"\xff\xfe")
assert collect_extension_events(tmp_path) == {}
def test_event_command_ref_canonicalized_via_manifest(self, tmp_path):
"""R1: events are read from a validated ExtensionManifest, so an
obsolete command ref (e.g. my-ext.boot) is canonicalized
@@ -199,24 +199,6 @@ class TestKimiTeardownLegacyCleanup:
assert user_skill.exists()
def test_teardown_preserves_non_utf8_user_skill(self, tmp_path):
i = get_integration("kimi")
user_skill = (
tmp_path
/ ".kimi"
/ "skills"
/ "speckit-user-owned"
/ "SKILL.md"
)
user_skill.parent.mkdir(parents=True)
user_skill.write_bytes(b"\xff\xfe")
m = IntegrationManifest("kimi", tmp_path)
i.teardown(tmp_path, m)
assert user_skill.read_bytes() == b"\xff\xfe"
class TestKimiCommandInvocation:
"""Kimi dispatch must use the native ``/skill:`` slash command."""
-48
View File
@@ -50,51 +50,3 @@ def test_init_command_registered():
cmd.callback.__name__ for cmd in app.registered_commands if cmd.callback
]
assert "init" in callback_names
def test_resolve_default_init_integration_unset(monkeypatch):
from specify_cli._agent_config import (
DEFAULT_INIT_INTEGRATION,
DEFAULT_INIT_INTEGRATION_ENV_VAR,
resolve_default_init_integration,
)
monkeypatch.delenv(DEFAULT_INIT_INTEGRATION_ENV_VAR, raising=False)
assert resolve_default_init_integration() == DEFAULT_INIT_INTEGRATION
def test_resolve_default_init_integration_valid_override(monkeypatch):
from specify_cli._agent_config import (
DEFAULT_INIT_INTEGRATION_ENV_VAR,
resolve_default_init_integration,
)
monkeypatch.setenv(DEFAULT_INIT_INTEGRATION_ENV_VAR, "gemini")
assert resolve_default_init_integration() == "gemini"
def test_resolve_default_init_integration_whitespace_trimmed(monkeypatch):
from specify_cli._agent_config import (
DEFAULT_INIT_INTEGRATION_ENV_VAR,
resolve_default_init_integration,
)
monkeypatch.setenv(DEFAULT_INIT_INTEGRATION_ENV_VAR, " gemini ")
assert resolve_default_init_integration() == "gemini"
def test_resolve_default_init_integration_invalid_warns_and_falls_back(
monkeypatch, capsys
):
from specify_cli._agent_config import (
DEFAULT_INIT_INTEGRATION,
DEFAULT_INIT_INTEGRATION_ENV_VAR,
resolve_default_init_integration,
)
monkeypatch.setenv(DEFAULT_INIT_INTEGRATION_ENV_VAR, "not-a-real-agent")
assert resolve_default_init_integration() == DEFAULT_INIT_INTEGRATION
captured = capsys.readouterr()
assert "not-a-real-agent" in captured.err
assert DEFAULT_INIT_INTEGRATION_ENV_VAR in captured.err
def test_resolve_default_init_integration_re_exported_from_init():
from specify_cli import resolve_default_init_integration
assert callable(resolve_default_init_integration)
+4 -407
View File
@@ -671,102 +671,6 @@ class TestExtensionManifest:
with pytest.raises(ValidationError, match=f"Invalid {section}"):
ExtensionManifest(manifest_path)
@pytest.mark.parametrize("field", ["id", "name", "version", "description"])
@pytest.mark.parametrize("bad", [1.0, 5, None, ["a"], {"a": 1}, True])
def test_extension_metadata_field_not_string_rejected(
self, temp_dir, valid_manifest_data, field, bad
):
"""A non-string extension.<field> must raise ValidationError, not a raw
TypeError.
The loop over these four fields only checked key PRESENCE, then fed the
values to ``re.match`` (id) and ``packaging.Version`` (version), both of
which raise a bare TypeError on a non-string. YAML makes that an easy
authoring slip: unquoted ``version: 1.0`` parses as a float and ``id: 2``
as an int. TypeError is not a ValidationError, so it escaped
list_installed()'s "Corrupted extension" fallback and made
`specify extension list` exit 1 with a raw traceback, hiding every
healthy extension too. The sibling IntegrationDescriptor already
type-checks the same four fields.
"""
import yaml
valid_manifest_data["extension"][field] = bad
manifest_path = temp_dir / "extension.yml"
with open(manifest_path, 'w') as f:
yaml.dump(valid_manifest_data, f)
with pytest.raises(ValidationError, match=f"Invalid extension.{field}"):
ExtensionManifest(manifest_path)
@pytest.mark.parametrize("bad", [1.0, 5, None, ["a"], {"a": 1}, True])
def test_command_name_not_string_rejected(
self, temp_dir, valid_manifest_data, bad
):
"""A non-string command name must raise ValidationError, not a raw
TypeError from the name-pattern match.
The sibling ``file`` field was already covered, since
relative_extension_path_violation() rejects a non-string value; ``name``
went straight into EXTENSION_COMMAND_NAME_PATTERN.match().
"""
import yaml
valid_manifest_data["provides"]["commands"][0]["name"] = bad
manifest_path = temp_dir / "extension.yml"
with open(manifest_path, 'w') as f:
yaml.dump(valid_manifest_data, f)
with pytest.raises(ValidationError, match="Invalid command name"):
ExtensionManifest(manifest_path)
def test_one_bad_manifest_does_not_hide_healthy_extensions(self, temp_dir):
"""End-to-end guard for the symptom: an unquoted ``version: 1.0`` in one
installed extension must degrade to "Corrupted extension" and still let
list_installed() report the healthy ones, instead of raising TypeError
out of the whole call.
"""
ext_root = temp_dir / ".specify" / "extensions"
for ext_id, version in (("good-ext", '"1.0.0"'), ("bad-ext", "1.0")):
ext_path = ext_root / ext_id
ext_path.mkdir(parents=True, exist_ok=True)
(ext_path / "extension.yml").write_text(
f"""schema_version: "1.0"
extension:
id: {ext_id}
name: {ext_id}
version: {version}
description: desc
requires:
speckit_version: ">=0.1.0"
provides:
commands:
- name: speckit.{ext_id}.hello
file: commands/hello.md
""",
encoding="utf-8",
)
(ext_root / ".registry").write_text(
json.dumps(
{
"schema_version": "1.0",
"extensions": {
"good-ext": {"version": "1.0.0", "enabled": True},
"bad-ext": {"version": "1.0", "enabled": True},
},
}
),
encoding="utf-8",
)
listed = {row["id"]: row for row in ExtensionManager(temp_dir).list_installed()}
assert set(listed) == {"good-ext", "bad-ext"}
assert "Corrupted" not in listed["good-ext"]["description"]
assert "Corrupted" in listed["bad-ext"]["description"]
def test_empty_provides_mapping_is_still_accepted_with_hooks(
self, temp_dir, valid_manifest_data
):
@@ -5099,9 +5003,9 @@ class TestExtensionCatalog:
catalog = self._make_catalog(temp_dir)
mock_response = MagicMock()
mock_response.read.side_effect = io.BytesIO(json.dumps(
mock_response.read.return_value = json.dumps(
{"schema_version": "1.0", "extensions": {}}
).encode()).read
).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "http://evil.test/catalog.json"
@@ -5147,9 +5051,9 @@ class TestExtensionCatalog:
catalog = self._make_catalog(temp_dir)
mock_response = MagicMock()
mock_response.read.side_effect = io.BytesIO(json.dumps(
mock_response.read.return_value = json.dumps(
{"schema_version": "1.0", "extensions": {}}
).encode()).read
).encode()
mock_response.__enter__ = lambda s: s
mock_response.__exit__ = MagicMock(return_value=False)
mock_response.geturl.return_value = "http://evil.test/catalog.json"
@@ -10550,310 +10454,3 @@ def test_forge_extension_info_hyphenates_command_names(
# not the manifest's dotted name.
assert "speckit-test-ext-hello" in output, output
assert "speckit.test-ext.hello" not in output, output
# ===== Extension Config Scaffolding Tests =====
class TestExtensionConfigScaffolding:
"""Test automatic config scaffolding during add/enable lifecycle."""
def _make_extension(self, ext_dir, config_entries=None):
"""Create a minimal extension with optional config templates."""
ext_dir.mkdir(parents=True, exist_ok=True)
manifest = {
"schema_version": "1.0",
"extension": {
"id": "test-ext",
"name": "Test Extension",
"version": "1.0.0",
"description": "Test extension",
"author": "Test",
"repository": "https://github.com/test/test",
"license": "MIT",
"homepage": "https://github.com/test/test",
},
"requires": {"speckit_version": ">=0.1.0"},
"provides": {
"commands": [{
"name": "speckit.test-ext.example",
"file": "commands/example.md",
"description": "Example command",
}],
},
"tags": ["test"],
}
if config_entries:
manifest["provides"]["config"] = config_entries
import yaml
(ext_dir / "extension.yml").write_text(yaml.dump(manifest, default_flow_style=False))
# Create command file so validation passes
(ext_dir / "commands").mkdir(exist_ok=True)
(ext_dir / "commands" / "example.md").write_text("# Example")
return manifest
def test_scaffold_config_deploys_template(self, tmp_path):
"""Config template lands where ConfigManager reads it, not in .specify/ root."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[{
"name": "test-config.yml",
"template": "config-template.yml",
"description": "Test config",
"required": True,
}])
(ext_dir / "config-template.yml").write_text("setting: default")
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == ["test-config.yml"]
assert skipped == []
assert failed == []
# ConfigManager._get_project_config() reads
# .specify/extensions/<id>/<name>, so that is where scaffolding must
# put it. Deploying to the .specify/ root left the file somewhere the
# extension never looks.
assert (ext_dir / "test-config.yml").exists()
assert (ext_dir / "test-config.yml").read_text() == "setting: default"
assert not (specify_dir / "test-config.yml").exists()
def test_scaffold_config_preserves_existing(self, tmp_path):
"""Existing config files should never be overwritten."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[{
"name": "test-config.yml",
"template": "config-template.yml",
"description": "Test config",
"required": True,
}])
(ext_dir / "config-template.yml").write_text("setting: default")
(ext_dir / "test-config.yml").write_text("setting: custom")
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == []
assert skipped == ["test-config.yml"]
assert failed == []
assert (ext_dir / "test-config.yml").read_text() == "setting: custom"
def test_scaffold_config_no_config_section(self, tmp_path):
"""Extensions without config section should return empty list."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir)
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == []
assert skipped == []
assert failed == []
def test_scaffold_config_missing_template_file(self, tmp_path):
"""Missing template files should be reported as failed."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[{
"name": "test-config.yml",
"template": "nonexistent.yml",
"description": "Test config",
}])
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == []
assert skipped == []
assert failed == ["test-config.yml"]
def test_scaffold_config_rejects_path_traversal(self, tmp_path):
"""Config names with path traversal should be rejected."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[
{"name": "../etc/passwd", "template": "config.yml"},
{"name": "safe.yml", "template": "../../secrets.yml"},
{"name": "/absolute/path.yml", "template": "config.yml"},
])
(ext_dir / "config.yml").write_text("safe: true")
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == []
assert skipped == []
assert failed == ["../etc/passwd", "safe.yml", "/absolute/path.yml"]
def test_scaffold_config_rejects_directory_template(self, tmp_path):
"""Directory templates should be rejected (must be regular files)."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[{
"name": "test-config.yml",
"template": "config-dir",
}])
(ext_dir / "config-dir").mkdir()
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == []
assert skipped == []
assert failed == ["test-config.yml"]
def test_scaffold_config_rejects_symlink_template(self, tmp_path):
"""Symlink templates should not be copied."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[{
"name": "test-config.yml",
"template": "config-link.yml",
}])
real_template = ext_dir / "config-template.yml"
real_template.write_text("setting: default")
(ext_dir / "config-link.yml").symlink_to(real_template)
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == []
assert skipped == []
assert failed == ["test-config.yml"]
assert not (specify_dir / "test-config.yml").exists()
def test_scaffold_config_malformed_manifest(self, tmp_path):
"""Malformed config sections should not crash."""
from specify_cli.extensions import ExtensionManager, ExtensionManifest
import yaml
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
manifest_data = self._make_extension(ext_dir)
manifest_data["provides"]["config"] = "not-a-list"
(ext_dir / "extension.yml").write_text(yaml.dump(manifest_data))
manifest = ExtensionManifest(ext_dir / "extension.yml")
assert manifest.config == []
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == []
assert skipped == []
assert failed == ["provides.config"]
def test_scaffold_config_missing_manifest_returns_consistent_result(self, tmp_path):
"""A missing extension manifest should return the documented tuple."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
(project / ".specify").mkdir(parents=True)
manager = ExtensionManager(project)
assert manager.scaffold_config("missing") == ([], [], [])
def test_scaffold_config_rejects_symlinked_config_root(self, tmp_path):
"""A symlinked .specify must not become the containment root.
Resolving .specify first and trusting the result lets a symlink point
anywhere: every target then satisfies relative_to and copy2 writes
outside the project.
"""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
project.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
(project / ".specify").symlink_to(outside, target_is_directory=True)
ext_dir = outside / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[{
"name": "test-config.yml",
"template": "config-template.yml",
"description": "Test config",
"required": True,
}])
(ext_dir / "config-template.yml").write_text("setting: default")
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == []
assert skipped == []
assert failed == ["provides.config"]
assert not (outside / "extensions" / "test-ext" / "test-config.yml").exists()
def test_scaffold_config_rejects_targets_removal_would_not_preserve(self, tmp_path):
"""Only top-level *-config.yml targets are scaffolded.
remove(keep_config=True) rmtree's every subdirectory and keeps only
top-level -config.yml / -config.local.yml files, and the backup path
globs the same pattern. Scaffolding anything else would hand the user a
file that `extension add --force` silently replaces with the template
default.
"""
from specify_cli.extensions import ExtensionManager
for target in ("nested/test-config.yml", "settings.yml", "test-config.yaml"):
project = tmp_path / f"project-{target.replace('/', '_')}"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[{
"name": target,
"template": "config-template.yml",
"description": "Test config",
"required": True,
}])
(ext_dir / "config-template.yml").write_text("setting: default")
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == [], target
assert skipped == [], target
assert failed == [target], target
def test_scaffold_config_accepts_local_override_name(self, tmp_path):
"""*-config.local.yml is preserved by removal, so it may be scaffolded."""
from specify_cli.extensions import ExtensionManager
project = tmp_path / "project"
specify_dir = project / ".specify"
specify_dir.mkdir(parents=True)
ext_dir = specify_dir / "extensions" / "test-ext"
self._make_extension(ext_dir, config_entries=[{
"name": "test-config.local.yml",
"template": "config-template.yml",
"description": "Test config",
"required": True,
}])
(ext_dir / "config-template.yml").write_text("setting: default")
manager = ExtensionManager(project)
deployed, skipped, failed = manager.scaffold_config("test-ext")
assert deployed == ["test-config.local.yml"]
assert failed == []
-26
View File
@@ -274,29 +274,3 @@ class TestClineRealPostProcess:
# _rewrite_handoff_references rewrote the dotted agent handoff
assert "agent: speckit-foo" in content
assert "agent: speckit.foo" not in content
def test_register_commands_propagates_programming_errors(tmp_path):
"""Regression: narrowed exception must not swallow TypeError/AttributeError.
The invoke separator resolution narrowed from bare 'except Exception' to
'except (ImportError, ValueError, KeyError)'. Programming errors like
TypeError must propagate instead of being silently swallowed.
"""
registrar = CommandRegistrar()
commands = [{"name": "test.cmd", "file": "commands/test.md"}]
ext_dir = tmp_path / "ext"
ext_dir.mkdir()
def _broken_get_integration(name):
raise TypeError("intentional programming error")
import specify_cli.integrations as integ_mod
original = integ_mod.get_integration
integ_mod.get_integration = _broken_get_integration
try:
with pytest.raises(TypeError, match="intentional programming error"):
registrar.register_commands("bob", commands, "ext", ext_dir, tmp_path)
finally:
integ_mod.get_integration = original
-230
View File
@@ -198,121 +198,6 @@ class TestPresetManifest:
with pytest.raises(PresetValidationError, match="YAML mapping"):
PresetManifest(manifest_path)
@pytest.mark.parametrize("section", ["preset", "requires", "provides"])
@pytest.mark.parametrize("bad_value", [None, [], "text"])
def test_required_section_not_mapping_raises_validation_error(
self, temp_dir, valid_pack_data, section, bad_value
):
"""Required manifest sections reject null, list, and scalar values."""
valid_pack_data[section] = bad_value
manifest_path = temp_dir / "preset.yml"
manifest_path.write_text(
yaml.safe_dump(valid_pack_data),
encoding="utf-8",
)
with pytest.raises(
PresetValidationError,
match=rf"Invalid {section}: expected a mapping",
):
PresetManifest(manifest_path)
@pytest.mark.parametrize("field", ["id", "name", "version", "description"])
@pytest.mark.parametrize("bad", [1.0, 5, None, ["a"], {"a": 1}, True])
def test_preset_metadata_field_not_string_raises_validation_error(
self, temp_dir, valid_pack_data, field, bad
):
"""A non-string preset.<field> raises PresetValidationError, not a raw
TypeError.
The loop over these four fields only checked key PRESENCE, then fed the
values to ``re.match`` (id) and ``packaging.Version`` (version), both of
which raise a bare TypeError on a non-string. YAML makes that an easy
authoring slip: unquoted ``version: 1.0`` parses as a float and ``id: 2``
as an int. TypeError is not a PresetValidationError, so it escaped
list_installed()'s "Corrupted preset" fallback and made
`specify preset list` exit 1 with a raw traceback, hiding every healthy
preset too. The sibling IntegrationDescriptor already type-checks the
same four fields.
"""
valid_pack_data["preset"][field] = bad
manifest_path = temp_dir / "preset.yml"
manifest_path.write_text(yaml.safe_dump(valid_pack_data), encoding="utf-8")
with pytest.raises(
PresetValidationError,
match=rf"Invalid preset\.{field}: expected a string",
):
PresetManifest(manifest_path)
@pytest.mark.parametrize("field", ["name", "file"])
@pytest.mark.parametrize("bad", [1.0, 5, None, ["a"], {"a": 1}, True])
def test_template_entry_field_not_string_raises_validation_error(
self, temp_dir, valid_pack_data, field, bad
):
"""A non-string template ``name``/``file`` raises PresetValidationError.
``name`` reaches ``re.match`` and ``file`` reaches ``os.path.normpath``;
both raise a bare TypeError on a non-string. The sibling extension
manifest already rejects a non-string command ``file`` via
relative_extension_path_violation().
"""
valid_pack_data["provides"]["templates"][0][field] = bad
manifest_path = temp_dir / "preset.yml"
manifest_path.write_text(yaml.safe_dump(valid_pack_data), encoding="utf-8")
with pytest.raises(
PresetValidationError,
match=rf"Invalid template {field}: expected a string",
):
PresetManifest(manifest_path)
def test_one_bad_manifest_does_not_hide_healthy_presets(self, temp_dir):
"""End-to-end guard for the symptom: an unquoted ``version: 1.0`` in one
installed preset must degrade to "Corrupted preset" and still let
list_installed() report the healthy ones, instead of raising TypeError
out of the whole call.
"""
preset_root = temp_dir / ".specify" / "presets"
for pack_id, version in (("good-pack", '"1.0.0"'), ("bad-pack", "1.0")):
pack_path = preset_root / pack_id
pack_path.mkdir(parents=True, exist_ok=True)
(pack_path / "preset.yml").write_text(
f"""schema_version: "1.0"
preset:
id: {pack_id}
name: {pack_id}
version: {version}
description: desc
requires:
speckit_version: ">=0.1.0"
provides:
templates:
- type: template
name: spec
file: templates/spec.md
""",
encoding="utf-8",
)
(preset_root / ".registry").write_text(
json.dumps(
{
"schema_version": "1.0",
"presets": {
"good-pack": {"version": "1.0.0", "enabled": True},
"bad-pack": {"version": "1.0", "enabled": True},
},
}
),
encoding="utf-8",
)
listed = {row["id"]: row for row in PresetManager(temp_dir).list_installed()}
assert set(listed) == {"good-pack", "bad-pack"}
assert "Corrupted" not in listed["good-pack"]["description"]
assert "Corrupted" in listed["bad-pack"]["description"]
@pytest.mark.parametrize(
"bad",
[
@@ -5021,103 +4906,6 @@ class TestPresetSkills:
assert "templates/commands/specify.md" in content, "Should reference core template"
assert "disable-model-invocation: false" in content
def test_skill_restored_on_preset_remove_without_project_core_templates(self, project_dir):
"""Removing a preset must restore core skills even when the project
has no ``.specify/templates/commands`` directory of its own — which
is the normal case, since ``specify init`` never populates it. The
real core commands live in the bundled core_pack/repo-root templates
tree, and restoration must fall back there instead of deleting the
skill outright (#3928).
"""
self._write_init_options(project_dir, ai="claude")
skills_dir = project_dir / ".claude" / "skills"
self._create_skill(skills_dir, "speckit-specify")
# The project_dir fixture's commands dir is empty, matching a real
# project — specify init never populates project-local overrides
# for unmodified core commands.
core_cmds = project_dir / ".specify" / "templates" / "commands"
assert core_cmds.exists() and not any(core_cmds.iterdir())
manager = PresetManager(project_dir)
install_self_test_preset(manager)
skill_file = skills_dir / "speckit-specify" / "SKILL.md"
assert "preset:self-test" in skill_file.read_text(encoding="utf-8")
manager.remove("self-test")
assert skill_file.exists(), "Core skill must be restored, not deleted"
content = skill_file.read_text(encoding="utf-8")
assert "preset:self-test" not in content
assert "templates/commands/specify.md" in content
assert "Create or update the feature specification" in content
def test_extension_wins_over_bundled_core_on_preset_remove(
self, project_dir, monkeypatch
):
"""When an installed extension owns the same skill name as a core
command, removing a preset that overrode that skill must restore it
from the extension, not silently from the bundled core template.
Extensions are resolved ahead of bundled core elsewhere, and the
bundled-core fallback added for #3928 must not replace that
higher-priority layer.
The extension-command namespace rules (``speckit.<ext>.<command>``)
make a genuine end-to-end name collision with a core command
cumbersome to construct through real manifests, so this stubs
``_build_extension_skill_restore_index`` to exercise the priority
ordering in ``_unregister_skills_in_dir`` directly -- the code path
under test doesn't care how the index entry was produced, only that
it wins over the bundled-core fallback when present.
"""
self._write_init_options(project_dir, ai="claude")
skills_dir = project_dir / ".claude" / "skills"
self._create_skill(skills_dir, "speckit-specify")
# No project-local core template override — the normal case, and
# the one that makes the bundled-core fallback kick in at all.
core_cmds = project_dir / ".specify" / "templates" / "commands"
assert core_cmds.exists() and not any(core_cmds.iterdir())
extension_dir = project_dir / ".specify" / "extensions" / "fakeext"
(extension_dir / "commands").mkdir(parents=True, exist_ok=True)
ext_specify_file = extension_dir / "commands" / "specify.md"
ext_specify_file.write_text(
"---\ndescription: Extension specify command\n---\n\n"
"extension:fakeext specify body\n"
)
manager = PresetManager(project_dir)
install_self_test_preset(manager)
skill_file = skills_dir / "speckit-specify" / "SKILL.md"
assert "preset:self-test" in skill_file.read_text(encoding="utf-8")
fake_restore_index = {
"speckit-specify": {
"command_name": "speckit.fakeext.specify",
"source_file": ext_specify_file,
"source": "extension:fakeext",
"extension_id": "fakeext",
"extension_dir": extension_dir,
}
}
monkeypatch.setattr(
manager,
"_build_extension_skill_restore_index",
lambda: fake_restore_index,
)
manager.remove("self-test")
assert skill_file.exists()
content = skill_file.read_text(encoding="utf-8")
assert "preset:self-test" not in content
assert "source: extension:fakeext" in content
assert "extension:fakeext specify body" in content
assert "templates/commands/specify.md" not in content
def test_skill_restored_on_remove_resolves_script_placeholders(self, project_dir):
"""Core restore should resolve {SCRIPT}/{ARGS} placeholders like other skill paths."""
self._write_init_options(project_dir, ai="claude", ai_skills=True, script="sh")
@@ -11819,24 +11607,6 @@ class TestResolveContent:
class TestCollectAllLayers:
"""Test PresetResolver.collect_all_layers() method."""
def test_non_utf8_legacy_command_keeps_replace_strategy(self, project_dir):
presets_dir = project_dir / ".specify" / "presets"
command_path = (
presets_dir / "legacy-pack" / "commands" / "speckit.legacy.md"
)
command_path.parent.mkdir(parents=True)
command_path.write_bytes(b"\xff\xfe")
PresetRegistry(presets_dir).add(
"legacy-pack", {"version": "1.0.0", "priority": 10}
)
layers = PresetResolver(project_dir).collect_all_layers(
"speckit.legacy", "command"
)
assert layers[0]["path"] == command_path
assert layers[0]["strategy"] == "replace"
def test_single_core_layer(self, project_dir):
"""Test collecting layers with only core template."""
resolver = PresetResolver(project_dir)
-335
View File
@@ -2176,59 +2176,6 @@ class TestInitStep:
assert "--ignore-agent-tools" in argv
assert (tmp_path / ".specify").is_dir()
def test_explicit_null_ignore_agent_tools_keeps_documented_default(
self, tmp_path
):
"""A bare ``ignore_agent_tools:`` must keep the documented default.
The class docstring says "Because workflows run unattended, the step
defaults to ``--ignore-agent-tools``" and the field docs say "defaults to
``true``". But ``config.get(key, True)`` applies the default only when the
key is ABSENT — a bare ``ignore_agent_tools:`` in YAML parses to None,
which ``_resolve_bool`` turned into False, dropping the flag and
re-enabling the agent-CLI presence check for an unattended run.
"""
from specify_cli.workflows.steps.init import InitStep
from specify_cli.workflows.base import StepContext, StepStatus
step = InitStep()
ctx = StepContext(
project_root=str(tmp_path), default_integration="copilot"
)
result = step.execute(
{
"id": "bootstrap",
"here": True,
"script": "sh",
"ignore_agent_tools": None,
},
ctx,
)
assert result.status == StepStatus.COMPLETED
assert "--ignore-agent-tools" in result.output["argv"]
def test_explicit_false_ignore_agent_tools_is_honoured(self, tmp_path):
"""An explicit ``false`` must still opt in to the agent-CLI check."""
from specify_cli.workflows.steps.init import InitStep
from specify_cli.workflows.base import StepContext
step = InitStep()
ctx = StepContext(
project_root=str(tmp_path), default_integration="copilot"
)
result = step.execute(
{
"id": "bootstrap",
"here": True,
"script": "sh",
"ignore_agent_tools": False,
},
ctx,
)
assert "--ignore-agent-tools" not in result.output["argv"]
def test_default_integration_falls_back_to_workflow_default(self, tmp_path):
from specify_cli.workflows.steps.init import InitStep
from specify_cli.workflows.base import StepContext, StepStatus
@@ -2243,24 +2190,6 @@ class TestInitStep:
assert result.status == StepStatus.COMPLETED
assert result.output["integration"] == "copilot"
def test_default_integration_honors_env_var(self, tmp_path, monkeypatch):
# With no step-level and no workflow-level default, the resolved
# SPECKIT_INTEGRATION_DEFAULT value must drive both output.integration
# and the argv passed to init (guards against reverting to the constant).
from specify_cli.workflows.steps.init import InitStep
from specify_cli.workflows.base import StepContext, StepStatus
monkeypatch.setenv("SPECKIT_INTEGRATION_DEFAULT", "gemini")
step = InitStep()
ctx = StepContext(project_root=str(tmp_path))
result = step.execute(
{"id": "bootstrap", "here": True, "script": "sh"}, ctx
)
assert result.status == StepStatus.COMPLETED
assert result.output["integration"] == "gemini"
argv = result.output["argv"]
assert "--integration" in argv and "gemini" in argv
def test_project_name_creates_subdirectory(self, tmp_path):
from specify_cli.workflows.steps.init import InitStep
from specify_cli.workflows.base import StepContext, StepStatus
@@ -2623,36 +2552,6 @@ steps:
})
assert any("on_reject" in e for e in errors)
@pytest.mark.parametrize(
"bad_on_reject", ["Abort", "fail", "stop", "SKIP", None, 5, ["abort"]]
)
def test_execute_invalid_on_reject_fails_loudly(self, bad_on_reject):
"""An unrecognised ``on_reject`` must not silently complete a rejection.
``validate`` rejects anything outside abort/skip/retry, but the engine
does not auto-validate before ``execute``. The reject branch handles only
"abort" and "retry", then falls through to its ``"skip"`` case — so a
REJECTED gate reported COMPLETED and the run continued past the review
the gate exists to enforce. Reachable by a capitalisation slip, a guessed
verb, a non-string, or a bare ``on_reject:`` (which yields None, since
``config.get(k, default)`` does not replace an explicit null).
"""
from specify_cli.workflows.steps.gate import GateStep
from specify_cli.workflows.base import StepContext, StepStatus
result = GateStep().execute(
{
"id": "review",
"message": "Review the spec.",
"options": ["approve", "reject"],
"on_reject": bad_on_reject,
"verdict_input": "spec_verdict",
},
StepContext(inputs={"spec_verdict": "reject"}),
)
assert result.status == StepStatus.FAILED
assert "'on_reject' must be" in (result.error or "")
def test_validate_non_string_options_does_not_raise(self):
"""Non-string options with on_reject=abort/retry must be REPORTED as an
error, not crash: the reject-choice check calls o.lower() on each option,
@@ -3730,44 +3629,6 @@ class TestFanInStep:
assert "'wait_for' must be a list" in (result.error or "")
assert result.output["results"] == []
@pytest.mark.parametrize(
"bad_output", [[], False, 0, "", ["a"], "oops", 5]
)
def test_execute_non_mapping_output_fails_loudly(self, bad_output):
"""A non-mapping ``output`` must fail the step, not drop every key.
``validate`` rejects it and says why: "execute() silently coerces a
non-mapping output to {}, so the author's declared aggregation keys would
vanish with no error." The engine does not auto-validate before
``execute``, so that is exactly what happened — and ``x or {}`` masked
the falsy shapes (``[]``, ``false``, ``0``, ``''``) before the isinstance
check even ran. The step still returned COMPLETED, so downstream
``steps.<id>.output.<key>`` resolved to None and interpolated as "".
"""
from specify_cli.workflows.steps.fan_in import FanInStep
from specify_cli.workflows.base import StepContext, StepStatus
step = FanInStep()
ctx = StepContext(steps={"a": {"output": {"x": 1}}})
result = step.execute(
{"id": "collect", "wait_for": ["a"], "output": bad_output}, ctx
)
assert result.status == StepStatus.FAILED
assert "'output' must be a mapping" in (result.error or "")
assert result.output["results"] == []
def test_execute_explicit_null_output_stays_valid(self):
"""An explicit ``output:`` (YAML null) is valid, matching ``validate``."""
from specify_cli.workflows.steps.fan_in import FanInStep
from specify_cli.workflows.base import StepContext, StepStatus
step = FanInStep()
ctx = StepContext(steps={"a": {"output": {"x": 1}}})
result = step.execute(
{"id": "collect", "wait_for": ["a"], "output": None}, ctx
)
assert result.status == StepStatus.COMPLETED
@pytest.mark.parametrize("bad_entry", [["a", "b"], {"a": 1}, 123, None])
def test_execute_non_string_wait_for_entry_fails_loudly(self, bad_entry):
"""A ``wait_for`` list with a non-string entry must fail the step, not
@@ -4812,149 +4673,6 @@ steps:
# No undeclared-input error (123 is not a string, so cross-check skips)
assert not any("undeclared input" in e for e in errors)
def test_retry_verdict_enum_must_allow_reset_sentinel(self):
# on_reject: retry resets the bound input to "" before pausing, and
# every resume re-resolves persisted inputs through _coerce_input. An
# enum that omits "" makes that reset value instantly illegal, so the
# next resume supplying any input dies with "value '' not in allowed
# values" and no verdict can reach the gate again.
errors = self._errors("""
workflow:
id: wf
name: wf
version: "1.0.0"
inputs:
spec_verdict:
type: string
enum: [approve, reject]
steps:
- id: review
type: gate
message: "Review?"
options: [approve, reject]
on_reject: retry
verdict_input: spec_verdict
""")
assert any(
"on_reject='retry' resets verdict input 'spec_verdict'" in e
for e in errors
), errors
def test_retry_verdict_enum_including_sentinel_passes(self):
errors = self._errors("""
workflow:
id: wf
name: wf
version: "1.0.0"
inputs:
spec_verdict:
type: string
enum: ["", approve, reject]
default: ""
steps:
- id: review
type: gate
message: "Review?"
options: [approve, reject]
on_reject: retry
verdict_input: spec_verdict
""")
assert not any("on_reject='retry'" in e for e in errors), errors
def test_verdict_enum_without_sentinel_passes_when_not_retry(self):
# abort/skip never reset the input, so the enum need not admit "".
for on_reject in ("abort", "skip"):
errors = self._errors(f"""
workflow:
id: wf
name: wf
version: "1.0.0"
inputs:
spec_verdict:
type: string
enum: [approve, reject]
steps:
- id: review
type: gate
message: "Review?"
options: [approve, reject]
on_reject: {on_reject}
verdict_input: spec_verdict
""")
assert not any("on_reject='retry'" in e for e in errors), (
on_reject,
errors,
)
def test_retry_verdict_without_enum_passes(self):
# No enum means _coerce_input accepts "" — the documented shape.
errors = self._errors("""
workflow:
id: wf
name: wf
version: "1.0.0"
inputs:
spec_verdict:
type: string
default: ""
steps:
- id: review
type: gate
message: "Review?"
options: [approve, reject]
on_reject: retry
verdict_input: spec_verdict
""")
assert not any("on_reject='retry'" in e for e in errors), errors
def test_retry_verdict_enum_wedge_is_reachable_end_to_end(self, tmp_path):
"""The validation error above guards a real, unrecoverable run state.
Without the guard this workflow installs and runs fine, then wedges:
the retry reset writes "" into the persisted inputs, and the next
resume that supplies *any* input re-resolves them and dies on the
enum. Only a resume with no inputs at all still works, so the bound
verdict can never be delivered.
"""
import pytest
import yaml as _yaml
from specify_cli.workflows.engine import WorkflowEngine
definition_data = {
"schema_version": "1.0",
"workflow": {"id": "wf", "name": "WF", "version": "1.0.0"},
"inputs": {
"spec_verdict": {"type": "string", "enum": ["approve", "reject"]},
"note": {"type": "string", "default": "a"},
},
"steps": [
{
"id": "review",
"type": "gate",
"message": "Review?",
"options": ["approve", "reject"],
"on_reject": "retry",
"verdict_input": "spec_verdict",
}
],
}
wf_dir = tmp_path / ".specify" / "workflows" / "wf"
wf_dir.mkdir(parents=True)
(wf_dir / "workflow.yml").write_text(
_yaml.safe_dump(definition_data), encoding="utf-8"
)
engine = WorkflowEngine(tmp_path)
definition = engine.load_workflow("wf")
state = engine.execute(definition, inputs={"spec_verdict": "reject"})
assert state.status.value == "paused"
# The retry reset persisted a value the input's own enum forbids.
assert state.inputs["spec_verdict"] == ""
with pytest.raises(ValueError, match="not in allowed values"):
engine.resume(state.run_id, inputs={"note": "b"})
def test_verdict_input_in_switch_case(self):
# Recursion coverage: bad reference inside a switch case must surface.
errors = self._errors("""
@@ -7531,59 +7249,6 @@ class TestWorkflowCatalog:
assert catalog._fetch_single_catalog(entry) == payload
@pytest.mark.parametrize("catalog_type", ["workflow", "step"])
def test_non_utf8_cached_catalog_is_refetched(
self, project_dir, monkeypatch, catalog_type
):
import io
from specify_cli.authentication import http as auth_http
from specify_cli.workflows.catalog import (
StepCatalog,
StepCatalogEntry,
WorkflowCatalog,
WorkflowCatalogEntry,
)
catalog_cls = WorkflowCatalog if catalog_type == "workflow" else StepCatalog
entry_cls = (
WorkflowCatalogEntry
if catalog_type == "workflow"
else StepCatalogEntry
)
payload_key = "workflows" if catalog_type == "workflow" else "steps"
url = f"https://example.com/{catalog_type}.json"
catalog = catalog_cls(project_dir)
cache_path, metadata_path = catalog._get_cache_paths(url)
cache_path.parent.mkdir(parents=True, exist_ok=True)
cache_path.write_bytes(b"\xff\xfe")
metadata_path.write_text(
json.dumps({"fetched_at": 4_102_444_800}),
encoding="utf-8",
)
payload = {"schema_version": "1.0", payload_key: {}}
class _FakeResponse(io.BytesIO):
def geturl(self):
return url
monkeypatch.setattr(
auth_http,
"open_url",
lambda url, timeout=30, redirect_validator=None: _FakeResponse(
json.dumps(payload).encode("utf-8")
),
)
entry = entry_cls(
url=url,
name="test",
priority=1,
install_allowed=True,
)
assert catalog._fetch_single_catalog(entry) == payload
assert json.loads(cache_path.read_text(encoding="utf-8")) == payload
def test_non_mapping_stale_workflow_catalog_is_rejected(
self, project_dir, monkeypatch
):
-12
View File
@@ -104,18 +104,6 @@ def test_fetch_rejects_malformed_source_url_cleanly(url):
fetcher(_source(url))
@pytest.mark.parametrize("use_file_url", [False, True], ids=["path", "file-url"])
def test_local_catalog_decode_errors_are_wrapped(tmp_path, use_file_url):
catalog_path = tmp_path / "catalog.json"
catalog_path.write_bytes(b"\xff\xfe")
url = catalog_path.as_uri() if use_file_url else str(catalog_path)
fetcher = adapters.make_catalog_fetcher(allow_network=False)
with pytest.raises(BundlerError, match="Could not read"):
fetcher(_source(url))
def test_builtin_community_catalog_fetches_repository_catalog_online(monkeypatch):
captured: dict = {}
-26
View File
@@ -207,30 +207,4 @@ def test_executable_bit_preserved_in_artifact(tmp_path: Path):
}
# Executable source -> 0755; plain text files -> 0644.
assert modes["scripts/hook.sh"] == 0o755
def test_toctou_stat_read_consistency(tmp_path: Path):
"""Regression: stat() and read() must use the same file descriptor.
The old implementation called file_path.stat() then file_path.read_bytes()
as separate syscalls. Between the two, another process could replace the
file. The fix opens the file once and uses os.fstat() + fh.read() on the
same handle. This test verifies the archived bytes and mode are consistent.
"""
bundle = _make_bundle(tmp_path / "b")
target = bundle / "assets" / "data.bin"
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(b"\x00\x01\x02\x03")
target.chmod(0o644)
result = build_bundle(bundle, output_dir=tmp_path / "out")
with zipfile.ZipFile(result.artifact_path) as archive:
content = archive.read("assets/data.bin")
modes = {
info.filename: (info.external_attr >> 16) & 0o777
for info in archive.infolist()
}
assert content == b"\x00\x01\x02\x03"
assert modes["assets/data.bin"] == 0o644
assert modes["README.md"] == 0o644