diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index f8afcf4f55..c9fd25f7df 100644 --- a/src/specify_cli/__init__.py +++ b/src/specify_cli/__init__.py @@ -214,10 +214,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", @@ -265,6 +267,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/agents.py b/src/specify_cli/agents.py index dede50e0b1..bc8dc3a499 100644 --- a/src/specify_cli/agents.py +++ b/src/specify_cli/agents.py @@ -557,6 +557,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. @@ -651,6 +679,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 = [] @@ -677,14 +710,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: @@ -835,18 +862,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, @@ -927,6 +950,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, @@ -1084,6 +1109,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 = ( @@ -1191,6 +1219,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: @@ -1241,6 +1271,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(): @@ -1275,6 +1308,8 @@ def register_commands_for_non_skill_agents( ) if registered: results[agent_name] = registered + except IntegrationOutputPathError: + raise except ValueError: continue return results @@ -1294,6 +1329,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 @@ -1302,6 +1347,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. @@ -1313,6 +1361,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 ) @@ -1330,25 +1382,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/commands/bundle/__init__.py b/src/specify_cli/commands/bundle/__init__.py index 165f674a36..65271145af 100644 --- a/src/specify_cli/commands/bundle/__init__.py +++ b/src/specify_cli/commands/bundle/__init__.py @@ -126,6 +126,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/commands/init.py b/src/specify_cli/commands/init.py index 2f686e2fa9..88c4febd59 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/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 @@ -53,6 +57,1191 @@ 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(): + try: + entries = list(real_destination.iterdir()) + except OSError: + entries = [] + 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 _ignore_special_files(directory: str, names: list[str]) -> set[str]: + """Skip sockets, devices, and other non-file tree entries during staging.""" + ignored: set[str] = set() + for name in names: + candidate = Path(directory) / name + try: + if ( + not candidate.is_symlink() + and not candidate.is_file() + and not candidate.is_dir() + ): + 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 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=_ignore_special_files, + ) + 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 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=_ignore_special_files, + ) + 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)] + home_seed_paths, home_owned_paths = _preview_home_seed_paths( + real_home, selected_integration + ) + + 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() + _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() + + 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.commands.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 +1331,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 +1354,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" @@ -223,6 +1424,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: @@ -233,6 +1441,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") @@ -250,6 +1461,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: @@ -336,6 +1548,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. @@ -391,13 +1613,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))}'" @@ -411,18 +1643,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() @@ -430,25 +1677,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. - console.print( - "[yellow]Template files will be merged with existing content and may overwrite existing files[/yellow]" - ) + if not (dry_run and json_output): console.print( - "[cyan]--force supplied: skipping confirmation and proceeding with merge[/cyan]" + f"[yellow]Warning:[/yellow] Current directory is not empty ({len(existing_items)} items)" ) - 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 @@ -487,23 +1738,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" @@ -518,6 +1778,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())}" ) @@ -525,10 +1791,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()} @@ -546,6 +1813,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" ) @@ -568,15 +1842,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" @@ -592,6 +1874,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())}" ) @@ -610,8 +1898,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") @@ -771,6 +2084,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 @@ -799,6 +2118,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 = { @@ -815,7 +2135,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: @@ -839,6 +2162,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." ) @@ -847,6 +2175,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." @@ -865,6 +2198,9 @@ def init( zip_path, speckit_ver ) except PresetError as preset_err: + _record_init_plan_failure( + "preset", preset, str(preset_err) + ) _print_cli_warning( "install", "preset", @@ -879,6 +2215,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", @@ -914,6 +2251,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 3968e4fcbe..a45898fb70 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -1378,6 +1378,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 @@ -1625,6 +1630,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") @@ -2044,6 +2051,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, @@ -2095,6 +2122,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 @@ -3183,6 +3212,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) @@ -3459,6 +3489,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/_commands.py b/src/specify_cli/extensions/_commands.py index 11ab50385e..2d9e9252bb 100644 --- a/src/specify_cli/extensions/_commands.py +++ b/src/specify_cli/extensions/_commands.py @@ -1227,6 +1227,7 @@ def extension_remove( ): """Uninstall an extension.""" from . import ExtensionManager + from ..integrations.base import IntegrationOutputPathError project_root = _require_specify_project() manager = ExtensionManager(project_root) @@ -1273,7 +1274,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 27c43582b0..ae67aff514 100644 --- a/src/specify_cli/integrations/base.py +++ b/src/specify_cli/integrations/base.py @@ -58,6 +58,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. @@ -963,6 +967,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 a82eb6fd4d..8e7bfd15bc 100644 --- a/src/specify_cli/integrations/hermes/__init__.py +++ b/src/specify_cli/integrations/hermes/__init__.py @@ -18,10 +18,40 @@ import yaml -from ..base import IntegrationOption, SkillsIntegration, yaml_quote +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 + try: + is_junction = current.is_junction() + except (AttributeError, OSError): + is_junction = False + if current.is_symlink() or is_junction: + 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. @@ -58,6 +88,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 @@ -102,6 +140,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") @@ -109,7 +163,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] = [] @@ -213,7 +266,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 @@ -243,7 +296,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()): @@ -253,9 +308,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/__init__.py b/src/specify_cli/presets/__init__.py index 6b80b4fe1f..30c0ff7530 100644 --- a/src/specify_cli/presets/__init__.py +++ b/src/specify_cli/presets/__init__.py @@ -3828,6 +3828,48 @@ def _unregister_skills_in_dir( return mutated_names + 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, @@ -3858,6 +3900,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: @@ -4193,6 +4236,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/_commands.py b/src/specify_cli/presets/_commands.py index ab74a8e029..d152ac007c 100644 --- a/src/specify_cli/presets/_commands.py +++ b/src/specify_cli/presets/_commands.py @@ -412,6 +412,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() @@ -421,7 +422,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..9b0bdb8dd6 100644 --- a/tests/integrations/test_integration_hermes.py +++ b/tests/integrations/test_integration_hermes.py @@ -10,8 +10,13 @@ non-destructive to a developer's real Hermes installation. """ +import json +import shutil from pathlib import Path +import pytest +import yaml + from specify_cli.integrations import get_integration from specify_cli.integrations.manifest import IntegrationManifest @@ -65,6 +70,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 +239,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 result.output + 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 aec32dc4ba..4f1d3d1bf6 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -43,6 +43,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 @@ -4730,6 +4731,55 @@ 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_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..0cdb41f642 --- /dev/null +++ b/tests/test_init_dry_run.py @@ -0,0 +1,1621 @@ +"""CLI contract tests for ``specify init --dry-run``.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from pathlib import Path + +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.commands.init import ( + _is_within_root, + _normalize_fs_path, + _preview_child_failure_message, + _preview_content_ownership, + _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_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.commands.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"] + ) + + +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()