From 5e12d0d3773095618fbd937175e147583b1251c8 Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Mon, 28 Sep 2026 21:26:12 +0800 Subject: [PATCH] feat(init): preview planned file changes before initialization writes Add human-readable and JSON plans for specify init --dry-run. Stage only relevant project and home state, preserve normal initialization behavior, and report creates, overwrites, preserves, skips, conflicts, and component failures. Harden staging against path escapes, symlinks, junctions, unreadable managed files, and permission-only changes. Expand home-relative preset paths consistently in preview and real initialization. Preserve safe nested command aliases during cleanup while rejecting traversal. Document the new flags and the JSON requirement in the public init reference. Assisted-by: Codex (model: GPT-6, autonomous) --- docs/reference/core.md | 5 + src/specify_cli/__init__.py | 7 +- src/specify_cli/_utils.py | 35 + src/specify_cli/agents.py | 155 +- src/specify_cli/bundles/_commands.py | 2 + src/specify_cli/command_init.py | 1472 ++++++++++++- src/specify_cli/extensions/__init__.py | 32 + src/specify_cli/extensions/command_remove.py | 10 +- src/specify_cli/integrations/base.py | 7 + .../integrations/hermes/__init__.py | 71 +- src/specify_cli/presets/_manager.py | 59 +- src/specify_cli/presets/command_remove.py | 13 +- tests/integrations/test_integration_hermes.py | 554 +++++ tests/test_extensions.py | 72 + tests/test_init_dry_run.py | 1884 +++++++++++++++++ tests/test_registrar_path_traversal.py | 13 +- 16 files changed, 4314 insertions(+), 77 deletions(-) create mode 100644 tests/test_init_dry_run.py diff --git a/docs/reference/core.md b/docs/reference/core.md index 026edf7a57..db40f25993 100644 --- a/docs/reference/core.md +++ b/docs/reference/core.md @@ -15,6 +15,8 @@ specify init [] | `--script sh\|ps\|py` | Script type: `sh` (bash/zsh), `ps` (PowerShell), or `py` (Python) | | `--here` | Initialize in the current directory instead of creating a new one | | `--force` | Force merge/overwrite when initializing in an existing directory | +| `--dry-run` | Preview planned initialization changes without writing to the target project | +| `--json` | Emit the dry-run preview as a single JSON document; requires `--dry-run` | | `--ignore-agent-tools` | Skip checks for AI coding agent CLI tools | | `--preset ` | Install a preset during initialization | @@ -44,6 +46,9 @@ specify init my-project --integration copilot --script ps # Install a preset during initialization specify init my-project --integration copilot --preset compliance + +# Preview planned changes as JSON without writing to the project +specify init my-project --integration copilot --dry-run --json ``` ### Environment Variables diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index 8797a1dd0d..34e947fa34 100644 --- a/src/specify_cli/__init__.py +++ b/src/specify_cli/__init__.py @@ -210,10 +210,12 @@ def _install_shared_infra_or_exit( raise typer.Exit(1) -def ensure_executable_scripts(project_path: Path, tracker: StepTracker | None = None) -> None: +def ensure_executable_scripts( + project_path: Path, tracker: StepTracker | None = None +) -> list[str]: """Ensure POSIX .sh scripts under .specify/scripts and .specify/extensions (recursively) have execute bits (no-op on Windows).""" if os.name == "nt": - return # Windows: skip silently + return [] # Windows: skip silently scan_roots = [ project_path / ".specify" / "scripts", project_path / ".specify" / "extensions", @@ -261,6 +263,7 @@ def ensure_executable_scripts(project_path: Path, tracker: StepTracker | None = console.print("[yellow]Some scripts could not be updated:[/yellow]") for f in failures: console.print(f" - {f}") + return failures # --------------------------------------------------------------------------- # Skills directory helpers diff --git a/src/specify_cli/_utils.py b/src/specify_cli/_utils.py index 300ca4ff58..d654f4fe02 100644 --- a/src/specify_cli/_utils.py +++ b/src/specify_cli/_utils.py @@ -19,6 +19,41 @@ DOCKER_AGENT_CHECK_TIMEOUT = 5 +def windows_path_is_junction(path: Path) -> bool: + """Detect a Windows junction using Python 3.11-compatible reparse data.""" + try: + path_stat = path.lstat() + except FileNotFoundError: + return False + except OSError as exc: + raise ValueError( + f"Cannot determine whether path is a junction: {path}: {exc}" + ) from exc + + reparse_tag = getattr(path_stat, "st_reparse_tag", None) + if reparse_tag is None: + raise ValueError(f"Cannot determine whether path is a junction: {path}") + mount_point_tag = getattr(stat, "IO_REPARSE_TAG_MOUNT_POINT", 0xA0000003) + return reparse_tag == mount_point_tag + + +def path_is_junction(path: Path) -> bool: + """Return whether *path* is a junction on supported Python versions.""" + checker = getattr(path, "is_junction", None) + if callable(checker): + try: + return checker() + except AttributeError: + pass + except OSError as exc: + raise ValueError( + f"Cannot determine whether path is a junction: {path}: {exc}" + ) from exc + if os.name == "nt": + return windows_path_is_junction(path) + return False + + def docker_agent_command(executable: str | None = None) -> list[str] | None: """Return a runnable Docker Agent command, or ``None`` if unavailable. diff --git a/src/specify_cli/agents.py b/src/specify_cli/agents.py index 134b4e6d2f..1a6ab73c83 100644 --- a/src/specify_cli/agents.py +++ b/src/specify_cli/agents.py @@ -9,7 +9,7 @@ import os import re from copy import deepcopy -from pathlib import Path +from pathlib import Path, PureWindowsPath from typing import Any, Dict, Iterable, List, Optional import yaml @@ -577,6 +577,34 @@ def _compute_output_name( return f"speckit-{short_name}" + def validate_integration_output_paths( + self, + agent_name: str, + command_names: Iterable[str], + project_root: Path, + ) -> None: + """Run the integration-owned path guard before any output is written.""" + self._ensure_configs() + agent_config = self.AGENT_CONFIGS.get(agent_name) + if not agent_config: + return + + from specify_cli.integrations import get_integration # noqa: PLC0415 + + integration = get_integration(agent_name) + if integration is None: + return + output_root = self._resolve_agent_dir( + agent_name, agent_config, project_root + ) + integration.validate_output_path(output_root, project_root) + for command_name in command_names: + output_name = self._compute_output_name( + agent_name, command_name, agent_config + ) + output_path = output_root / f"{output_name}{agent_config['extension']}" + integration.validate_output_path(output_path, project_root) + @staticmethod def _ensure_inside(candidate: Path, base: Path) -> None: """Validate that a write target stays within the expected base directory. @@ -597,12 +625,20 @@ def _ensure_inside(candidate: Path, base: Path) -> None: if not normalized.is_relative_to(base_normalized): raise ValueError(f"Output path {candidate!r} escapes directory {base!r}") - @staticmethod - def _is_safe_command_name(name: str) -> bool: - """Reject names that could escape the commands directory via path traversal.""" - if os.path.sep in name or "/" in name or "\\" in name: + @classmethod + def _is_safe_command_name(cls, name: str) -> bool: + """Apply registration's containment rule to relative command subpaths.""" + if not isinstance(name, str) or not name: + return False + windows_name = PureWindowsPath(name) + if windows_name.drive or windows_name.root: return False - return os.path.normpath(name) == name + root = Path("_commands") + try: + cls._ensure_inside(root / name.replace("\\", "/"), root) + except ValueError: + return False + return True @staticmethod def _same_lexical_path(left: Path, right: Path) -> bool: @@ -673,6 +709,11 @@ def register_commands( commands_dir = _resolved_dir or self._resolve_agent_dir( agent_name, agent_config, project_root, ) + from specify_cli.integrations import get_integration # noqa: PLC0415 + + _integration = get_integration(agent_name) + if _integration is not None: + _integration.validate_output_path(commands_dir, project_root) commands_dir.mkdir(parents=True, exist_ok=True) registered = [] @@ -699,14 +740,8 @@ def register_commands( # ``.bob/commands``. _sep = agent_config.get("invoke_separator", ".") registrar_writes_skills = agent_config.get("extension") == "/SKILL.md" - try: - from specify_cli.integrations import get_integration # noqa: PLC0415 - - _integ = get_integration(agent_name) - if _integ is not None: - _sep = _integ.invoke_separator_for_mode(registrar_writes_skills) - except (ImportError, ValueError, KeyError): - pass + if _integration is not None: + _sep = _integration.invoke_separator_for_mode(registrar_writes_skills) _prefix = get_invocation_prefix(agent_name, registrar_writes_skills) for cmd_info in commands: @@ -860,18 +895,14 @@ def register_commands( raise ValueError(f"Unsupported format: {agent_config['format']}") # -- Post-process for non-skills agents ----------------------- - _integration = None if agent_config["extension"] != "/SKILL.md": - from specify_cli.integrations import ( # noqa: PLC0415 - get_integration, - ) - - _integration = get_integration(agent_name) if _integration is not None: output = _integration.post_process_command_content(output) dest_file = commands_dir / f"{output_name}{agent_config['extension']}" self._ensure_inside(dest_file, commands_dir) + if _integration is not None: + _integration.validate_output_path(dest_file, project_root) dest_file.parent.mkdir(parents=True, exist_ok=True) self._write_registered_output( dest_file, @@ -954,6 +985,8 @@ def register_commands( commands_dir / f"{alias_output_name}{agent_config['extension']}" ) self._ensure_inside(alias_file, commands_dir) + if _integration is not None: + _integration.validate_output_path(alias_file, project_root) alias_file.parent.mkdir(parents=True, exist_ok=True) self._write_registered_output( alias_file, @@ -1113,6 +1146,9 @@ def register_commands_for_all_agents( Dictionary mapping agent names to list of registered commands """ results = {} + from specify_cli.integrations.base import ( # noqa: PLC0415 + IntegrationOutputPathError, + ) self._ensure_configs() active_skills_agent = ( @@ -1221,6 +1257,8 @@ def register_commands_for_all_agents( active_created_skills_dir = ( recovered_active_skills_dir or agent_dir ) + except IntegrationOutputPathError: + raise except ValueError: continue except OSError: @@ -1271,6 +1309,9 @@ def register_commands_for_non_skill_agents( Dictionary mapping agent names to list of registered commands """ results = {} + from specify_cli.integrations.base import ( # noqa: PLC0415 + IntegrationOutputPathError, + ) self._ensure_configs() extra_agents_set = frozenset(extra_agents) if extra_agents else frozenset() for agent_name, agent_config in self.AGENT_CONFIGS.items(): @@ -1305,6 +1346,8 @@ def register_commands_for_non_skill_agents( ) if registered: results[agent_name] = registered + except IntegrationOutputPathError: + raise except ValueError: continue return results @@ -1324,6 +1367,16 @@ def unregister_commands( project_root: Path to project root """ self._ensure_configs() + from specify_cli.integrations import get_integration # noqa: PLC0415 + from specify_cli.integrations.base import ( # noqa: PLC0415 + IntegrationOutputPathError, + ) + from specify_cli.shared_infra import ( # noqa: PLC0415 + _ensure_safe_shared_destination, + ) + + cleanup_files: list[tuple[Path, Path]] = [] + copilot_prompts: set[Path] = set() for agent_name, cmd_names in registered_commands.items(): if agent_name not in self.AGENT_CONFIGS: continue @@ -1332,6 +1385,9 @@ def unregister_commands( commands_dir = self._resolve_agent_dir( agent_name, agent_config, project_root, ) + integration = get_integration(agent_name) + if integration is not None: + integration.validate_output_path(commands_dir, project_root) # Collect all directories to clean: canonical (or resolved # legacy) plus the legacy dir if it exists separately. @@ -1343,6 +1399,10 @@ def unregister_commands( dirs_to_clean.append(legacy_dir) for cmd_name in cmd_names: + if not self._is_safe_command_name(cmd_name): + raise IntegrationOutputPathError( + f"Unsafe registered command name: {cmd_name!r}" + ) output_name = self._compute_output_name( agent_name, cmd_name, agent_config ) @@ -1360,25 +1420,50 @@ def unregister_commands( self._ensure_inside(cmd_file, target_dir) except ValueError: continue - if cmd_file.exists() or cmd_file.is_symlink(): - cmd_file.unlink() - # For SKILL.md agents each command lives in its own - # subdirectory (e.g. .agents/skills/speckit-ext-cmd/ - # SKILL.md). Remove the parent dir when it becomes - # empty to avoid orphaned directories. - parent = cmd_file.parent - if parent != target_dir and parent.exists(): - try: - parent.rmdir() - except OSError: - pass + if integration is not None: + integration.validate_output_path( + cmd_file, project_root + ) + cleanup_files.append((cmd_file, target_dir)) if agent_name == "copilot": prompt_file = ( - project_root / ".github" / "prompts" / f"{cmd_name}.prompt.md" + project_root + / ".github" + / "prompts" + / f"{cmd_name}.prompt.md" ) - if prompt_file.exists(): - prompt_file.unlink() + try: + _ensure_safe_shared_destination( + project_root, + prompt_file, + parent_must_exist=False, + ) + except ValueError as exc: + raise IntegrationOutputPathError(str(exc)) from exc + copilot_prompts.add(prompt_file) + + seen: set[Path] = set() + for cmd_file, target_dir in cleanup_files: + if cmd_file in seen: + continue + seen.add(cmd_file) + if cmd_file.exists() or cmd_file.is_symlink(): + cmd_file.unlink() + # For SKILL.md agents each command lives in its own + # subdirectory (e.g. .agents/skills/speckit-ext-cmd/ + # SKILL.md). Remove the parent dir when it becomes + # empty to avoid orphaned directories. + parent = cmd_file.parent + if parent != target_dir and parent.exists(): + try: + parent.rmdir() + except OSError: + pass + + for prompt_file in copilot_prompts: + if prompt_file.exists(): + prompt_file.unlink() # Populate AGENT_CONFIGS after class definition. diff --git a/src/specify_cli/bundles/_commands.py b/src/specify_cli/bundles/_commands.py index 5619ecc04b..5d7973e068 100644 --- a/src/specify_cli/bundles/_commands.py +++ b/src/specify_cli/bundles/_commands.py @@ -105,6 +105,8 @@ def _run_init(integration: str, *, script_type: str, offline: bool = False) -> N integration_options=None, extensions=None, trust_extension_urls=False, + dry_run=False, + json_output=False, ) except typer.Exit as exc: if exc.exit_code: diff --git a/src/specify_cli/command_init.py b/src/specify_cli/command_init.py index 459551b36c..5ed24dfbe2 100644 --- a/src/specify_cli/command_init.py +++ b/src/specify_cli/command_init.py @@ -2,11 +2,15 @@ from __future__ import annotations +import hashlib +import json import os import shlex import shutil +import stat import subprocess import sys +import tempfile from pathlib import Path from typing import Any @@ -26,7 +30,10 @@ get_speckit_version, ) from ._console import StepTracker, console, select_with_arrows, show_banner -from ._utils import check_tool +from ._utils import ( + check_tool, + path_is_junction as _path_is_junction, +) def _stdin_is_interactive() -> bool: @@ -53,6 +60,1260 @@ def _ext_spec_is_url(ext_spec: str) -> bool: return False +def _file_fingerprint(path: Path) -> str: + """Return a digest of *path* contents and permission bits.""" + digest = hashlib.sha256() + digest.update(path.read_bytes()) + digest.update(b"\0mode=") + digest.update(format(stat.S_IMODE(path.stat().st_mode), "o").encode()) + return digest.hexdigest() + + +def _snapshot_files(root: Path) -> dict[str, str]: + """Return fingerprints for regular files below *root*.""" + if not root.exists(): + return {} + + files: dict[str, str] = {} + for path in root.rglob("*"): + if not path.is_file() or path.is_symlink(): + continue + files[path.relative_to(root).as_posix()] = _file_fingerprint(path) + return files + + +def _snapshot_tree_entries(root: Path) -> dict[str, str]: + """Return a structural snapshot including directories and symlinks.""" + if not root.exists(): + return {} + + entries: dict[str, str] = {} + for path in root.rglob("*"): + relative = path.relative_to(root).as_posix() + try: + if path.is_symlink(): + entries[relative] = f"symlink:{os.readlink(path)}" + elif path.is_file(): + entries[relative] = f"file:{_file_fingerprint(path)}" + elif path.is_dir(): + entries[relative] = "directory" + except OSError: + entries[relative] = "unreadable" + return entries + + +def _resolve_preview_child_path(spec: str) -> str: + """Resolve caller-relative local specs before changing the child cwd.""" + if spec.startswith(("~", "./", "../", "/", ".\\", "..\\")): + return str(Path(spec).expanduser().resolve()) + return spec + + +def _resolve_preview_preset_path(spec: str) -> str: + """Resolve a local preset directory before the staged child changes cwd.""" + try: + candidate = Path(spec).expanduser().resolve() + except OSError: + return _resolve_preview_child_path(spec) + if candidate.is_dir() and (candidate / "preset.yml").is_file(): + return str(candidate) + return _resolve_preview_child_path(spec) + + +def _preview_subprocess_env(staged_home: Path) -> dict[str, str]: + """Return a child environment with user-scoped paths isolated in staging.""" + env = os.environ.copy() + home = str(staged_home) + env.update( + { + "HOME": home, + "USERPROFILE": home, + "XDG_CACHE_HOME": str(staged_home / ".cache"), + "XDG_CONFIG_HOME": str(staged_home / ".config"), + "XDG_DATA_HOME": str(staged_home / ".local" / "share"), + "XDG_STATE_HOME": str(staged_home / ".local" / "state"), + "APPDATA": str(staged_home / "AppData" / "Roaming"), + "LOCALAPPDATA": str(staged_home / "AppData" / "Local"), + "PYTHONIOENCODING": "utf-8", + "PYTHONUTF8": "1", + } + ) + home_drive, home_path = os.path.splitdrive(home) + if home_drive: + env["HOMEDRIVE"] = home_drive + env["HOMEPATH"] = home_path + else: + env.pop("HOMEDRIVE", None) + env.pop("HOMEPATH", None) + return env + + +def _seed_preview_home(staged_home: Path, real_home: Path) -> None: + """Copy the read-only catalog and auth settings resolved from HOME.""" + for filename in ( + "auth.json", + "extension-catalogs.yml", + "preset-catalogs.yml", + ): + source = real_home / ".specify" / filename + if source.is_file(): + destination = staged_home / ".specify" / filename + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + + +def _preview_home_seed_paths( + real_home: Path, selected_integration: str +) -> tuple[set[Path], set[Path]]: + """Return home paths to stage and selected-integration paths to own.""" + from .agents import CommandRegistrar + from .integrations import get_integration + + integration = get_integration(selected_integration) + if integration is None or not integration.registrar_config: + return set(), set() + + registrar_config = integration.registrar_config + directory = registrar_config.get("dir") + extension = registrar_config.get("extension") + if ( + not isinstance(directory, str) + or not directory.startswith("~") + or not isinstance(extension, str) + ): + return set(), set() + + destination = Path(directory[1:].lstrip("/\\")) + owned_paths: set[Path] = set() + for template in integration.list_command_templates(): + command_name = f"speckit.{template.stem}" + output_name = CommandRegistrar._compute_output_name( + selected_integration, command_name, registrar_config + ) + owned_paths.add(destination / f"{output_name}{extension}") + + paths = set(owned_paths) + real_destination = real_home / destination + if real_destination.is_dir() and not real_destination.is_symlink(): + entries = list(real_destination.iterdir()) + for entry in entries: + if not entry.name.startswith(("speckit-", "speckit.")): + continue + if extension == "/SKILL.md": + paths.add(destination / entry.name / "SKILL.md") + elif entry.name.endswith(extension): + paths.add(destination / entry.name) + return paths, owned_paths + + +_INIT_PLAN_ENV = "SPECIFY_INIT_PLAN_PATH" +_PREVIEW_CHILD_ACTIVE = False + + +def _run_staged_preview_child() -> None: + """Run the CLI with the existing-directory prompt bypass scoped in-process.""" + global _PREVIEW_CHILD_ACTIVE + + from specify_cli import main + + _PREVIEW_CHILD_ACTIVE = True + try: + main() + finally: + _PREVIEW_CHILD_ACTIVE = False + + +def _staging_confirmation_is_accepted() -> bool: + """Return whether a staged preview child may skip its directory prompt.""" + return _PREVIEW_CHILD_ACTIVE and bool(os.environ.get(_INIT_PLAN_ENV)) + + +def _record_init_plan_action( + action: str, + path: str, + provenance: str, + source_id: str | None = None, +) -> None: + """Append one initializer outcome when a preview plan path is configured.""" + plan_path = os.environ.get(_INIT_PLAN_ENV) + if not _PREVIEW_CHILD_ACTIVE or not plan_path: + return + record: dict[str, str] = { + "action": action, + "path": path, + "provenance": provenance, + } + if source_id: + record["source_id"] = source_id + try: + with open(plan_path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(record, sort_keys=True) + "\n") + except OSError as exc: + sys.stderr.write(f"specify: failed to record init plan action: {exc}\n") + + +def _record_init_plan_failure(component: str, source_id: str, error: str) -> None: + """Append an optional component failure when a preview plan is configured.""" + plan_path = os.environ.get(_INIT_PLAN_ENV) + if not _PREVIEW_CHILD_ACTIVE or not plan_path: + return + record = { + "outcome": "failure", + "component": component, + "source_id": source_id, + "error": error, + } + try: + with open(plan_path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(record, sort_keys=True) + "\n") + except OSError as exc: + sys.stderr.write(f"specify: failed to record init plan failure: {exc}\n") + + +def _recorded_plan_failures(plan_path: Path) -> list[dict[str, str]]: + """Read structured optional component failures from a staged preview.""" + if not plan_path.is_file(): + return [] + try: + lines = plan_path.read_text(encoding="utf-8").splitlines() + except OSError: + return [] + + failures: list[dict[str, str]] = [] + for line in lines: + try: + record = json.loads(line) + except (TypeError, ValueError): + continue + if not isinstance(record, dict) or record.get("outcome") != "failure": + continue + component = record.get("component") + source_id = record.get("source_id") + error = record.get("error") + if all(isinstance(value, str) for value in (component, source_id, error)): + failures.append( + { + "component": component, + "source_id": source_id, + "error": error, + } + ) + return failures + + +def _merge_recorded_plan_actions( + actions: list[dict[str, str]], plan_path: Path +) -> list[dict[str, str]]: + """Fold initializer-recorded skip outcomes into the digest-based preview.""" + if not plan_path.is_file(): + return actions + try: + lines = plan_path.read_text(encoding="utf-8").splitlines() + except OSError: + return actions + + by_path = {record["path"]: record for record in actions} + for line in lines: + line = line.strip() + if not line: + continue + try: + recorded = json.loads(line) + except (TypeError, ValueError): + continue + if not isinstance(recorded, dict): + continue + if recorded.get("action") != "skip" or not isinstance(recorded.get("path"), str): + continue + path = recorded["path"] + existing = by_path.get(path) + if existing is not None and existing.get("action") != "preserve": + continue + merged: dict[str, str] = { + "action": "skip", + "path": path, + "provenance": str(recorded.get("provenance") or "core"), + } + source_id = recorded.get("source_id") + if source_id: + merged["source_id"] = str(source_id) + by_path[path] = merged + return list(by_path.values()) + + +PreviewOwnership = tuple[str, str | None] + + +def _preview_manifest_ownership(staged_root: Path) -> dict[str, PreviewOwnership]: + """Map manifest-tracked staged paths to provenance category and source ID.""" + ownership: dict[str, PreviewOwnership] = {} + manifests = staged_root / ".specify" / "integrations" + if not manifests.is_dir(): + return ownership + + for manifest_path in manifests.glob("*.manifest.json"): + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + key = str( + manifest.get( + "integration", + manifest.get("key", manifest_path.stem.removesuffix(".manifest")), + ) + ) + source = ("core", key) if key == "speckit" else ("integration", key) + for relative_path in manifest.get("files", {}): + ownership[str(relative_path)] = source + except (OSError, TypeError, ValueError): + continue + return ownership + + +def _preview_registry_entries(staged_root: Path) -> list[tuple[str, dict[str, Any]]]: + """Load valid source entries from staged extension and preset registries.""" + registries: list[tuple[str, dict[str, Any]]] = [] + registry_specs = ( + ( + "extension", + staged_root / ".specify" / "extensions" / ".registry", + "extensions", + ), + ("preset", staged_root / ".specify" / "presets" / ".registry", "presets"), + ) + for category, registry_path, collection_key in registry_specs: + try: + data = json.loads(registry_path.read_text(encoding="utf-8")) + entries = data.get(collection_key, {}) + except (AttributeError, OSError, TypeError, ValueError): + continue + if not isinstance(entries, dict): + continue + registries.append( + ( + category, + { + source_id: metadata + for source_id, metadata in entries.items() + if isinstance(source_id, str) and isinstance(metadata, dict) + }, + ) + ) + return registries + + +def _preview_registry_sources(staged_root: Path) -> dict[str, set[str]]: + """Return source ID to provenance categories from staged registries.""" + sources: dict[str, set[str]] = {} + for category, entries in _preview_registry_entries(staged_root): + for source_id in entries: + sources.setdefault(source_id, set()).add(category) + return sources + + +def _preview_registry_ownership( + staged_root: Path, +) -> tuple[dict[str, PreviewOwnership], dict[str, PreviewOwnership]]: + """Map registered command outputs in project and home staging scopes.""" + from .agents import CommandRegistrar + + registrar = CommandRegistrar() + project_ownership: dict[str, PreviewOwnership] = {} + home_ownership: dict[str, PreviewOwnership] = {} + for category, entries in _preview_registry_entries(staged_root): + for source_id, metadata in entries.items(): + registered = metadata.get("registered_commands", {}) + if not isinstance(registered, dict): + continue + for agent_name, command_names in registered.items(): + agent_config = registrar.AGENT_CONFIGS.get(agent_name) + if not isinstance(agent_config, dict) or not isinstance( + command_names, list + ): + continue + dir_value = agent_config.get("dir") + extension = agent_config.get("extension") + if not isinstance(dir_value, str) or not isinstance(extension, str): + continue + if dir_value.startswith("~"): + destination = Path(dir_value[1:].lstrip("/")) + scope = home_ownership + else: + destination = Path(dir_value) + if destination.is_absolute(): + continue + canonical = staged_root / destination + legacy = agent_config.get("legacy_dir") + if ( + not canonical.exists() + and isinstance(legacy, str) + and (staged_root / legacy).exists() + ): + destination = Path(legacy) + scope = project_ownership + for command_name in command_names: + if not isinstance(command_name, str): + continue + output_name = registrar._compute_output_name( + agent_name, command_name, agent_config + ) + relative_path = ( + destination / f"{output_name}{extension}" + ).as_posix() + scope[relative_path] = category, source_id + if agent_name == "copilot": + prompt_path = ( + Path(".github") / "prompts" / f"{command_name}.prompt.md" + ).as_posix() + project_ownership[prompt_path] = category, source_id + return project_ownership, home_ownership + + +def _preview_content_ownership( + content: str, registry_sources: dict[str, set[str]] +) -> PreviewOwnership | None: + """Read generated ownership markers, using registries to type bare IDs.""" + bare_source_id: str | None = None + for line in content.splitlines(): + marker = line.strip() + if marker.startswith("source:"): + value = marker.removeprefix("source:").strip().strip("\"'") + if value.startswith(("preset:", "extension:")): + category, source_id = value.split(":", 1) + source_id = source_id.split(":", 1)[0] + if category in registry_sources.get(source_id, set()): + return category, source_id + if marker.startswith(""): + value = marker.removeprefix("").strip() + if value.startswith(("preset:", "extension:")): + category, source_id = value.split(":", 1) + if category in registry_sources.get(source_id, set()): + return category, source_id + if value.startswith("Source:"): + bare_source_id = value.removeprefix("Source:").strip() + elif marker.startswith("# Source:"): + bare_source_id = marker.removeprefix("# Source:").strip() + + categories = registry_sources.get(bare_source_id or "", set()) + if len(categories) == 1 and bare_source_id: + return next(iter(categories)), bare_source_id + if bare_source_id: + for category in ("preset", "extension"): + if category in categories and f"{category}:{bare_source_id}" in content: + return category, bare_source_id + return None + + +def _preview_marker_ownership( + staged_root: Path, + registry_sources: dict[str, set[str]], + relative_paths: set[str] | None = None, +) -> dict[str, PreviewOwnership]: + """Map staged generated artifacts using their embedded ownership markers.""" + ownership: dict[str, PreviewOwnership] = {} + paths = ( + relative_paths + if relative_paths is not None + else set(_snapshot_files(staged_root)) + ) + for relative_path in paths: + path = staged_root / relative_path + try: + content = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + source = _preview_content_ownership(content, registry_sources) + if source is not None: + ownership[relative_path] = source + return ownership + + +def _preview_default_ownership( + relative_path: str, default: PreviewOwnership +) -> PreviewOwnership: + if ( + relative_path.startswith(".specify/integrations/") + and relative_path.endswith(".manifest.json") + ): + integration_id = Path(relative_path).name.removesuffix(".manifest.json") + if integration_id == "speckit": + return "core", "speckit" + return "integration", integration_id + if relative_path.startswith(".specify/workflows/"): + remainder = relative_path.removeprefix(".specify/workflows/") + workflow_id = remainder.split("/", 1)[0] + return "workflow", workflow_id if "/" in remainder else None + if relative_path.startswith(".specify/extensions/"): + remainder = relative_path.removeprefix(".specify/extensions/") + extension_id = remainder.split("/", 1)[0] + return "extension", extension_id if "/" in remainder else None + if relative_path.startswith(".specify/presets/"): + remainder = relative_path.removeprefix(".specify/presets/") + preset_id = remainder.split("/", 1)[0] + return "preset", preset_id if "/" in remainder else None + if relative_path.startswith(".specify/"): + return "core", None + return default + + +def _constitution_plan_ownership(project_path: Path) -> PreviewOwnership: + """Read the materialized constitution's source from its sidecar.""" + provenance = ( + project_path / ".specify" / "memory" / ".constitution-template.json" + ) + try: + metadata = json.loads(provenance.read_text(encoding="utf-8")) + except (OSError, TypeError, ValueError): + return "core", None + if not isinstance(metadata, dict): + return "core", None + + source = metadata.get("source") + if not isinstance(source, str): + return "core", None + if source.startswith("extension:"): + source_id = source.removeprefix("extension:").split(" ", 1)[0] + return "extension", source_id or None + if source in {"core", "core (bundled)", "project override"}: + return "core", None + source_id = source.split(" v", 1)[0].strip() + return ("preset", source_id) if source_id else ("core", None) + + +def _build_preview_actions( + initial_files: dict[str, str], + staged_root: Path, + *, + path_prefix: str = "", + ownership: dict[str, PreviewOwnership] | None = None, + default_ownership: PreviewOwnership = ("integration", None), + directory_conflict: bool = False, + staged_files: dict[str, str] | None = None, +) -> list[dict[str, str]]: + """Classify files produced by a staged initialization.""" + if staged_files is None: + staged_files = _snapshot_files(staged_root) + ownership = ownership or {} + candidates = { + path + for path, digest in staged_files.items() + if initial_files.get(path) != digest + } + candidates.update(path for path in initial_files if path not in staged_files) + candidates.update(path for path in ownership if path in staged_files) + + actions: list[dict[str, str]] = [] + for path in sorted(candidates): + staged_digest = staged_files.get(path) + initial_digest = initial_files.get(path) + if staged_digest is None: + action = "remove" + elif initial_digest is None: + action = "create" + elif initial_digest != staged_digest: + action = "conflict" if directory_conflict else "overwrite" + else: + action = "preserve" + provenance, source_id = ownership.get( + path, _preview_default_ownership(path, default_ownership) + ) + record = { + "action": action, + "path": f"{path_prefix}{path}", + "provenance": provenance, + } + if source_id: + record["source_id"] = source_id + actions.append(record) + return actions + + +def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None: + """Render a stable human or machine-readable initialization preview.""" + if json_output: + typer.echo(json.dumps(payload, sort_keys=True)) + return + + console.print("\n[bold cyan]Initialization preview[/bold cyan]") + if payload.get("gate") == "force_required": + console.print( + "[yellow]conflict[/yellow] target directory exists; applying this plan requires --force" + ) + elif payload.get("gate") == "confirmation_required": + console.print("[yellow]confirmation required[/yellow] target directory is not empty") + if payload.get("error"): + console.print( + f"[red]failed[/red] {_escape_markup(str(payload['error']))}" + ) + for failure in payload["failures"]: + component = _escape_markup(str(failure["component"])) + source_id = _escape_markup(str(failure["source_id"])) + error = _escape_markup(str(failure["error"])) + console.print( + f"failed {component}:{source_id} {error}" + ) + for record in payload["actions"]: + action = _escape_markup(str(record["action"])) + path = _escape_markup(str(record["path"])) + source = str(record["provenance"]) + if record.get("source_id"): + source = f"{source}:{record['source_id']}" + console.print( + f"{action:<10} {path} [dim]({_escape_markup(source)})[/dim]" + ) + + +def _raise_dry_run_json_error( + message: str, + *, + project_name: str | None, + here: bool, +) -> None: + """Emit one stable JSON error document for parent-side validation.""" + if here or project_name == ".": + target: str | None = str(Path.cwd()) + elif project_name: + target = str(Path(project_name).resolve()) + else: + target = None + _emit_dry_run_preview( + { + "dry_run": True, + "target": target, + "conflict": False, + "gate": "none", + "actions": [], + "failures": [], + "error": message, + }, + json_output=True, + ) + raise typer.Exit(1) + + +def _strip_windows_extended_prefix(text: str) -> str: + """Strip the ``\\\\?\\`` / ``//?/`` prefix Windows adds to long paths.""" + if text.startswith("\\\\?\\"): + text = text[4:] + if text[:4].upper() == "UNC\\": + return "\\\\" + text[4:] + return text + if text.startswith("//?/"): + text = text[4:] + if text[:4].upper() == "UNC/": + return "//" + text[4:] + return text + return text + + +def _normalize_fs_path(path: Path) -> Path: + """Resolve *path* and drop Windows extended prefixes so containment works.""" + text = _strip_windows_extended_prefix(os.fsdecode(os.fspath(path))) + path = Path(text) + try: + path = path.resolve() + except (OSError, RuntimeError): + pass + text = _strip_windows_extended_prefix(os.fsdecode(os.fspath(path))) + if os.name == "nt": + text = os.path.normcase(text) + return Path(text) + + +def _is_within_root(path: Path, root: Path) -> bool: + path_text = os.fspath(_normalize_fs_path(path)) + root_text = os.fspath(_normalize_fs_path(root)) + try: + common = os.path.commonpath((path_text, root_text)) + except ValueError: + return False + if os.name == "nt": + return os.path.normcase(common) == os.path.normcase(root_text) + return common == root_text + + +def _symlink_target(path: Path) -> Path | None: + raw = _strip_windows_extended_prefix(os.fsdecode(os.readlink(path))) + raw_target = Path(raw) + if not raw_target.is_absolute(): + raw_target = path.parent / raw_target + return _normalize_fs_path(raw_target) + + +def _iter_symlinks(root: Path) -> list[Path]: + found: list[Path] = [] + for dirpath, dirnames, filenames in os.walk(root, followlinks=False): + for name in (*dirnames, *filenames): + candidate = Path(dirpath) / name + if candidate.is_symlink(): + found.append(candidate) + return found + + +def _remap_symlink_to_staged( + path: Path, project_root: Path, staged_root: Path +) -> None: + target = _symlink_target(path) + if target is None: + _quarantine_symlink(path, staged_root) + return + try: + relative = os.path.relpath(os.fspath(target), os.fspath(project_root)) + except ValueError: + _quarantine_symlink(path, staged_root) + return + relative_path = Path(relative) + if relative_path.is_absolute() or ".." in relative_path.parts: + _quarantine_symlink(path, staged_root) + return + remapped = staged_root / relative_path + was_dir = path.is_dir() + path.unlink() + path.symlink_to(remapped, target_is_directory=was_dir) + + +def _quarantine_root(staged_root: Path) -> Path: + return _normalize_fs_path( + staged_root.parent / f"{staged_root.name}-quarantine" + ) + + +def _quarantine_symlink(path: Path, staged_root: Path) -> None: + """Retarget an external symlink at an isolated dummy outside *staged_root*. + + The dummy stays outside the staged project so ``Path.resolve()`` still + escapes, matching real init containment checks, while writes cannot reach + the original live target. + """ + dummy = _quarantine_root(staged_root) / path.relative_to(staged_root) + dummy.parent.mkdir(parents=True, exist_ok=True) + was_dir = path.is_dir() + path.unlink() + if was_dir: + dummy.mkdir(parents=True, exist_ok=True) + else: + dummy.touch() + path.symlink_to(dummy, target_is_directory=was_dir) + + +def _remap_in_project_symlinks(project_root: Path, staged_root: Path) -> None: + """Keep staged symlinks from pointing at the live project or external paths. + + In-project absolute links are retargeted at the staged copy. Links that + resolve outside the project are retargeted at an isolated dummy outside + staging so containment checks still fail, without writing through to the + live target. + """ + project_root = _normalize_fs_path(project_root) + staged_root = _normalize_fs_path(staged_root) + quarantine_root = _quarantine_root(staged_root) + for _ in range(32): + changed = False + for path in _iter_symlinks(staged_root): + resolved = _symlink_target(path) + if resolved is not None and ( + _is_within_root(resolved, staged_root) + or _is_within_root(resolved, quarantine_root) + ): + continue + if resolved is not None and _is_within_root(resolved, project_root): + _remap_symlink_to_staged(path, project_root, staged_root) + changed = True + continue + _quarantine_symlink(path, staged_root) + changed = True + if not changed: + return + raise RuntimeError("staged symlink isolation did not converge") + + +def _preview_path_is_managed(path: Path, copy_root: Path) -> bool: + """Return whether an unreadable path may affect initialization behavior.""" + try: + relative = path.relative_to(copy_root) + except ValueError: + return True + parts = (copy_root.name, *relative.parts) + if copy_root.name == ".specify" and relative.parts: + return relative.parts[0] in { + ".gitignore", + "extensions", + "extensions.yml", + "init-options.json", + "integration.json", + "integrations", + "memory", + "presets", + "scripts", + "templates", + "workflows", + } + return any(part.startswith(("speckit-", "speckit.")) for part in parts) + + +def _path_is_readable_for_staging(path: Path) -> bool: + """Probe whether copytree can read a regular file or enumerate a directory.""" + try: + if path.is_file(): + with path.open("rb"): + pass + elif path.is_dir(): + with os.scandir(path): + pass + return True + except OSError: + return False + + +def _ignore_special_files( + directory: str, + names: list[str], + *, + copy_root: Path | None = None, +) -> set[str]: + """Skip inaccessible unrelated entries and unsupported filesystem nodes.""" + ignored: set[str] = set() + root = copy_root or Path(directory) + for name in names: + candidate = Path(directory) / name + try: + if _path_is_junction(candidate): + raise ValueError( + f"Preview staging refuses Windows directory junction: {candidate}" + ) + if ( + not candidate.is_symlink() + and not candidate.is_file() + and not candidate.is_dir() + ): + ignored.add(name) + continue + if ( + not candidate.is_symlink() + and not _path_is_readable_for_staging(candidate) + and not _preview_path_is_managed(candidate, root) + ): + ignored.add(name) + except OSError: + ignored.add(name) + return ignored + + +def _copy_staged_path(source: Path, destination: Path) -> None: + """Copy one selected path without following symlinks.""" + if _path_is_junction(source): + raise ValueError( + f"Preview staging refuses Windows directory junction: {source}" + ) + if source.is_symlink(): + if os.path.lexists(destination): + return + destination.parent.mkdir(parents=True, exist_ok=True) + destination.symlink_to( + os.readlink(source), target_is_directory=source.is_dir() + ) + elif source.is_dir(): + shutil.copytree( + source, + destination, + symlinks=True, + dirs_exist_ok=True, + ignore=lambda directory, names: _ignore_special_files( + directory, names, copy_root=source + ), + ) + elif source.is_file(): + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + + +def _copy_selected_staged_path( + project_path: Path, + staged_root: Path, + relative_path: Path, +) -> None: + """Copy a selected path while preserving any symlinked parent component.""" + current = Path() + for index, part in enumerate(relative_path.parts): + current /= part + source = project_path / current + destination = staged_root / current + if _path_is_junction(source): + raise ValueError( + f"Preview staging refuses Windows directory junction: {source}" + ) + if source.is_symlink(): + _copy_staged_path(source, destination) + return + if not source.exists(): + return + if index < len(relative_path.parts) - 1 and not source.is_dir(): + _copy_staged_path(source, destination) + return + _copy_staged_path(project_path / relative_path, staged_root / relative_path) + + +def _copy_staged_symlink_targets(project_path: Path, staged_root: Path) -> None: + """Copy project-local targets reached by selected staged symlinks.""" + for _ in range(32): + copied = False + for staged_link in _iter_symlinks(staged_root): + relative_link = staged_link.relative_to(staged_root) + source_link = project_path / relative_link + try: + source_target = _symlink_target(source_link) + except OSError: + continue + if source_target is None or not _is_within_root( + source_target, project_path + ): + continue + relative_target = source_target.relative_to( + _normalize_fs_path(project_path) + ) + staged_target = staged_root / relative_target + if os.path.lexists(staged_target): + continue + _copy_staged_path(project_path / relative_target, staged_target) + copied = True + if not copied: + return + raise RuntimeError("staged symlink target copy did not converge") + + +def _stage_project_copy( + project_path: Path, + staged_root: Path, + relative_paths: set[Path] | None = None, +) -> None: + """Copy selected project paths into staging and isolate live symlinks.""" + if relative_paths is None: + shutil.copytree( + project_path, + staged_root, + symlinks=True, + ignore=lambda directory, names: _ignore_special_files( + directory, names, copy_root=project_path + ), + ) + else: + staged_root.mkdir(parents=True, exist_ok=True) + selected: list[Path] = [] + for relative_path in sorted(relative_paths, key=lambda path: len(path.parts)): + if relative_path.is_absolute() or ".." in relative_path.parts: + continue + if any( + relative_path == parent or parent in relative_path.parents + for parent in selected + ): + continue + selected.append(relative_path) + _copy_selected_staged_path(project_path, staged_root, relative_path) + _copy_staged_symlink_targets(project_path, staged_root) + _remap_in_project_symlinks(project_path.resolve(), staged_root.resolve()) + + +def _preview_seed_paths( + selected_integration: str, + integration_options: str | None, + script_type: str, +) -> set[Path]: + """Return project paths whose existing state can affect initialization.""" + from .integrations import INTEGRATION_REGISTRY, get_integration + + integration = get_integration(selected_integration) + paths = {Path(".specify")} + if integration is None: + return paths + + config = integration.config or {} + registrar = integration.registrar_config or {} + folder = config.get("folder") + commands_subdir = config.get("commands_subdir") + values = [ + registrar.get("dir"), + registrar.get("legacy_dir"), + registrar.get("detect_dir"), + getattr(integration, "legacy_flat_command_dir", None), + ".opencode/plugin/speckit-events.ts", + ] + values.extend( + getattr(candidate, "events_config_file", None) + for candidate in INTEGRATION_REGISTRY.values() + ) + if isinstance(folder, str) and isinstance(commands_subdir, str): + values.append(str(Path(folder) / commands_subdir)) + + if selected_integration == "generic" and integration_options: + resolver = getattr(integration, "_resolve_commands_dir", None) + if callable(resolver): + try: + values.append(resolver(None, {"raw_options": integration_options})) + except (TypeError, ValueError): + pass + + extras = { + "bob": (".bob/skills",), + "copilot": ( + ".github/skills", + ".github/prompts", + ".vscode/settings.json", + ), + "kimi": (".kimi/skills",), + "rovodev": (".rovodev/prompts", ".rovodev/prompts.yml"), + } + values.extend(extras.get(selected_integration, ())) + if script_type == "py": + values.extend((".venv/bin/python", ".venv/Scripts/python.exe")) + + for value in values: + if not isinstance(value, str) or value.startswith("~"): + continue + relative_path = Path(value) + if ( + relative_path == Path(".") + or relative_path.is_absolute() + or ".." in relative_path.parts + ): + continue + paths.add(relative_path) + return paths + + +def _preview_child_failure_message(result: subprocess.CompletedProcess[str]) -> str: + """Extract the initializer failure from captured child output.""" + combined = " ".join( + part.strip().replace("\n", " ") + for part in (result.stderr, result.stdout) + if part + ) + combined = " ".join( + "".join( + " " if "\u2500" <= char <= "\u257f" else char for char in combined + ).split() + ) + marker = "Initialization failed: " + if marker in combined: + combined = combined[combined.index(marker) + len(marker) :] + elif "escapes project root" in combined: + start = combined.find("Integration destination") + if start >= 0: + combined = combined[start:] + return combined[:500] + + +def _preview_init( + *, + project_path: Path, + gate: str, + force: bool, + here: bool, + script_type: str, + selected_integration: str, + ignore_agent_tools: bool, + preset: str | None, + integration_options: str | None, + extensions: list[str] | None, + trust_extension_urls: bool, + json_output: bool, +) -> None: + """Run the canonical initializer in staging and report its file plan.""" + payload: dict[str, Any] = { + "dry_run": True, + "target": str(project_path), + "conflict": gate != "none", + "gate": gate, + "actions": [], + "failures": [], + } + + real_home = Path.home() + url_extensions = [spec for spec in extensions or [] if _ext_spec_is_url(spec)] + staged_extensions = [spec for spec in extensions or [] if not _ext_spec_is_url(spec)] + + with tempfile.TemporaryDirectory(prefix="specify-init-preview-") as tmp_dir: + staged_root = Path(tmp_dir) / "project" + staged_home = Path(tmp_dir) / "home" + staged_home.mkdir() + try: + home_seed_paths, home_owned_paths = _preview_home_seed_paths( + real_home, selected_integration + ) + _seed_preview_home(staged_home, real_home) + if home_seed_paths: + _stage_project_copy(real_home, staged_home, home_seed_paths) + if project_path.exists(): + _stage_project_copy( + project_path, + staged_root, + _preview_seed_paths( + selected_integration, + integration_options, + script_type, + ), + ) + else: + staged_root.mkdir() + except (OSError, RuntimeError, ValueError, shutil.Error) as exc: + payload["error"] = f"failed to stage preview inputs: {exc}" + _emit_dry_run_preview(payload, json_output=json_output) + raise typer.Exit(1) from None + + initial_project_files = _snapshot_files(staged_root) + initial_home_files = _snapshot_files(staged_home) + initial_registry_sources = _preview_registry_sources(staged_root) + initial_project_ownership = _preview_manifest_ownership(staged_root) + initial_registry_project, initial_registry_home = ( + _preview_registry_ownership(staged_root) + ) + initial_project_ownership.update(initial_registry_project) + initial_project_ownership.update( + _preview_marker_ownership( + staged_root, + initial_registry_sources, + set(initial_project_files), + ) + ) + initial_home_ownership = dict(initial_registry_home) + initial_home_ownership.update( + _preview_marker_ownership( + staged_home, + initial_registry_sources, + set(initial_home_files), + ) + ) + for relative_path in home_owned_paths: + initial_home_ownership.setdefault( + relative_path.as_posix(), + ("integration", selected_integration), + ) + quarantine_states = { + "project": _snapshot_tree_entries(_quarantine_root(staged_root)), + "home": _snapshot_tree_entries(_quarantine_root(staged_home)), + } + + # Run the same public CLI path in a child process. Besides preventing + # mutations of the target root, this isolates Rich's Live output from + # the preview's human/JSON output contract. The staging-only + # confirmation signal avoids a prompt without changing force mode. + command = [ + sys.executable, + "-I", + "-c", + ( + "from specify_cli.command_init import " + "_run_staged_preview_child; _run_staged_preview_child()" + ), + "init", + "--non-interactive", + "--integration", + selected_integration, + "--script", + script_type, + ] + if here: + command.append("--here") + else: + command.append(str(staged_root)) + if force or gate == "force_required": + command.append("--force") + command.append("--ignore-agent-tools") + if integration_options: + command.extend(["--integration-options", integration_options]) + if preset: + command.extend(["--preset", _resolve_preview_preset_path(preset)]) + for extension in staged_extensions: + command.extend(["--extension", _resolve_preview_child_path(extension)]) + if trust_extension_urls: + command.append("--trust-extension-urls") + + plan_path = Path(tmp_dir) / "init-plan.jsonl" + env = _preview_subprocess_env(staged_home) + env[_INIT_PLAN_ENV] = str(plan_path) + result = subprocess.run( + command, + cwd=staged_root if here else Path.cwd(), + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + check=False, + env=env, + ) + if result.returncode: + payload["error"] = _preview_child_failure_message(result) + elif any( + _snapshot_tree_entries( + _quarantine_root(staged_root if scope == "project" else staged_home) + ) + != before + for scope, before in quarantine_states.items() + ): + payload["error"] = ( + "staged initialization attempted to write through an external " + "symlink" + ) + else: + staged_project_files = _snapshot_files(staged_root) + registry_sources = _preview_registry_sources(staged_root) + project_ownership = dict(initial_project_ownership) + project_ownership.update(_preview_manifest_ownership(staged_root)) + registry_project_ownership, registry_home_ownership = ( + _preview_registry_ownership(staged_root) + ) + project_ownership.update(registry_project_ownership) + project_ownership.update( + _preview_marker_ownership( + staged_root, registry_sources, set(staged_project_files) + ) + ) + payload["actions"] = _build_preview_actions( + initial_project_files, + staged_root, + ownership=project_ownership, + default_ownership=("integration", selected_integration), + directory_conflict=gate == "force_required", + staged_files=staged_project_files, + ) + staged_home_files = _snapshot_files(staged_home) + home_ownership = dict(initial_home_ownership) + home_ownership.update(registry_home_ownership) + home_ownership.update( + _preview_marker_ownership( + staged_home, registry_sources, set(staged_home_files) + ) + ) + payload["actions"].extend( + _build_preview_actions( + initial_home_files, + staged_home, + path_prefix="~/", + ownership=home_ownership, + default_ownership=("integration", selected_integration), + directory_conflict=False, + staged_files=staged_home_files, + ) + ) + payload["actions"] = _merge_recorded_plan_actions( + payload["actions"], plan_path + ) + payload["failures"] = _recorded_plan_failures(plan_path) + + if not payload.get("error"): + for spec in url_extensions: + payload["actions"].append( + { + "action": "unresolved", + "path": spec, + "provenance": "extension", + "source_id": spec, + "reason": "URL extensions are not fetched during dry-run", + } + ) + payload["actions"].sort(key=lambda action: action["path"]) + _emit_dry_run_preview(payload, json_output=json_output) + if payload.get("error"): + raise typer.Exit(1) + + def _confirm_extension_url_trust( url_specs: list[str], *, @@ -142,6 +1403,12 @@ def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_ve bundled_path = _locate_bundled_extension(ext_spec) if bundled_path is not None: if manager.registry.is_installed(ext_spec): + _record_init_plan_action( + "skip", + f".specify/extensions/{ext_spec}/extension.yml", + "extension", + ext_spec, + ) return "already installed" manifest = manager.install_from_directory(bundled_path, speckit_version) return f"{manifest.name} v{manifest.version} installed" @@ -159,6 +1426,12 @@ def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_ve bundled_path = _locate_bundled_extension(resolved_id) if bundled_path is not None: if manager.registry.is_installed(resolved_id): + _record_init_plan_action( + "skip", + f".specify/extensions/{resolved_id}/extension.yml", + "extension", + resolved_id, + ) return "already installed" manifest = manager.install_from_directory(bundled_path, speckit_version) return f"{manifest.name} v{manifest.version} installed" @@ -227,6 +1500,13 @@ def ensure_constitution_from_template( if tracker: tracker.add("constitution", "Constitution setup") tracker.skip("constitution", "existing file preserved") + provenance, source_id = _constitution_plan_ownership(project_path) + _record_init_plan_action( + "skip", + ".specify/memory/constitution.md", + provenance, + source_id, + ) return try: @@ -237,6 +1517,9 @@ def ensure_constitution_from_template( if tracker: tracker.add("constitution", "Constitution setup") tracker.error("constitution", "template not found") + _record_init_plan_failure( + "constitution", "constitution", "template not found" + ) return if tracker: tracker.add("constitution", "Constitution setup") @@ -254,6 +1537,7 @@ def ensure_constitution_from_template( console.print( f"[yellow]Warning: Could not initialize constitution: {e}[/yellow]" ) + _record_init_plan_failure("constitution", "constitution", str(e)) def register(app: typer.Typer) -> None: @@ -340,6 +1624,16 @@ def init( "--trust-extension-urls", help="Pre-authorize installing extensions from external URLs without the interactive trust prompt (required for non-interactive URL installs).", ), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="Preview initialization changes without writing to the target project.", + ), + json_output: bool = typer.Option( + False, + "--json", + help="Emit the dry-run preview as a single JSON document.", + ), ): """ Initialize a new Specify project. @@ -395,13 +1689,23 @@ def init( _write_integration_json, ) - show_banner() + if not (dry_run and json_output): + show_banner() + + if json_output and not dry_run: + console.print("[red]Error:[/red] --json requires --dry-run") + raise typer.Exit(1) from .integrations import INTEGRATION_REGISTRY, get_integration if integration: resolved_integration = get_integration(integration) if not resolved_integration: + message = f"Unknown integration: '{integration}'" + if dry_run and json_output: + _raise_dry_run_json_error( + message, project_name=project_name, here=here + ) console.print( f"[red]Error:[/red] Unknown integration: " f"'{_escape_markup(str(integration))}'" @@ -415,18 +1719,33 @@ def init( project_name = None if here and project_name: + if dry_run and json_output: + _raise_dry_run_json_error( + "Cannot specify both project name and --here flag", + project_name=project_name, + here=here, + ) console.print( "[red]Error:[/red] Cannot specify both project name and --here flag" ) raise typer.Exit(1) if not here and not project_name: + if dry_run and json_output: + _raise_dry_run_json_error( + "Must specify either a project name, use '.' for current " + "directory, or use --here flag", + project_name=project_name, + here=here, + ) console.print( "[red]Error:[/red] Must specify either a project name, use '.' for current directory, or use --here flag" ) raise typer.Exit(1) dir_existed_before = False + directory_conflict = False + staging_confirmation_accepted = _staging_confirmation_is_accepted() if here: project_name = Path.cwd().name project_path = Path.cwd() @@ -434,25 +1753,29 @@ def init( existing_items = list(project_path.iterdir()) if existing_items: - console.print( - f"[yellow]Warning:[/yellow] Current directory is not empty ({len(existing_items)} items)" - ) - if force: - # Proceeding: the merge/overwrite warning is accurate here. + if not (dry_run and json_output): console.print( - "[yellow]Template files will be merged with existing content and may overwrite existing files[/yellow]" + f"[yellow]Warning:[/yellow] Current directory is not empty ({len(existing_items)} items)" ) - console.print( - "[cyan]--force supplied: skipping confirmation and proceeding with merge[/cyan]" - ) - elif non_interactive: + if dry_run and not force: + directory_conflict = True + elif force: + # Proceeding: the merge/overwrite warning is accurate here. + if not (dry_run and json_output): + console.print( + "[yellow]Template files will be merged with existing content and may overwrite existing files[/yellow]" + ) + console.print( + "[cyan]--force supplied: skipping confirmation and proceeding with merge[/cyan]" + ) + elif non_interactive and not staging_confirmation_accepted: console.print( "[red]Error:[/red] Current directory is not empty and " "--non-interactive was set. Re-run with " "[bold]--force[/bold] to merge into it." ) raise typer.Exit(1) - else: + elif not staging_confirmation_accepted: # Fold the merge risk into the confirmation prompt rather than # printing it unconditionally first: on the EOF/no-input path # below the command exits without changing anything, so a @@ -491,23 +1814,32 @@ def init( if project_path.exists(): safe_name = _escape_markup(str(project_name)) if not project_path.is_dir(): + if dry_run and json_output: + _raise_dry_run_json_error( + f"'{project_name}' exists but is not a directory", + project_name=project_name, + here=here, + ) console.print( f"[red]Error:[/red] '{safe_name}' exists but is not a directory." ) raise typer.Exit(1) existing_items = list(project_path.iterdir()) - if force: - if existing_items: + if dry_run and not force: + directory_conflict = True + elif force: + if existing_items and not (dry_run and json_output): console.print( f"[yellow]Warning:[/yellow] Directory '{safe_name}' is not empty ({len(existing_items)} items)" ) console.print( "[yellow]Template files will be merged with existing content and may overwrite existing files[/yellow]" ) - console.print( - f"[cyan]--force supplied: merging into existing directory '[cyan]{safe_name}[/cyan]'[/cyan]" - ) - else: + if not (dry_run and json_output): + console.print( + f"[cyan]--force supplied: merging into existing directory '[cyan]{safe_name}[/cyan]'[/cyan]" + ) + elif not staging_confirmation_accepted: error_panel = Panel( f"Directory already exists: '[cyan]{safe_name}[/cyan]'\n" "Please choose a different project name or remove the existing directory.\n" @@ -522,6 +1854,12 @@ def init( if integration: if integration not in AGENT_CONFIG: + if dry_run and json_output: + _raise_dry_run_json_error( + f"Invalid integration '{integration}'", + project_name=project_name, + here=here, + ) console.print( f"[red]Error:[/red] Invalid integration '{_escape_markup(str(integration))}'. Choose from: {', '.join(AGENT_CONFIG.keys())}" ) @@ -529,10 +1867,11 @@ def init( selected_ai = integration elif not _prompts_allowed(non_interactive): default_integration = resolve_default_init_integration() - console.print( - f"[dim]Non-interactive session detected: defaulting to '{default_integration}'. " - "Use --integration to choose a different agent.[/dim]" - ) + if not (dry_run and json_output): + console.print( + f"[dim]Non-interactive session detected: defaulting to '{default_integration}'. " + "Use --integration to choose a different agent.[/dim]" + ) selected_ai = default_integration else: ai_choices = {key: config["name"] for key, config in AGENT_CONFIG.items()} @@ -550,6 +1889,13 @@ def init( raise typer.Exit(1) if selected_ai == "generic" and not integration_options: + if dry_run and json_output: + _raise_dry_run_json_error( + "--integration generic requires --integration-options " + "with --commands-dir", + project_name=project_name, + here=here, + ) console.print( "[red]Error:[/red] --integration generic requires --integration-options with --commands-dir" ) @@ -572,15 +1918,23 @@ def init( f"{'Target Path':<15} [dim]{_escape_markup(str(project_path))}[/dim]" ) - console.print( - Panel("\n".join(setup_lines), border_style="cyan", padding=(1, 2)) - ) + if not (dry_run and json_output): + console.print( + Panel("\n".join(setup_lines), border_style="cyan", padding=(1, 2)) + ) if not ignore_agent_tools: agent_config = AGENT_CONFIG.get(selected_ai) if agent_config and agent_config["requires_cli"]: install_url = agent_config["install_url"] if not check_tool(selected_ai): + if dry_run and json_output: + _raise_dry_run_json_error( + f"{selected_ai} not found; {agent_config['name']} is " + "required to continue with this project type", + project_name=project_name, + here=here, + ) error_panel = Panel( f"[cyan]{selected_ai}[/cyan] not found\n" f"Install from: [cyan]{install_url}[/cyan]\n" @@ -596,6 +1950,12 @@ def init( if script_type: if script_type not in SCRIPT_TYPE_CHOICES: + if dry_run and json_output: + _raise_dry_run_json_error( + f"Invalid script type '{script_type}'", + project_name=project_name, + here=here, + ) console.print( f"[red]Error:[/red] Invalid script type '{_escape_markup(str(script_type))}'. Choose from: {', '.join(SCRIPT_TYPE_CHOICES.keys())}" ) @@ -614,8 +1974,33 @@ def init( else: selected_script = default_script - console.print(f"[cyan]Selected coding agent integration:[/cyan] {selected_ai}") - console.print(f"[cyan]Selected script type:[/cyan] {selected_script}") + if not (dry_run and json_output): + console.print(f"[cyan]Selected coding agent integration:[/cyan] {selected_ai}") + console.print(f"[cyan]Selected script type:[/cyan] {selected_script}") + + if dry_run: + gate = ( + "confirmation_required" + if here and directory_conflict + else "force_required" + if directory_conflict + else "none" + ) + _preview_init( + project_path=project_path, + gate=gate, + force=force, + here=here, + script_type=selected_script, + selected_integration=selected_ai, + ignore_agent_tools=ignore_agent_tools, + preset=preset, + integration_options=integration_options, + extensions=extensions, + trust_extension_urls=trust_extension_urls, + json_output=json_output, + ) + return tracker = StepTracker("Initialize Specify Project") @@ -775,6 +2160,12 @@ def init( wf_registry = WorkflowRegistry(project_path) if wf_registry.is_installed("speckit"): tracker.complete("workflow", "already installed") + _record_init_plan_action( + "skip", + ".specify/workflows/speckit/workflow.yml", + "workflow", + "speckit", + ) else: import shutil as _shutil @@ -803,6 +2194,7 @@ def init( tracker.skip("workflow", "bundled workflow not found") except Exception as wf_err: sanitized_wf = str(wf_err).replace("\n", " ").strip() + _record_init_plan_failure("workflow", "speckit", sanitized_wf) tracker.error("workflow", f"install failed: {sanitized_wf[:120]}") init_opts = { @@ -819,7 +2211,10 @@ def init( init_opts["ai_skills"] = True save_init_options(project_path, init_opts) - ensure_executable_scripts(project_path, tracker=tracker) + for chmod_failure in ensure_executable_scripts( + project_path, tracker=tracker + ): + _record_init_plan_failure("chmod", chmod_failure, chmod_failure) if preset: try: @@ -828,7 +2223,7 @@ def init( preset_manager = PresetManager(project_path) speckit_ver = get_speckit_version() - local_path = Path(preset).resolve() + local_path = Path(preset).expanduser().resolve() if local_path.is_dir() and (local_path / "preset.yml").exists(): preset_manager.install_from_directory( local_path, speckit_ver @@ -843,6 +2238,11 @@ def init( preset_catalog = PresetCatalog(project_path) pack_info = preset_catalog.get_pack_info(preset) if not pack_info: + _record_init_plan_failure( + "preset", + preset, + f"Preset '{preset}' not found in catalog", + ) console.print( f"[yellow]Warning:[/yellow] Preset '{preset}' not found in catalog. Skipping." ) @@ -851,6 +2251,11 @@ def init( ): from .extensions import REINSTALL_COMMAND + _record_init_plan_failure( + "preset", + preset, + "bundled preset not found in installed package", + ) console.print( f"[yellow]Warning:[/yellow] Preset '{preset}' is bundled with spec-kit " f"but could not be found in the installed package." @@ -871,6 +2276,9 @@ def init( catalog_name=pack_info.get("_catalog_name"), ) except PresetError as preset_err: + _record_init_plan_failure( + "preset", preset, str(preset_err) + ) _print_cli_warning( "install", "preset", @@ -885,6 +2293,7 @@ def init( except OSError: pass except Exception as preset_err: + _record_init_plan_failure("preset", preset, str(preset_err)) _print_cli_warning( "install", "preset", @@ -920,6 +2329,9 @@ def init( any_extension_installed = True except Exception as ext_err: sanitized_ext = str(ext_err).replace("\n", " ").strip() + _record_init_plan_failure( + "extension", ext_spec, sanitized_ext + ) tracker.error( f"extension-{i}", f"failed: {_escape_markup(sanitized_ext[:120])}", diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index e4b9e7de9d..6316b6a8d9 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -1415,6 +1415,11 @@ def _ensure_usable(skills_dir: Path) -> Optional[Path]: skills_dir = registrar._resolve_agent_dir( selected_ai, agent_config, self.project_root ) + from ..integrations import get_integration + + integration = get_integration(selected_ai) + if integration is not None: + integration.validate_output_path(skills_dir, self.project_root) return _ensure_usable(skills_dir) @staticmethod @@ -1662,6 +1667,8 @@ def _replacement(match: re.Match[str]) -> str: # Check if skill already exists before creating the directory skill_subdir = skills_dir / skill_name skill_file = skill_subdir / "SKILL.md" + if integration is not None: + integration.validate_output_path(skill_file, self.project_root) cache_root = extension_dir / ".specify-dev" / "extension-skills" cache_file = cache_root / skill_name / "SKILL.md" use_dev_symlink = link_outputs and not agent_config.get("dev_no_symlink") @@ -2082,6 +2089,26 @@ def check_compatibility( return True + def _validate_active_integration_outputs( + self, manifest: ExtensionManifest + ) -> None: + """Validate active-agent command destinations before installation writes.""" + from ..agents import CommandRegistrar + + registrar = CommandRegistrar() + command_names: list[str] = [] + for command in manifest.commands: + command_name = command.get("name") + if isinstance(command_name, str): + command_names.append(command_name) + aliases = command.get("aliases", []) + if isinstance(aliases, list): + command_names.extend(alias for alias in aliases if isinstance(alias, str)) + for agent_name in self._command_registration_targets(): + registrar.validate_integration_output_paths( + agent_name, command_names, self.project_root + ) + def install_from_directory( self, source_dir: Path, @@ -2135,6 +2162,8 @@ def install_from_directory( # Reject manifests that would shadow core commands or installed extensions. self._validate_install_conflicts(manifest) + self._validate_active_integration_outputs(manifest) + # Refuse to install an extension from its own install destination — with # --force this would delete the source before copying it (issue #2990). dest_dir = self.extensions_dir / manifest.id @@ -3238,6 +3267,7 @@ def register_enabled_extensions_for_agent(self, agent_name: str, *, force: bool return from .. import load_init_options + from ..integrations.base import IntegrationOutputPathError registrar = CommandRegistrar() agent_config = registrar.AGENT_CONFIGS.get(agent_name) @@ -3514,6 +3544,8 @@ def register_enabled_extensions_for_agent(self, agent_name: str, *, force: bool if updates: self.registry.update(ext_id, updates) + except IntegrationOutputPathError: + raise except Exception as ext_err: # Best-effort per extension: warn and move on so a single bad # extension cannot silently drop the others. See #2950. diff --git a/src/specify_cli/extensions/command_remove.py b/src/specify_cli/extensions/command_remove.py index c210dae96b..fbd39e7fbd 100644 --- a/src/specify_cli/extensions/command_remove.py +++ b/src/specify_cli/extensions/command_remove.py @@ -20,6 +20,7 @@ def extension_remove( ): """Uninstall an extension.""" from . import ExtensionManager + from ..integrations.base import IntegrationOutputPathError project_root = _commands._require_specify_project() manager = ExtensionManager(project_root) @@ -68,7 +69,14 @@ def extension_remove( raise typer.Exit(0) # Remove extension - success = manager.remove(extension_id, keep_config=keep_config) + try: + success = manager.remove(extension_id, keep_config=keep_config) + except IntegrationOutputPathError as exc: + console.print( + f"[red]Error:[/red] Cannot safely remove extension: " + f"{_escape_markup(str(exc))}" + ) + raise typer.Exit(1) from None if success: console.print(f"\n[green]✓[/green] Extension '{_escape_markup(str(display_name))}' removed successfully") diff --git a/src/specify_cli/integrations/base.py b/src/specify_cli/integrations/base.py index e698b9e289..938d74514e 100644 --- a/src/specify_cli/integrations/base.py +++ b/src/specify_cli/integrations/base.py @@ -59,6 +59,10 @@ } +class IntegrationOutputPathError(ValueError): + """Raised when an integration output would cross an unsafe path boundary.""" + + def yaml_quote(value: str) -> str: """Emit *value* as a double-quoted YAML scalar on a single line. @@ -1034,6 +1038,9 @@ def supports_events(self) -> bool: """Return True if this integration supports agent-native events.""" return bool(getattr(self, "CANONICAL_TO_NATIVE", None) and getattr(self, "events_config_file", None)) + def validate_output_path(self, path: Path, project_root: Path) -> None: + """Validate an integration-owned output before shared writers touch it.""" + # Context-injection envelope for hook stdout, keyed by canonical event # (with "*" as the fallback). Not every agent injects a hook's plain-text # stdout as model context: Gemini/Tabnine/Qwen/Devin are JSON-only diff --git a/src/specify_cli/integrations/hermes/__init__.py b/src/specify_cli/integrations/hermes/__init__.py index 582a7bf66e..ce0111a022 100644 --- a/src/specify_cli/integrations/hermes/__init__.py +++ b/src/specify_cli/integrations/hermes/__init__.py @@ -19,10 +19,37 @@ import yaml -from ..base import IntegrationOption, SkillsIntegration, yaml_quote +from ..._utils import path_is_junction +from ..base import ( + IntegrationOption, + IntegrationOutputPathError, + SkillsIntegration, + yaml_quote, +) from ..manifest import IntegrationManifest +def _has_symlinked_component(path: Path, trusted_root: Path) -> bool: + """Return whether *path* escapes *trusted_root* or traverses a symlink.""" + try: + relative = path.relative_to(trusted_root) + trusted_root_resolved = trusted_root.resolve() + except ValueError: + return True + + current = trusted_root + for part in relative.parts: + current = current / part + if current.is_symlink() or path_is_junction(current): + return True + if current.exists(): + try: + current.resolve().relative_to(trusted_root_resolved) + except (OSError, ValueError): + return True + return False + + class HermesIntegration(SkillsIntegration): """Integration for Hermes Agent skills. @@ -59,6 +86,14 @@ def _hermes_home_skills_dir() -> Path: """Return ``~/.hermes/skills/`` — the global skills directory.""" return Path.home() / ".hermes" / "skills" + def validate_output_path(self, path: Path, project_root: Path) -> None: + """Reject global registrar outputs that traverse symlinks or junctions.""" + if _has_symlinked_component(path, Path.home()): + raise IntegrationOutputPathError( + f"Hermes destination {path} contains a symlinked path component; " + "refusing to write through it." + ) + # -- Options ----------------------------------------------------------- @classmethod @@ -103,6 +138,22 @@ def setup( f"project_root ({project_root_resolved})" ) + global_skills_dir = self._hermes_home_skills_dir() + local_marker_dir = project_root / ".hermes" / "skills" + skill_targets = [ + global_skills_dir + / f"speckit-{src_file.stem.replace('.', '-')}" + / "SKILL.md" + for src_file in templates + ] + if _has_symlinked_component(local_marker_dir, project_root): + raise IntegrationOutputPathError( + f"Hermes destination {local_marker_dir} contains a symlinked path " + "component; refusing to install into it." + ) + for skill_target in skill_targets: + self.validate_output_path(skill_target, project_root) + script_type = opts.get("script_type", "sh") arg_placeholder = ( self.registrar_config.get("args", "$ARGUMENTS") @@ -110,7 +161,6 @@ def setup( else "$ARGUMENTS" ) - global_skills_dir = self._hermes_home_skills_dir() global_skills_dir.mkdir(parents=True, exist_ok=True) created: list[Path] = [] @@ -214,7 +264,7 @@ def setup( # Create project-local marker directory so extension commands # (e.g. git) can detect Hermes as an active integration. # Hermes itself ignores this directory — skills live globally. - (project_root / ".hermes" / "skills").mkdir(parents=True, exist_ok=True) + local_marker_dir.mkdir(parents=True, exist_ok=True) return created @@ -244,7 +294,9 @@ def teardown( # Remove project-local marker directory if empty local_skills_dir = project_root / ".hermes" / "skills" - if local_skills_dir.is_dir() and not any(local_skills_dir.iterdir()): + if _has_symlinked_component(local_skills_dir, project_root): + skipped.append(local_skills_dir) + elif local_skills_dir.is_dir() and not any(local_skills_dir.iterdir()): local_skills_dir.rmdir() hermes_dir = project_root / ".hermes" if hermes_dir.is_dir() and not any(hermes_dir.iterdir()): @@ -254,9 +306,16 @@ def teardown( # removed on uninstall regardless of the force flag, matching the # standard behaviour where all integration files are cleaned up. global_skills_dir = self._hermes_home_skills_dir() - if global_skills_dir.is_dir(): + if _has_symlinked_component(global_skills_dir, Path.home()): + skipped.append(global_skills_dir) + elif global_skills_dir.is_dir(): for skill_dir in sorted(global_skills_dir.iterdir()): - if skill_dir.is_dir() and skill_dir.name.startswith("speckit-"): + if not skill_dir.name.startswith("speckit-"): + continue + if _has_symlinked_component(skill_dir, Path.home()): + skipped.append(skill_dir) + continue + if skill_dir.is_dir(): try: rmtree(skill_dir) removed.append(skill_dir) diff --git a/src/specify_cli/presets/_manager.py b/src/specify_cli/presets/_manager.py index 33274dfdfc..c190baf18e 100644 --- a/src/specify_cli/presets/_manager.py +++ b/src/specify_cli/presets/_manager.py @@ -11,7 +11,11 @@ from packaging.specifiers import InvalidSpecifier, SpecifierSet from .._download_security import safe_extract_archive -from .._init_options import is_ai_skills_enabled, resolve_active_agent_for_registration +from .._init_options import ( + MISSING_INIT_OPTIONS_FILE, + is_ai_skills_enabled, + resolve_active_agent_for_registration, +) from .._utils import version_satisfies from ..extensions import REINSTALL_COMMAND, normalize_priority from ..shared_infra import ( @@ -356,6 +360,48 @@ def find_unmet_extension_dependencies( return unmet + def _validate_active_integration_outputs( + self, manifest: PresetManifest + ) -> None: + """Validate active-agent command destinations before preset writes.""" + from ..agents import CommandRegistrar + + command_names: list[str] = [] + for template in manifest.templates: + if template.get("type") != "command": + continue + command_name = template.get("name") + if isinstance(command_name, str): + command_names.append(command_name) + aliases = template.get("aliases", []) + if isinstance(aliases, list): + command_names.extend( + alias for alias in aliases if isinstance(alias, str) + ) + registrar = CommandRegistrar() + resolved_agent = resolve_active_agent_for_registration(self.project_root) + if resolved_agent is MISSING_INIT_OPTIONS_FILE: + agents: list[str] = [] + for agent_name, agent_config in registrar.AGENT_CONFIGS.items(): + detect_dir = agent_config.get("detect_dir") + if detect_dir: + if (self.project_root / detect_dir).is_dir(): + agents.append(agent_name) + continue + if registrar._resolve_agent_dir( + agent_name, agent_config, self.project_root + ).is_dir(): + agents.append(agent_name) + elif isinstance(resolved_agent, str): + agents = [resolved_agent] + else: + agents = [] + + for agent_name in agents: + registrar.validate_integration_output_paths( + agent_name, command_names, self.project_root + ) + def install_from_directory( self, source_dir: Path, @@ -388,6 +434,7 @@ def install_from_directory( manifest = PresetManifest(manifest_path) self.check_compatibility(manifest, speckit_version) + self._validate_active_integration_outputs(manifest) if self.registry.is_installed(manifest.id): if not force: @@ -745,6 +792,16 @@ def remove(self, pack_id: str) -> bool: # names from registered_commands are still unregistered. pass + if _CommandRegistrarForScope is not None: + artifact_agents = set(registered_commands) + if isinstance(registered_skills, dict): + artifact_agents.update(registered_skills) + registrar = _CommandRegistrarForScope() + for agent_name in artifact_agents: + registrar.validate_integration_output_paths( + agent_name, removed_cmd_names, self.project_root + ) + affected_skill_dirs: Dict[ Path, tuple[Optional[str], List[str]] ] = {} diff --git a/src/specify_cli/presets/command_remove.py b/src/specify_cli/presets/command_remove.py index 3654f1e04f..8c405975e4 100644 --- a/src/specify_cli/presets/command_remove.py +++ b/src/specify_cli/presets/command_remove.py @@ -3,6 +3,7 @@ from __future__ import annotations import typer +from rich.markup import escape as _escape_markup from .._console import console from ._commands import preset_app @@ -14,6 +15,7 @@ def preset_remove( ): """Remove an installed preset.""" from .. import _require_specify_project + from ..integrations.base import IntegrationOutputPathError from . import PresetManager project_root = _require_specify_project() @@ -23,7 +25,16 @@ def preset_remove( console.print(f"[red]Error:[/red] Preset '{preset_id}' is not installed") raise typer.Exit(1) - if manager.remove(preset_id): + try: + removed = manager.remove(preset_id) + except IntegrationOutputPathError as exc: + console.print( + f"[red]Error:[/red] Cannot safely remove preset: " + f"{_escape_markup(str(exc))}" + ) + raise typer.Exit(1) from None + + if removed: console.print(f"[green]✓[/green] Preset '{preset_id}' removed successfully") else: console.print(f"[red]Error:[/red] Failed to remove preset '{preset_id}'") diff --git a/tests/integrations/test_integration_hermes.py b/tests/integrations/test_integration_hermes.py index b2106050b7..1a3dbe69c5 100644 --- a/tests/integrations/test_integration_hermes.py +++ b/tests/integrations/test_integration_hermes.py @@ -10,7 +10,14 @@ non-destructive to a developer's real Hermes installation. """ +import json +import shutil +import stat from pathlib import Path +from types import SimpleNamespace + +import pytest +import yaml from specify_cli.integrations import get_integration from specify_cli.integrations.manifest import IntegrationManifest @@ -31,6 +38,31 @@ class TestHermesIntegration(SkillsIntegrationTests): COMMANDS_SUBDIR = "skills" REGISTRAR_DIR = "~/.hermes/skills" + def test_path_guard_uses_python311_junction_fallback( + self, tmp_path, monkeypatch + ): + from specify_cli.integrations.hermes import _has_symlinked_component + import specify_cli._utils as utils + + home = _fake_home(tmp_path) + junction = home / ".hermes" / "skills" / "speckit-plan" + junction.mkdir(parents=True) + target = junction / "SKILL.md" + mount_point_tag = getattr(stat, "IO_REPARSE_TAG_MOUNT_POINT", 0xA0000003) + + monkeypatch.setattr(Path, "is_junction", None, raising=False) + monkeypatch.setattr(utils.os, "name", "nt") + monkeypatch.setattr( + Path, + "lstat", + lambda path: SimpleNamespace( + st_mode=stat.S_IFDIR | 0o755, + st_reparse_tag=mount_point_tag if path == junction else 0 + ), + ) + + assert _has_symlinked_component(target, home) + # -- Hermes-specific setup: skills go to ~/.hermes/skills/ ------------- def test_setup_writes_to_global_skills_dir(self, tmp_path, monkeypatch): @@ -65,6 +97,48 @@ def test_local_marker_dir_created(self, tmp_path, monkeypatch): children = list(marker.iterdir()) assert children == [], f"Marker directory should be empty, got: {children}" + def test_setup_rejects_symlinked_global_skill_file( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + external = tmp_path / "external-skill.md" + external.write_text("external\n", encoding="utf-8") + skill_file = home / ".hermes" / "skills" / "speckit-plan" / "SKILL.md" + skill_file.parent.mkdir(parents=True) + try: + skill_file.symlink_to(external) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + i = get_integration(self.KEY) + m = IntegrationManifest(self.KEY, tmp_path) + with pytest.raises(ValueError, match="symlinked path component"): + i.setup(tmp_path, m) + + assert external.read_text(encoding="utf-8") == "external\n" + assert not (tmp_path / ".hermes").exists() + + def test_setup_rejects_symlinked_project_marker_before_global_writes( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + external = tmp_path / "external-marker" + external.mkdir() + try: + (tmp_path / ".hermes").symlink_to(external, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + i = get_integration(self.KEY) + m = IntegrationManifest(self.KEY, tmp_path) + with pytest.raises(ValueError, match="symlinked path component"): + i.setup(tmp_path, m) + + assert list(external.iterdir()) == [] + assert not (home / ".hermes").exists() + # -- Override shared tests that assume project-local skills ------------ def test_setup_writes_to_correct_directory(self, tmp_path, monkeypatch): @@ -192,6 +266,486 @@ def test_pre_existing_skills_not_removed(self, tmp_path, monkeypatch): "Foreign skill was removed by teardown" ) + def test_teardown_does_not_follow_symlinked_global_skills_directory( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + external = tmp_path / "external-hermes-skills" + protected = external / "speckit-userdata" / "keep.txt" + protected.parent.mkdir(parents=True) + protected.write_text("keep\n", encoding="utf-8") + hermes_dir = home / ".hermes" + hermes_dir.mkdir() + try: + (hermes_dir / "skills").symlink_to( + external, target_is_directory=True + ) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + i = get_integration(self.KEY) + m = IntegrationManifest(self.KEY, tmp_path) + removed, skipped = i.teardown(tmp_path, m) + + assert removed == [] + assert home / ".hermes" / "skills" in skipped + assert protected.read_text(encoding="utf-8") == "keep\n" + + def test_teardown_does_not_follow_symlinked_project_marker( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + external = tmp_path / "external-marker" + external.mkdir() + marker_parent = tmp_path / ".hermes" + marker_parent.mkdir() + try: + (marker_parent / "skills").symlink_to( + external, target_is_directory=True + ) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + i = get_integration(self.KEY) + m = IntegrationManifest(self.KEY, tmp_path) + removed, skipped = i.teardown(tmp_path, m) + + assert removed == [] + assert tmp_path / ".hermes" / "skills" in skipped + assert external.is_dir() + + def test_extension_registration_rejects_symlinked_global_skill( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + external = tmp_path / "external-extension-skill" + external.mkdir() + external_skill = external / "SKILL.md" + external_skill.write_text("external\n", encoding="utf-8") + global_skills = home / ".hermes" / "skills" + global_skills.mkdir(parents=True) + try: + (global_skills / "speckit-git-feature").symlink_to( + external, target_is_directory=True + ) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + from typer.testing import CliRunner + from specify_cli import app + + target = tmp_path / "hermes-extension-link" + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + "--extension", + "git", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert external_skill.read_text(encoding="utf-8") == "external\n" + assert not (target / ".specify" / "extensions" / "git").exists() + + preview = CliRunner().invoke( + app, + [ + "init", + str(tmp_path / "hermes-extension-link-preview"), + "--dry-run", + "--json", + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + "--extension", + "git", + ], + catch_exceptions=False, + ) + assert preview.exit_code == 0, preview.output + failures = json.loads(preview.output)["failures"] + assert any("symlinked path component" in failure["error"] for failure in failures) + assert external_skill.read_text(encoding="utf-8") == "external\n" + + def test_extension_alias_is_preflighted_before_any_install_write( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + external = tmp_path / "external-alias-skill" + external.mkdir() + external_skill = external / "SKILL.md" + external_skill.write_text("external\n", encoding="utf-8") + global_skills = home / ".hermes" / "skills" + global_skills.mkdir(parents=True) + try: + (global_skills / "speckit-my-extension-example-short").symlink_to( + external, target_is_directory=True + ) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + from typer.testing import CliRunner + from specify_cli import app + + extension = Path(__file__).parents[2] / "extensions" / "template" + target = tmp_path / "hermes-extension-alias-link" + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + "--extension", + str(extension), + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert external_skill.read_text(encoding="utf-8") == "external\n" + assert not ( + global_skills / "speckit-my-extension-example" / "SKILL.md" + ).exists() + assert not ( + target / ".specify" / "extensions" / "my-extension" + ).exists() + + def test_legacy_extension_preflight_uses_detected_hermes_target( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + external = tmp_path / "external-legacy-extension" + external.mkdir() + external_skill = external / "SKILL.md" + external_skill.write_text("external\n", encoding="utf-8") + global_skills = home / ".hermes" / "skills" + global_skills.mkdir(parents=True) + try: + (global_skills / "speckit-my-extension-example").symlink_to( + external, target_is_directory=True + ) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + (tmp_path / ".hermes" / "skills").mkdir(parents=True) + + from specify_cli.extensions import ExtensionManager + from specify_cli.integrations.base import IntegrationOutputPathError + + extension = Path(__file__).parents[2] / "extensions" / "template" + manager = ExtensionManager(tmp_path) + with pytest.raises(IntegrationOutputPathError): + manager.install_from_directory(extension, "0.3.0") + + assert external_skill.read_text(encoding="utf-8") == "external\n" + assert not ( + tmp_path / ".specify" / "extensions" / "my-extension" + ).exists() + + def test_extension_remove_retains_registry_when_hermes_cleanup_is_unsafe( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.extensions import ExtensionManager + from specify_cli.integrations.base import IntegrationOutputPathError + + target = tmp_path / "hermes-extension-remove" + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + "--extension", + "git", + ], + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + + global_skills = home / ".hermes" / "skills" + safe_skill = global_skills / "speckit-git-commit" / "SKILL.md" + unsafe_dir = global_skills / "speckit-git-feature" + assert safe_skill.is_file() + shutil.rmtree(unsafe_dir) + external = tmp_path / "external-remove-target" + external.mkdir() + external_skill = external / "SKILL.md" + external_skill.write_text("external\n", encoding="utf-8") + try: + unsafe_dir.symlink_to(external, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + manager = ExtensionManager(target) + with pytest.raises(IntegrationOutputPathError): + manager.remove("git") + + assert manager.registry.is_installed("git") + assert safe_skill.is_file() + assert unsafe_dir.is_symlink() + assert external_skill.read_text(encoding="utf-8") == "external\n" + + monkeypatch.chdir(target) + cli_result = CliRunner().invoke( + app, + ["extension", "remove", "git", "--force"], + catch_exceptions=False, + ) + assert cli_result.exit_code == 1, cli_result.output + assert "Cannot safely remove extension" in cli_result.output + assert manager.registry.is_installed("git") + + def test_extension_remove_preflights_legacy_hermes_names( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.extensions import ExtensionManager + from specify_cli.integrations.base import IntegrationOutputPathError + + target = tmp_path / "hermes-legacy-remove" + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + "--extension", + "git", + ], + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + + global_skills = home / ".hermes" / "skills" + modern_skill = global_skills / "speckit-git-feature" / "SKILL.md" + assert modern_skill.is_file() + external = tmp_path / "external-legacy-remove" + external.mkdir() + external_skill = external / "SKILL.md" + external_skill.write_text("external\n", encoding="utf-8") + legacy_dir = global_skills / "speckit.git.feature" + try: + legacy_dir.symlink_to(external, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + manager = ExtensionManager(target) + with pytest.raises(IntegrationOutputPathError): + manager.remove("git") + + assert manager.registry.is_installed("git") + assert modern_skill.is_file() + assert legacy_dir.is_symlink() + assert external_skill.read_text(encoding="utf-8") == "external\n" + + def test_preset_remove_retains_registry_when_hermes_cleanup_is_unsafe( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + preset = tmp_path / "remove-preflight" + command = preset / "commands" / "remove.md" + command.parent.mkdir(parents=True) + command.write_text( + "---\ndescription: Remove preflight\n---\n\nBody\n", + encoding="utf-8", + ) + (preset / "preset.yml").write_text( + yaml.safe_dump( + { + "schema_version": "1.0", + "preset": { + "id": "remove-preflight", + "name": "Remove Preflight", + "version": "1.0.0", + "description": "Removal safety test", + }, + "requires": {"speckit_version": ">=0.1.0"}, + "provides": { + "templates": [ + { + "type": "command", + "name": "speckit.remove-preflight", + "file": "commands/remove.md", + } + ] + }, + } + ), + encoding="utf-8", + ) + + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.integrations.base import IntegrationOutputPathError + from specify_cli.presets import PresetManager + + target = tmp_path / "hermes-preset-remove" + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + "--preset", + str(preset), + ], + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + + global_skills = home / ".hermes" / "skills" + unsafe_dir = global_skills / "speckit-remove-preflight" + shutil.rmtree(unsafe_dir) + external = tmp_path / "external-preset-remove" + external.mkdir() + external_skill = external / "SKILL.md" + external_skill.write_text("external\n", encoding="utf-8") + try: + unsafe_dir.symlink_to(external, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + manager = PresetManager(target) + with pytest.raises(IntegrationOutputPathError): + manager.remove("remove-preflight") + + assert manager.registry.is_installed("remove-preflight") + assert unsafe_dir.is_symlink() + assert external_skill.read_text(encoding="utf-8") == "external\n" + + def test_preset_alias_is_preflighted_before_any_install_write( + self, tmp_path, monkeypatch + ): + home = _fake_home(tmp_path) + monkeypatch.setattr(Path, "home", lambda: home) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + external = tmp_path / "external-preset-alias" + external.mkdir() + external_skill = external / "SKILL.md" + external_skill.write_text("external\n", encoding="utf-8") + global_skills = home / ".hermes" / "skills" + global_skills.mkdir(parents=True) + try: + (global_skills / "speckit-my-preset-alias").symlink_to( + external, target_is_directory=True + ) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + preset = tmp_path / "alias-preflight" + command = preset / "commands" / "primary.md" + command.parent.mkdir(parents=True) + command.write_text( + "---\ndescription: Alias preflight\n---\n\nBody\n", + encoding="utf-8", + ) + (preset / "preset.yml").write_text( + yaml.safe_dump( + { + "schema_version": "1.0", + "preset": { + "id": "alias-preflight", + "name": "Alias Preflight", + "version": "1.0.0", + "description": "Alias safety test", + }, + "requires": {"speckit_version": ">=0.1.0"}, + "provides": { + "templates": [ + { + "type": "command", + "name": "speckit.my-preset-primary", + "file": "commands/primary.md", + "aliases": ["speckit.my-preset-alias"], + } + ] + }, + } + ), + encoding="utf-8", + ) + + from typer.testing import CliRunner + from specify_cli import app + + target = tmp_path / "hermes-preset-alias-link" + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + "--preset", + str(preset), + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "Hermes destination" in " ".join(result.output.split()) + assert external_skill.read_text(encoding="utf-8") == "external\n" + assert not ( + global_skills / "speckit-my-preset-primary" / "SKILL.md" + ).exists() + assert not ( + target / ".specify" / "presets" / "alias-preflight" + ).exists() + def test_hook_sections_explain_dotted_command_conversion(self, tmp_path, monkeypatch): """Override: Hermes skills live in global ~/.hermes/skills/.""" home = _fake_home(tmp_path) diff --git a/tests/test_extensions.py b/tests/test_extensions.py index b512742389..4a5894533a 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -42,6 +42,7 @@ CompatibilityError, normalize_priority, ) +from specify_cli.integrations.base import IntegrationOutputPathError from specify_cli._utils import version_satisfies # Minimal valid ZIP (empty end-of-central-directory record). Passes @@ -5060,6 +5061,77 @@ def test_copilot_cleanup_removes_prompt_files(self, extension_dir, project_dir): assert not agent_file.exists() assert not prompt_file.exists() + def test_copilot_cleanup_removes_nested_alias( + self, extension_dir, project_dir, valid_manifest_data + ): + valid_manifest_data["provides"]["commands"][0]["aliases"] = ["group/run"] + (extension_dir / "extension.yml").write_text( + yaml.safe_dump(valid_manifest_data), encoding="utf-8" + ) + agents_dir = project_dir / ".github" / "agents" + agents_dir.mkdir(parents=True) + + manager = ExtensionManager(project_dir) + manager.install_from_directory(extension_dir, "0.1.0") + agent_file = agents_dir / "group" / "run.agent.md" + prompt_file = project_dir / ".github" / "prompts" / "group" / "run.prompt.md" + assert agent_file.is_file() + assert prompt_file.is_file() + + assert manager.remove("test-ext") is True + assert not manager.registry.is_installed("test-ext") + assert not agent_file.exists() + assert not prompt_file.exists() + + def test_copilot_cleanup_rejects_symlinked_prompt_parent( + self, extension_dir, project_dir + ): + agents_dir = project_dir / ".github" / "agents" + agents_dir.mkdir(parents=True) + manager = ExtensionManager(project_dir) + manager.install_from_directory(extension_dir, "0.1.0") + agent_file = agents_dir / "speckit.test-ext.hello.agent.md" + prompt_name = "speckit.test-ext.hello.prompt.md" + prompts_dir = project_dir / ".github" / "prompts" + assert agent_file.is_file() + shutil.rmtree(prompts_dir) + external = project_dir.parent / "external-prompts" + external.mkdir() + external_prompt = external / prompt_name + external_prompt.write_text("external\n", encoding="utf-8") + try: + prompts_dir.symlink_to(external, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + with pytest.raises(IntegrationOutputPathError): + manager.remove("test-ext") + + assert manager.registry.is_installed("test-ext") + assert agent_file.is_file() + assert external_prompt.read_text(encoding="utf-8") == "external\n" + + def test_copilot_cleanup_rejects_traversal_in_registered_command( + self, extension_dir, project_dir + ): + agents_dir = project_dir / ".github" / "agents" + agents_dir.mkdir(parents=True) + manager = ExtensionManager(project_dir) + manager.install_from_directory(extension_dir, "0.1.0") + agent_file = agents_dir / "speckit.test-ext.hello.agent.md" + metadata = manager.registry.get("test-ext") + metadata["registered_commands"]["copilot"].append("../../outside") + manager.registry.update("test-ext", metadata) + outside = project_dir / "outside.prompt.md" + outside.write_text("outside\n", encoding="utf-8") + + with pytest.raises(IntegrationOutputPathError): + manager.remove("test-ext") + + assert manager.registry.is_installed("test-ext") + assert agent_file.is_file() + assert outside.read_text(encoding="utf-8") == "outside\n" + def test_multiple_extensions(self, temp_dir, project_dir): """Test installing multiple extensions.""" import yaml diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py new file mode 100644 index 0000000000..5549fb55b0 --- /dev/null +++ b/tests/test_init_dry_run.py @@ -0,0 +1,1884 @@ +"""CLI contract tests for ``specify init --dry-run``.""" + +from __future__ import annotations + +import json +import os +import shutil +import stat +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import pytest +from typer.testing import CliRunner + +from specify_cli import app +from specify_cli._assets import _locate_bundled_extension, _locate_bundled_preset +from specify_cli._utils import ( + windows_path_is_junction as _windows_path_is_junction, +) +from specify_cli.command_init import ( + _is_within_root, + _normalize_fs_path, + _preview_child_failure_message, + _preview_content_ownership, + _preview_home_seed_paths, + _preview_seed_paths, + _seed_preview_home, + _snapshot_files, + _snapshot_tree_entries, + _stage_project_copy, + _strip_windows_extended_prefix, +) + +_PROVENANCE_CATEGORIES = {"core", "integration", "preset", "workflow", "extension"} + + +def _action_for(payload: dict, path: str) -> dict: + return next(action for action in payload["actions"] if action["path"] == path) + + +def _assert_same_path(left: Path | str, right: Path | str) -> None: + left_path = Path(left) + right_path = Path(right) + if left_path.exists() and right_path.exists(): + assert os.path.samefile(left_path, right_path) + return + assert _normalize_fs_path(left_path) == _normalize_fs_path(right_path) + + +def test_seed_preview_home_copies_auth_config(tmp_path: Path) -> None: + real_home = tmp_path / "real-home" + auth_config = real_home / ".specify" / "auth.json" + auth_config.parent.mkdir(parents=True) + auth_config.write_text('{"providers": []}\n', encoding="utf-8") + auth_config.chmod(0o600) + staged_home = tmp_path / "staged-home" + staged_home.mkdir() + + _seed_preview_home(staged_home, real_home) + + staged_auth = staged_home / ".specify" / "auth.json" + assert staged_auth.read_text(encoding="utf-8") == '{"providers": []}\n' + assert staged_auth.stat().st_mode & 0o777 == auth_config.stat().st_mode & 0o777 + + +def test_seed_preview_home_copies_symlinked_read_only_config(tmp_path: Path) -> None: + real_home = tmp_path / "real-home" + external_auth = tmp_path / "external-auth.json" + external_auth.write_text('{"providers": []}\n', encoding="utf-8") + auth_config = real_home / ".specify" / "auth.json" + auth_config.parent.mkdir(parents=True) + try: + auth_config.symlink_to(external_auth) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + staged_home = tmp_path / "staged-home" + staged_home.mkdir() + + _seed_preview_home(staged_home, real_home) + + staged_auth = staged_home / ".specify" / "auth.json" + assert staged_auth.read_text(encoding="utf-8") == '{"providers": []}\n' + assert not staged_auth.is_symlink() + + +def test_preview_home_seed_paths_propagates_unreadable_destination( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + real_home = tmp_path / "real-home" + skills_dir = real_home / ".hermes" / "skills" + skills_dir.mkdir(parents=True) + original_iterdir = Path.iterdir + + def unreadable_iterdir(path: Path): + if path == skills_dir: + raise PermissionError(13, "Permission denied", str(path)) + return original_iterdir(path) + + monkeypatch.setattr(Path, "iterdir", unreadable_iterdir) + + with pytest.raises(PermissionError, match="Permission denied"): + _preview_home_seed_paths(real_home, "hermes") + + +def test_dry_run_reports_unreadable_home_seed_as_json_error( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + real_home = tmp_path / "real-home" + skills_dir = real_home / ".hermes" / "skills" + skills_dir.mkdir(parents=True) + monkeypatch.setenv("HOME", str(real_home)) + monkeypatch.setenv("USERPROFILE", str(real_home)) + original_iterdir = Path.iterdir + + def unreadable_iterdir(path: Path): + if path == skills_dir: + raise PermissionError(13, "Permission denied", str(path)) + return original_iterdir(path) + + monkeypatch.setattr(Path, "iterdir", unreadable_iterdir) + + result = CliRunner().invoke( + app, + [ + "init", + str(tmp_path / "unreadable-home-preview"), + "--dry-run", + "--json", + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + payload = json.loads(result.output) + assert "failed to stage preview inputs" in payload["error"] + assert "Permission denied" in payload["error"] + + +def test_preview_seed_paths_include_cross_integration_event_targets() -> None: + paths = _preview_seed_paths("copilot", None, "sh") + + assert Path("opencode.json") in paths + assert Path(".opencode/plugin/speckit-events.ts") in paths + + +def test_preview_seed_paths_include_both_bob_layouts() -> None: + paths = _preview_seed_paths("bob", None, "sh") + + assert Path(".bob/commands") in paths + assert Path(".bob/skills") in paths + + +def test_dry_run_preserves_bob_skills_mode_when_both_layouts_exist( + tmp_path: Path, +) -> None: + target = tmp_path / "bob-layout-preview" + legacy_command = target / ".bob" / "commands" / "speckit.plan.md" + legacy_command.parent.mkdir(parents=True) + legacy_command.write_text("# stale command\n", encoding="utf-8") + managed_skill = target / ".bob" / "skills" / "speckit-plan" / "SKILL.md" + managed_skill.parent.mkdir(parents=True) + managed_skill.write_text("# managed skill\n", encoding="utf-8") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "bob", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + action_paths = {action["path"] for action in json.loads(result.output)["actions"]} + assert ".bob/skills/speckit-specify/SKILL.md" in action_paths + assert ".bob/commands/speckit.specify.md" not in action_paths + + +def test_public_staging_environment_cannot_bypass_directory_guard( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "existing-project" + target.mkdir() + sentinel = target / "keep.txt" + sentinel.write_text("keep\n", encoding="utf-8") + monkeypatch.setenv("SPECIFY_INIT_PLAN_PATH", str(tmp_path / "forged-plan.jsonl")) + monkeypatch.setenv("SPECIFY_INIT_STAGING_CONFIRMATION", "1") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--non-interactive", + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + assert "Directory already exists" in result.output + assert sentinel.read_text(encoding="utf-8") == "keep\n" + assert not (target / ".specify").exists() + + +def test_dry_run_child_ignores_pythonpath_sitecustomize( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + caller = tmp_path / "untrusted-caller" + caller.mkdir() + marker = tmp_path / "sitecustomize-executed" + (caller / "sitecustomize.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed\\n', encoding='utf-8')\n", + encoding="utf-8", + ) + monkeypatch.chdir(caller) + monkeypatch.setenv("PYTHONPATH", str(caller)) + + result = CliRunner().invoke( + app, + [ + "init", + str(tmp_path / "isolated-preview"), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + json.loads(result.output) + assert not marker.exists() + + +def test_dry_run_child_does_not_import_shadow_package_from_caller_directory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + caller = tmp_path / "untrusted-caller" + shadow_package = caller / "specify_cli" + shadow_package.mkdir(parents=True) + marker = tmp_path / "shadow-package-executed" + (shadow_package / "__init__.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed\\n', encoding='utf-8')\n", + encoding="utf-8", + ) + monkeypatch.chdir(caller) + + result = CliRunner().invoke( + app, + [ + "init", + str(tmp_path / "isolated-shadow-preview"), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + json.loads(result.output) + assert not marker.exists() + + +@pytest.mark.parametrize( + "content", + [ + "source: extension:anything\n", + "\n", + ], +) +def test_preview_content_ownership_rejects_unregistered_typed_markers( + content: str, +) -> None: + assert _preview_content_ownership(content, {"git": {"extension"}}) is None + + +def test_preview_content_ownership_rejects_typed_marker_with_wrong_category() -> None: + assert ( + _preview_content_ownership( + "source: extension:self-test\n", {"self-test": {"preset"}} + ) + is None + ) + + +def test_preview_child_failure_message_ignores_rich_panel_line_wrapping() -> None: + result = subprocess.CompletedProcess( + args=[], + returncode=1, + stdout=( + "│ Initialization failed: Integration destination │\n" + "│ /tmp/quarantine/.kilo/commands escapes project │\n" + "│ root /tmp/project │\n" + ), + stderr="", + ) + + assert "escapes project root" in _preview_child_failure_message(result) + + +def test_snapshot_tree_entries_detects_empty_directory_creation( + tmp_path: Path, +) -> None: + before = _snapshot_tree_entries(tmp_path) + (tmp_path / "created-empty-directory").mkdir() + + assert _snapshot_tree_entries(tmp_path) != before + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + (r"\\?\C:\Users\runner\proj", r"C:\Users\runner\proj"), + ("//?/C:/Users/runner/proj", "C:/Users/runner/proj"), + (r"\\?\UNC\server\share\dir", r"\\server\share\dir"), + ("//?/UNC/server/share/dir", "//server/share/dir"), + ("/tmp/proj", "/tmp/proj"), + (r"C:\Users\runner\proj", r"C:\Users\runner\proj"), + ], +) +def test_strip_windows_extended_prefix(raw: str, expected: str) -> None: + assert _strip_windows_extended_prefix(raw) == expected + + +def test_is_within_root_for_nested_real_paths(tmp_path: Path) -> None: + nested = tmp_path / "store" + nested.mkdir() + assert _is_within_root(nested, tmp_path) + assert _is_within_root(tmp_path, tmp_path) + assert not _is_within_root(tmp_path.parent, tmp_path) + + +def test_is_within_root_does_not_match_prefix_sibling(tmp_path: Path) -> None: + project = tmp_path / "proj" + sibling = tmp_path / "proj-evil" + project.mkdir() + sibling.mkdir() + assert not _is_within_root(sibling, project) + + +def test_dry_run_reports_new_project_files_without_creating_target(tmp_path: Path) -> None: + target = tmp_path / "preview-project" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "Initialization preview" in result.output + assert ".github/skills/speckit-plan/SKILL.md" in result.output + assert not target.exists() + + +def test_dry_run_json_is_machine_readable_and_has_no_target_writes(tmp_path: Path) -> None: + target = tmp_path / "json-preview-project" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["dry_run"] is True + assert {action["path"] for action in payload["actions"]} >= { + ".github/skills/speckit-plan/SKILL.md" + } + plan_action = _action_for(payload, ".github/skills/speckit-plan/SKILL.md") + assert plan_action["provenance"] == "integration" + assert plan_action["source_id"] == "copilot" + manifest_action = _action_for( + payload, ".specify/integrations/copilot.manifest.json" + ) + assert manifest_action["provenance"] == "integration" + assert manifest_action["source_id"] == "copilot" + assert { + action["provenance"] for action in payload["actions"] + } <= _PROVENANCE_CATEGORIES + assert not target.exists() + + +def test_dry_run_json_is_pure_json_without_explicit_integration(tmp_path: Path) -> None: + target = tmp_path / "default-integration-json-preview" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--non-interactive", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["dry_run"] is True + assert payload["actions"] + assert "Non-interactive session detected" not in result.output + assert not target.exists() + + +def test_forced_dry_run_reports_overwrite_without_changing_existing_file( + tmp_path: Path, +) -> None: + target = tmp_path / "existing-project" + shared_template = target / ".specify" / "templates" / "plan-template.md" + shared_template.parent.mkdir(parents=True) + shared_template.write_text("user-owned content\n", encoding="utf-8") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert {(action["action"], action["path"]) for action in payload["actions"]} >= { + ("overwrite", ".specify/templates/plan-template.md") + } + assert shared_template.read_text(encoding="utf-8") == "user-owned content\n" + + +def test_non_forced_here_dry_run_matches_confirmed_preserve_behavior( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "nonempty-project" + shared_template = target / ".specify" / "templates" / "plan-template.md" + shared_template.parent.mkdir(parents=True) + shared_template.write_text("user-owned content\n", encoding="utf-8") + existing = target / "keep.txt" + existing.write_text("keep\n", encoding="utf-8") + monkeypatch.chdir(target) + + result = CliRunner().invoke( + app, + [ + "init", + "--here", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["conflict"] is True + actions = {(action["action"], action["path"]) for action in payload["actions"]} + assert ("preserve", ".specify/templates/plan-template.md") in actions + assert ("create", ".github/skills/speckit-specify/SKILL.md") in actions + assert all(action["action"] != "overwrite" for action in payload["actions"]) + assert all(action["path"] != "keep.txt" for action in payload["actions"]) + assert shared_template.read_text(encoding="utf-8") == "user-owned content\n" + assert existing.read_text(encoding="utf-8") == "keep\n" + + actual = CliRunner().invoke( + app, + [ + "init", + "--here", + "--integration", + "copilot", + "--script", + "sh", + ], + input="y\n", + catch_exceptions=False, + ) + + assert actual.exit_code == 0, actual.output + assert shared_template.read_text(encoding="utf-8") == "user-owned content\n" + + +def test_non_forced_dry_run_reports_directory_conflict_without_overlapping_files( + tmp_path: Path, +) -> None: + target = tmp_path / "unrelated-nonempty-project" + target.mkdir() + existing = target / "keep.txt" + existing.write_text("keep\n", encoding="utf-8") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["conflict"] is True + assert payload["actions"] + assert all(action["action"] == "create" for action in payload["actions"]) + assert {action["path"] for action in payload["actions"]} >= { + ".github/skills/speckit-plan/SKILL.md" + } + assert all(action["path"] != "keep.txt" for action in payload["actions"]) + assert existing.read_text(encoding="utf-8") == "keep\n" + + +def test_non_forced_dry_run_human_preview_lists_conflicting_artifacts( + tmp_path: Path, +) -> None: + target = tmp_path / "nonempty-human-preview" + command = target / ".github" / "skills" / "speckit-plan" / "SKILL.md" + command.parent.mkdir(parents=True) + command.write_text("user-owned content\n", encoding="utf-8") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + lines = result.output.splitlines() + assert any(line.startswith("conflict") and line.endswith("target directory exists; applying this plan requires --force") for line in lines) + assert any( + line.startswith("conflict .github/skills/speckit-plan/SKILL.md") + for line in lines + ) + assert any( + line.startswith("create .github/skills/speckit-specify/SKILL.md") + for line in lines + ) + assert command.read_text(encoding="utf-8") == "user-owned content\n" + + +def test_dry_run_reports_skip_for_already_installed_bundled_workflow( + tmp_path: Path, +) -> None: + target = tmp_path / "reinit-workflow-preview" + arguments = [ + "init", + str(target), + "--force", + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + "--extension", + "git", + ] + created = CliRunner().invoke(app, arguments, catch_exceptions=False) + assert created.exit_code == 0, created.output + workflow = target / ".specify" / "workflows" / "speckit" / "workflow.yml" + constitution = target / ".specify" / "memory" / "constitution.md" + extension = target / ".specify" / "extensions" / "git" / "extension.yml" + assert workflow.is_file() + assert constitution.is_file() + assert extension.is_file() + workflow_before = workflow.read_text(encoding="utf-8") + constitution_before = constitution.read_text(encoding="utf-8") + extension_before = extension.read_text(encoding="utf-8") + + preview = CliRunner().invoke( + app, [*arguments, "--dry-run", "--json"], catch_exceptions=False + ) + assert preview.exit_code == 0, preview.output + payload = json.loads(preview.output) + workflow_action = _action_for( + payload, ".specify/workflows/speckit/workflow.yml" + ) + assert workflow_action["action"] == "skip" + assert workflow_action["provenance"] == "workflow" + assert workflow_action["source_id"] == "speckit" + constitution_action = _action_for( + payload, ".specify/memory/constitution.md" + ) + assert constitution_action["action"] == "skip" + assert constitution_action["provenance"] == "core" + extension_action = _action_for( + payload, ".specify/extensions/git/extension.yml" + ) + assert extension_action["action"] == "skip" + assert extension_action["provenance"] == "extension" + assert extension_action["source_id"] == "git" + assert workflow.read_text(encoding="utf-8") == workflow_before + assert constitution.read_text(encoding="utf-8") == constitution_before + assert extension.read_text(encoding="utf-8") == extension_before + + +def test_dry_run_preserves_preset_constitution_provenance_on_reinit( + tmp_path: Path, +) -> None: + target = tmp_path / "preset-constitution-preview" + arguments = [ + "init", + str(target), + "--force", + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + "--preset", + "self-test", + ] + created = CliRunner().invoke(app, arguments, catch_exceptions=False) + assert created.exit_code == 0, created.output + + preview = CliRunner().invoke( + app, [*arguments, "--dry-run", "--json"], catch_exceptions=False + ) + + assert preview.exit_code == 0, preview.output + constitution_action = _action_for( + json.loads(preview.output), ".specify/memory/constitution.md" + ) + assert constitution_action["action"] == "skip" + assert constitution_action["provenance"] == "preset" + assert constitution_action["source_id"] == "self-test" + + +@pytest.mark.parametrize( + ("arguments", "error_fragment"), + [ + (["--integration", "copilot", "--script", "sh"], "Must specify"), + (["project", "--integration", "unknown-agent"], "Unknown integration"), + ( + ["project", "--integration", "copilot", "--script", "invalid"], + "Invalid script type", + ), + ( + ["project", "--integration", "generic", "--script", "sh"], + "requires --integration-options", + ), + ], +) +def test_dry_run_json_validation_errors_use_single_json_envelope( + arguments: list[str], error_fragment: str +) -> None: + result = CliRunner().invoke( + app, + ["init", *arguments, "--dry-run", "--json", "--ignore-agent-tools"], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + payload = json.loads(result.output) + assert payload["dry_run"] is True + assert error_fragment in payload["error"] + assert payload["actions"] == [] + + +def test_dry_run_json_missing_agent_cli_uses_single_json_envelope( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr("specify_cli.command_init.check_tool", lambda _name: False) + + result = CliRunner().invoke( + app, + [ + "init", + str(tmp_path / "missing-cli-preview"), + "--dry-run", + "--json", + "--integration", + "kimi", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + payload = json.loads(result.output) + assert "kimi not found" in payload["error"] + assert payload["actions"] == [] + + +def test_dry_run_json_file_target_uses_single_json_envelope(tmp_path: Path) -> None: + target = tmp_path / "not-a-directory" + target.write_text("file\n", encoding="utf-8") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + payload = json.loads(result.output) + assert "exists but is not a directory" in payload["error"] + assert payload["actions"] == [] + + +def test_dry_run_leaves_url_extension_unresolved_without_creating_target( + tmp_path: Path, +) -> None: + target = tmp_path / "url-extension-preview" + extension_url = "https://example.com/spec-kit-extension.zip" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--extension", + extension_url, + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + url_action = _action_for(payload, extension_url) + assert url_action == { + "action": "unresolved", + "path": extension_url, + "provenance": "extension", + "source_id": extension_url, + "reason": "URL extensions are not fetched during dry-run", + } + assert not target.exists() + + +def test_dry_run_human_preview_escapes_untrusted_action_paths(tmp_path: Path) -> None: + target = tmp_path / "markup-safe-preview" + extension_url = "https://example.com/extensions/[/].zip" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--integration", + "copilot", + "--script", + "sh", + "--extension", + extension_url, + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert extension_url in result.output + assert not target.exists() + + +def test_dry_run_reports_optional_extension_failure(tmp_path: Path) -> None: + target = tmp_path / "failed-extension-preview" + extension = "nonexistent-xyz-ext" + + json_result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--extension", + extension, + ], + catch_exceptions=False, + ) + + assert json_result.exit_code == 0, json_result.output + payload = json.loads(json_result.output) + assert payload["failures"] == [ + { + "component": "extension", + "source_id": extension, + "error": f"Extension '{extension}' not found in bundled extensions or catalog", + } + ] + + human_result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--integration", + "copilot", + "--script", + "sh", + "--extension", + extension, + ], + catch_exceptions=False, + ) + + assert human_result.exit_code == 0, human_result.output + normalized = " ".join(human_result.output.split()) + assert f"failed extension:{extension}" in normalized + assert ( + f"Extension '{extension}' not found in bundled extensions or catalog" + in normalized + ) + assert not target.exists() + + +def test_dry_run_reports_optional_preset_failure(tmp_path: Path) -> None: + target = tmp_path / "failed-preset-preview" + preset = "nonexistent-xyz-preset" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--preset", + preset, + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert json.loads(result.output)["failures"] == [ + { + "component": "preset", + "source_id": preset, + "error": f"Preset '{preset}' not found in catalog", + } + ] + assert not target.exists() + + +def test_dry_run_changed_paths_match_a_forced_real_initialization(tmp_path: Path) -> None: + target = tmp_path / "parity-project" + command = target / ".github" / "skills" / "speckit-plan" / "SKILL.md" + command.parent.mkdir(parents=True) + command.write_text("user-owned content\n", encoding="utf-8") + before = _snapshot_files(target) + arguments = [ + "init", + str(target), + "--force", + "--integration", + "copilot", + "--script", + "sh", + ] + + preview = CliRunner().invoke( + app, [*arguments, "--dry-run", "--json"], catch_exceptions=False + ) + assert preview.exit_code == 0, preview.output + predicted = { + action["path"] + for action in json.loads(preview.output)["actions"] + if action["action"] in {"create", "overwrite"} + } + + actual = CliRunner().invoke(app, arguments, catch_exceptions=False) + assert actual.exit_code == 0, actual.output + after = _snapshot_files(target) + changed = {path for path, digest in after.items() if before.get(path) != digest} + + assert predicted == changed + + +def test_dry_run_includes_bundled_extension_artifacts(tmp_path: Path) -> None: + target = tmp_path / "extension-preview-project" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--extension", + "git", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + extension_action = _action_for( + payload, ".github/skills/speckit-git-feature/SKILL.md" + ) + assert extension_action["provenance"] == "extension" + assert extension_action["source_id"] == "git" + assert not target.exists() + + +def test_dry_run_uses_preset_registry_and_skill_marker_for_provenance( + tmp_path: Path, +) -> None: + target = tmp_path / "preset-preview-project" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--preset", + "self-test", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + preset_action = _action_for(payload, ".github/skills/speckit-specify/SKILL.md") + assert preset_action["provenance"] == "preset" + assert preset_action["source_id"] == "self-test" + assert not target.exists() + + +def test_dry_run_uses_registries_for_command_integration_provenance( + tmp_path: Path, +) -> None: + target = tmp_path / "command-provenance-preview" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "gemini", + "--script", + "sh", + "--ignore-agent-tools", + "--preset", + "self-test", + "--extension", + "git", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + preset_action = _action_for(payload, ".gemini/commands/speckit.specify.toml") + extension_action = _action_for(payload, ".gemini/commands/speckit.git.feature.toml") + assert (preset_action["provenance"], preset_action["source_id"]) == ( + "preset", + "self-test", + ) + assert (extension_action["provenance"], extension_action["source_id"]) == ( + "extension", + "git", + ) + assert not target.exists() + + +def test_dry_run_registry_owns_markerless_copilot_companion_prompt( + tmp_path: Path, +) -> None: + target = tmp_path / "copilot-command-preview" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--integration-options=--commands", + "--script", + "sh", + "--extension", + "git", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + prompt_action = _action_for( + payload, ".github/prompts/speckit.git.feature.prompt.md" + ) + assert (prompt_action["provenance"], prompt_action["source_id"]) == ( + "extension", + "git", + ) + assert not target.exists() + + +def test_dry_run_isolates_and_reports_hermes_home_writes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + real_home = tmp_path / "real-home" + existing_skill = real_home / ".hermes" / "skills" / "speckit-plan" / "SKILL.md" + existing_skill.parent.mkdir(parents=True) + existing_skill.write_text("user-owned content\n", encoding="utf-8") + monkeypatch.setenv("HOME", str(real_home)) + monkeypatch.setenv("USERPROFILE", str(real_home)) + target = tmp_path / "hermes-preview-project" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert existing_skill.read_text(encoding="utf-8") == "user-owned content\n" + payload = json.loads(result.output) + hermes_action = _action_for(payload, "~/.hermes/skills/speckit-plan/SKILL.md") + assert hermes_action["action"] == "overwrite" + assert hermes_action["provenance"] == "integration" + assert hermes_action["source_id"] == "hermes" + assert not target.exists() + + +def test_dry_run_rejects_symlinked_hermes_home_skill( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + real_home = tmp_path / "real-home" + external_skill = tmp_path / "external-skill.md" + external_skill.write_text("external content\n", encoding="utf-8") + skill_file = real_home / ".hermes" / "skills" / "speckit-plan" / "SKILL.md" + skill_file.parent.mkdir(parents=True) + try: + skill_file.symlink_to(external_skill) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + monkeypatch.setenv("HOME", str(real_home)) + monkeypatch.setenv("USERPROFILE", str(real_home)) + target = tmp_path / "hermes-home-link-preview" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + payload = json.loads(result.output) + assert "symlinked path component" in payload["error"] + assert external_skill.read_text(encoding="utf-8") == "external content\n" + assert skill_file.is_symlink() + assert not target.exists() + + +@pytest.mark.parametrize("linked_component", ["hermes", "skills"]) +def test_dry_run_reports_json_for_symlinked_hermes_home_parent( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + linked_component: str, +) -> None: + real_home = tmp_path / "real-home" + external = tmp_path / "external-home-target" + external.mkdir() + if linked_component == "hermes": + real_home.mkdir() + link = real_home / ".hermes" + else: + (real_home / ".hermes").mkdir(parents=True) + link = real_home / ".hermes" / "skills" + try: + link.symlink_to(external, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + monkeypatch.setenv("HOME", str(real_home)) + monkeypatch.setenv("USERPROFILE", str(real_home)) + + result = CliRunner().invoke( + app, + [ + "init", + str(tmp_path / f"hermes-{linked_component}-link-preview"), + "--dry-run", + "--json", + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + assert "symlinked path component" in json.loads(result.output)["error"] + assert list(external.iterdir()) == [] + + +def test_dry_run_reports_preserve_for_unchanged_hermes_home_skill( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + real_home = tmp_path / "real-home" + real_home.mkdir() + monkeypatch.setenv("HOME", str(real_home)) + monkeypatch.setenv("USERPROFILE", str(real_home)) + target = tmp_path / "hermes-reinit-preview" + arguments = [ + "init", + str(target), + "--force", + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + ] + created = CliRunner().invoke(app, arguments, catch_exceptions=False) + assert created.exit_code == 0, created.output + + preview = CliRunner().invoke( + app, [*arguments, "--dry-run", "--json"], catch_exceptions=False + ) + + assert preview.exit_code == 0, preview.output + action = _action_for( + json.loads(preview.output), "~/.hermes/skills/speckit-plan/SKILL.md" + ) + assert action["action"] == "preserve" + assert action["provenance"] == "integration" + assert action["source_id"] == "hermes" + + +def test_dry_run_omits_unmanaged_hermes_home_skill( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + real_home = tmp_path / "real-home" + unmanaged = ( + real_home / ".hermes" / "skills" / "speckit-obsolete" / "SKILL.md" + ) + unmanaged.parent.mkdir(parents=True) + unmanaged.write_text("user-owned\n", encoding="utf-8") + monkeypatch.setenv("HOME", str(real_home)) + monkeypatch.setenv("USERPROFILE", str(real_home)) + + result = CliRunner().invoke( + app, + [ + "init", + str(tmp_path / "hermes-unmanaged-preview"), + "--dry-run", + "--json", + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert all( + action["path"] != "~/.hermes/skills/speckit-obsolete/SKILL.md" + for action in json.loads(result.output)["actions"] + ) + assert unmanaged.read_text(encoding="utf-8") == "user-owned\n" + + +def test_dry_run_reports_kimi_legacy_removals(tmp_path: Path) -> None: + target = tmp_path / "kimi-migration-preview" + legacy_skill = target / ".kimi" / "skills" / "speckit-oldcmd" / "SKILL.md" + legacy_skill.parent.mkdir(parents=True) + legacy_skill.write_text("# Legacy\n", encoding="utf-8") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "kimi", + "--integration-options=--migrate-legacy", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert ( + _action_for(payload, ".kimi/skills/speckit-oldcmd/SKILL.md")["action"] + == "remove" + ) + assert ( + _action_for(payload, ".kimi-code/skills/speckit-oldcmd/SKILL.md")["action"] + == "create" + ) + assert legacy_skill.read_text(encoding="utf-8") == "# Legacy\n" + + +def test_dry_run_does_not_write_through_external_hermes_symlink( + tmp_path: Path, +) -> None: + external = tmp_path / "external-hermes" + external.mkdir() + target = tmp_path / "hermes-external-link" + target.mkdir() + try: + (target / ".hermes").symlink_to(external.resolve()) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "hermes", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + assert "symlinked path component" in json.loads(result.output)["error"] + assert list(external.iterdir()) == [] + assert (target / ".hermes").is_symlink() + assert (target / ".hermes").resolve() == external.resolve() + + +def test_dry_run_remaps_in_project_absolute_symlinks(tmp_path: Path) -> None: + target = tmp_path / "symlink-preview" + real_commands = target / "kilo-store" / "commands" + real_commands.mkdir(parents=True) + kilo_dir = target / ".kilo" + kilo_dir.mkdir() + try: + (kilo_dir / "commands").symlink_to(real_commands.resolve()) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "kilocode", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert { + action["path"] for action in payload["actions"] + } >= {"kilo-store/commands/speckit.plan.md"} + assert list(real_commands.iterdir()) == [] + assert (kilo_dir / "commands").resolve() == real_commands.resolve() + + +@pytest.mark.skipif(os.name == "nt", reason="chmod does not remove read access on Windows") +def test_dry_run_ignores_unreadable_unmanaged_file(tmp_path: Path) -> None: + target = tmp_path / "unreadable-unmanaged" + target.mkdir() + unrelated = target / "unrelated.bin" + unrelated.write_bytes(b"not used by init") + unrelated.chmod(0) + + try: + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + finally: + unrelated.chmod(0o600) + + assert result.exit_code == 0, result.output + assert all( + action["path"] != "unrelated.bin" + for action in json.loads(result.output)["actions"] + ) + + +@pytest.mark.skipif(os.name == "nt", reason="chmod does not remove read access on Windows") +def test_dry_run_ignores_unreadable_unmanaged_file_in_selected_root( + tmp_path: Path, +) -> None: + target = tmp_path / "unreadable-selected-root" + unrelated = target / ".github" / "skills" / "private" / "secret.txt" + unrelated.parent.mkdir(parents=True) + unrelated.write_text("not managed by Spec Kit\n", encoding="utf-8") + unrelated.chmod(0) + + try: + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + finally: + unrelated.chmod(0o600) + + assert result.exit_code == 0, result.output + assert all( + action["path"] != ".github/skills/private/secret.txt" + for action in json.loads(result.output)["actions"] + ) + assert unrelated.read_text(encoding="utf-8") == "not managed by Spec Kit\n" + + +@pytest.mark.skipif(os.name == "nt", reason="chmod does not remove read access on Windows") +def test_dry_run_reports_unreadable_managed_file_in_selected_root( + tmp_path: Path, +) -> None: + target = tmp_path / "unreadable-managed-output" + managed = target / ".github" / "skills" / "speckit-plan" / "SKILL.md" + managed.parent.mkdir(parents=True) + managed.write_text("managed but unreadable\n", encoding="utf-8") + managed.chmod(0) + + try: + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + finally: + managed.chmod(0o600) + + assert result.exit_code == 1, result.output + payload = json.loads(result.output) + assert "failed to stage preview inputs" in payload["error"] + assert "Permission denied" in payload["error"] + assert managed.read_text(encoding="utf-8") == "managed but unreadable\n" + + +def test_dry_run_rejects_selected_directory_junction_before_copy( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "junction-preview" + selected_root = target / ".github" / "skills" + selected_root.mkdir(parents=True) + (selected_root / "keep.txt").write_text("external-like\n", encoding="utf-8") + real_is_junction = getattr(Path, "is_junction", None) + + def fake_is_junction(path: Path) -> bool: + return path == selected_root or bool( + real_is_junction and real_is_junction(path) + ) + + monkeypatch.setattr(Path, "is_junction", fake_is_junction, raising=False) + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + payload = json.loads(result.output) + assert "junction" in payload["error"].lower() + assert (selected_root / "keep.txt").read_text(encoding="utf-8") == "external-like\n" + + +def test_windows_junction_fallback_uses_reparse_tag( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + mount_point_tag = getattr(stat, "IO_REPARSE_TAG_MOUNT_POINT", 0xA0000003) + monkeypatch.setattr( + Path, + "lstat", + lambda _path: SimpleNamespace(st_reparse_tag=mount_point_tag), + ) + + assert _windows_path_is_junction(tmp_path) + + +@pytest.mark.skipif(os.name != "nt", reason="Windows junction behavior") +def test_dry_run_rejects_real_windows_junction_before_copy(tmp_path: Path) -> None: + target = tmp_path / "real-junction-preview" + external = tmp_path / "external-skills" + external.mkdir() + selected_root = target / ".github" / "skills" + selected_root.parent.mkdir(parents=True) + subprocess.run( + ["cmd", "/c", "mklink", "/J", str(selected_root), str(external)], + check=True, + capture_output=True, + text=True, + ) + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "ps", + "--ignore-agent-tools", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + assert "junction" in json.loads(result.output)["error"].lower() + assert list(external.iterdir()) == [] + + +def test_remap_in_project_absolute_symlinks_points_at_staged_copy( + tmp_path: Path, +) -> None: + project = tmp_path / "proj" + store = project / "store" + store.mkdir(parents=True) + (store / "file.txt").write_text("ok\n", encoding="utf-8") + link = project / "link" + try: + link.symlink_to(store.resolve()) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + staged = tmp_path / "staged" + _stage_project_copy(project, staged) + + remapped = Path(os.readlink(staged / "link")) + _assert_same_path(remapped, staged / "store") + assert (staged / "link" / "file.txt").read_text(encoding="utf-8") == "ok\n" + _assert_same_path(Path(os.readlink(link)), store) + + +def test_stage_project_copy_limits_copy_to_selected_paths(tmp_path: Path) -> None: + project = tmp_path / "proj" + managed = project / ".specify" / "state.json" + managed.parent.mkdir(parents=True) + managed.write_text("{}\n", encoding="utf-8") + unrelated = project / ".git" / "objects" / "large-object" + unrelated.parent.mkdir(parents=True) + unrelated.write_bytes(b"unrelated") + staged = tmp_path / "staged" + + _stage_project_copy(project, staged, {Path(".specify")}) + + assert (staged / ".specify" / "state.json").read_text(encoding="utf-8") == "{}\n" + assert not (staged / ".git").exists() + + +def test_stage_selected_path_quarantines_symlinked_parent(tmp_path: Path) -> None: + project = tmp_path / "proj" + project.mkdir() + outside = tmp_path / "outside-agent" + commands = outside / "commands" + commands.mkdir(parents=True) + (commands / "keep.md").write_text("external\n", encoding="utf-8") + try: + (project / ".agent").symlink_to(outside, target_is_directory=True) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + staged = tmp_path / "staged" + + _stage_project_copy(project, staged, {Path(".agent/commands")}) + + staged_agent = staged / ".agent" + assert staged_agent.is_symlink() + assert not (staged_agent / "commands" / "keep.md").exists() + assert (commands / "keep.md").read_text(encoding="utf-8") == "external\n" + + +def test_stage_copy_quarantines_external_absolute_symlinks(tmp_path: Path) -> None: + project = tmp_path / "proj" + project.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (outside / "keep.txt").write_text("keep\n", encoding="utf-8") + link = project / "escape" + try: + link.symlink_to(outside.resolve()) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + staged = tmp_path / "staged" + _stage_project_copy(project, staged) + + staged_escape = staged / "escape" + assert staged_escape.is_symlink() + dummy = Path(os.readlink(staged_escape)) + if not dummy.is_absolute(): + dummy = staged_escape.parent / dummy + assert not _is_within_root(dummy, staged) + assert not _is_within_root(dummy, outside) + (staged_escape / "skills").mkdir() + assert not (outside / "skills").exists() + assert (outside / "keep.txt").read_text(encoding="utf-8") == "keep\n" + _assert_same_path(Path(os.readlink(link)), outside) + + +def test_dry_run_does_not_write_through_external_command_symlink( + tmp_path: Path, +) -> None: + target = tmp_path / "escape-preview" + outside = tmp_path / "outside-commands" + outside.mkdir() + kilo_dir = target / ".kilo" + kilo_dir.mkdir(parents=True) + try: + (kilo_dir / "commands").symlink_to(outside.resolve()) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "kilocode", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 1, result.output + payload = json.loads(result.output) + assert payload["dry_run"] is True + assert "escapes project root" in payload["error"] + assert not any( + action["path"].startswith(".kilo/commands/") for action in payload["actions"] + ) + assert list(outside.iterdir()) == [] + assert (kilo_dir / "commands").is_symlink() + assert (kilo_dir / "commands").resolve() == outside.resolve() + + +def test_snapshot_fingerprint_includes_permission_bits(tmp_path: Path) -> None: + script = tmp_path / "script.sh" + script.write_text("#!/bin/sh\necho ok\n", encoding="utf-8") + script.chmod(0o644) + mode_before = script.stat().st_mode & 0o777 + before = _snapshot_files(tmp_path) + script.chmod(0o755) + mode_after = script.stat().st_mode & 0o777 + after = _snapshot_files(tmp_path) + if mode_before == mode_after: + pytest.skip("filesystem does not distinguish permission bits") + assert before != after + + +@pytest.mark.skipif(os.name == "nt", reason="ensure_executable_scripts is a no-op on Windows") +def test_dry_run_reports_overwrite_when_shebang_script_lacks_execute_bit( + tmp_path: Path, +) -> None: + target = tmp_path / "chmod-preview" + arguments = [ + "init", + str(target), + "--force", + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + ] + created = CliRunner().invoke(app, arguments, catch_exceptions=False) + assert created.exit_code == 0, created.output + scripts = [ + path + for path in (target / ".specify" / "scripts").rglob("*.sh") + if path.is_file() and path.read_bytes().startswith(b"#!") + ] + assert scripts + script = scripts[0] + script.chmod(script.stat().st_mode & ~0o111) + assert not (script.stat().st_mode & 0o111) + + preview = CliRunner().invoke( + app, [*arguments, "--dry-run", "--json"], catch_exceptions=False + ) + assert preview.exit_code == 0, preview.output + relative = script.relative_to(target).as_posix() + action = _action_for(json.loads(preview.output), relative) + assert action["action"] == "overwrite" + assert not (script.stat().st_mode & 0o111) + + +def test_dry_run_resolves_home_relative_local_extension( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + bundled = _locate_bundled_extension("git") + assert bundled is not None + home = tmp_path / "home" + extension = home / "exts" / "git" + shutil.copytree(bundled, extension) + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + + target = tmp_path / "tilde-extension-preview" + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--extension", + "~/exts/git", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + extension_action = _action_for( + payload, ".specify/extensions/git/extension.yml" + ) + assert extension_action["provenance"] == "extension" + assert extension_action["source_id"] == "git" + assert not target.exists() + + +def test_here_dry_run_resolves_bare_relative_local_preset( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + bundled = _locate_bundled_preset("self-test") + assert bundled is not None + target = tmp_path / "bare-relative-preset-preview" + target.mkdir() + shutil.copytree(bundled, target / "local-preset") + monkeypatch.chdir(target) + + result = CliRunner().invoke( + app, + [ + "init", + "--here", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--preset", + "local-preset", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["failures"] == [] + preset_action = _action_for(payload, ".github/skills/speckit-specify/SKILL.md") + assert preset_action["provenance"] == "preset" + assert not (target / ".specify").exists() + + +def test_init_resolves_home_relative_local_preset( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + bundled = _locate_bundled_preset("self-test") + assert bundled is not None + home = tmp_path / "home" + home.mkdir() + shutil.copytree(bundled, home / "local-preset") + target = tmp_path / "home-relative-preset-project" + target.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + monkeypatch.chdir(target) + + result = CliRunner().invoke( + app, + [ + "init", + "--here", + "--force", + "--non-interactive", + "--ignore-agent-tools", + "--integration", + "copilot", + "--script", + "sh", + "--preset", + "~/local-preset", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert (target / ".specify" / "presets" / "self-test" / "preset.yml").is_file() diff --git a/tests/test_registrar_path_traversal.py b/tests/test_registrar_path_traversal.py index 141fa05186..2e50c229b3 100644 --- a/tests/test_registrar_path_traversal.py +++ b/tests/test_registrar_path_traversal.py @@ -400,7 +400,7 @@ def test_safe_command_and_alias_still_register(self, tmp_path): / "SKILL.md" ).exists() - def test_copilot_nested_alias_creates_companion_prompt(self, tmp_path): + def test_copilot_nested_alias_creates_and_removes_companion_prompt(self, tmp_path): project, ext_dir = _project_and_source(tmp_path) agents_dir = project / ".github" / "agents" agents_dir.mkdir(parents=True) @@ -423,3 +423,14 @@ def test_copilot_nested_alias_creates_companion_prompt(self, tmp_path): / "group" / "run.prompt.md" ).is_file() + + registrar.unregister_commands({"copilot": registered}, project) + + assert not (agents_dir / "group" / "run.agent.md").exists() + assert not ( + project + / ".github" + / "prompts" + / "group" + / "run.prompt.md" + ).exists()