From 70420547aad9ed789bf5ccb0b7467c2e6ea1db4f Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Tue, 25 Aug 2026 20:08:57 +0800 Subject: [PATCH 1/8] Preview init changes before a forced merge can alter a project Dry-run stages the target and invokes the public initializer in an isolated child process, then reports create, overwrite, and preserve actions without writing to the requested project. This keeps previews aligned with integration-specific installation behavior. --- src/specify_cli/commands/bundle/__init__.py | 2 + src/specify_cli/commands/init.py | 268 ++++++++++++++++++-- tests/test_init_dry_run.py | 214 ++++++++++++++++ 3 files changed, 464 insertions(+), 20 deletions(-) create mode 100644 tests/test_init_dry_run.py 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..0f4b5d70fd 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -2,11 +2,14 @@ from __future__ import annotations +import hashlib +import json import os import shlex import shutil import subprocess import sys +import tempfile from pathlib import Path from typing import Any @@ -53,6 +56,191 @@ def _ext_spec_is_url(ext_spec: str) -> bool: return False +def _snapshot_files(root: Path) -> dict[str, str]: + """Return SHA-256 digests 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 + digest = hashlib.sha256(path.read_bytes()).hexdigest() + files[path.relative_to(root).as_posix()] = digest + return files + + +def _preview_manifest_provenance(staged_root: Path) -> dict[str, str]: + """Map manifest-tracked staged paths to their installation source.""" + provenance: dict[str, str] = {} + manifests = staged_root / ".specify" / "integrations" + if not manifests.is_dir(): + return provenance + + for manifest_path in manifests.glob("*.manifest.json"): + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + key = str(manifest.get("key", manifest_path.stem.removesuffix(".manifest"))) + source = "core" if key == "speckit" else f"integration:{key}" + for relative_path in manifest.get("files", {}): + provenance[str(relative_path)] = source + except (OSError, TypeError, ValueError): + continue + return provenance + + +def _preview_default_provenance(relative_path: str) -> str: + if relative_path.startswith(".specify/workflows/"): + return "workflow" + if relative_path.startswith(".specify/extensions/"): + return "extension" + if relative_path.startswith(".specify/presets/"): + return "preset" + if relative_path.startswith(".specify/"): + return "core" + return "integration" + + +def _build_preview_actions( + initial_files: dict[str, str], staged_root: Path +) -> list[dict[str, str]]: + """Classify files produced by a staged initialization.""" + staged_files = _snapshot_files(staged_root) + provenance = _preview_manifest_provenance(staged_root) + candidates = { + path + for path, digest in staged_files.items() + if initial_files.get(path) != digest + } + candidates.update(path for path in provenance if path in staged_files) + + actions: list[dict[str, str]] = [] + for path in sorted(candidates): + staged_digest = staged_files[path] + initial_digest = initial_files.get(path) + action = ( + "create" + if initial_digest is None + else "overwrite" + if initial_digest != staged_digest + else "preserve" + ) + actions.append( + { + "action": action, + "path": path, + "provenance": provenance.get(path, _preview_default_provenance(path)), + } + ) + 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["conflict"]: + console.print( + "[yellow]conflict[/yellow] target directory is non-empty; rerun with --force " + "to preview a forced merge" + ) + return + + for record in payload["actions"]: + console.print( + f"{record['action']:<10} {record['path']} " + f"[dim]({record['provenance']})[/dim]" + ) + + +def _preview_init( + *, + project_path: Path, + directory_conflict: 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": directory_conflict, + "actions": [], + } + if directory_conflict: + _emit_dry_run_preview(payload, json_output=json_output) + return + + initial_files = _snapshot_files(project_path) + url_extensions = [spec for spec in extensions or [] if _ext_spec_is_url(spec)] + staged_extensions = [spec for spec in extensions or [] if not _ext_spec_is_url(spec)] + + with tempfile.TemporaryDirectory(prefix="specify-init-preview-") as tmp_dir: + staged_root = Path(tmp_dir) / "project" + if project_path.exists(): + shutil.copytree(project_path, staged_root, symlinks=True) + + # 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. + command = [ + sys.executable, + "-c", + "from specify_cli import main; main()", + "init", + str(staged_root), + "--force", + "--non-interactive", + "--integration", + selected_integration, + "--script", + script_type, + ] + if ignore_agent_tools: + command.append("--ignore-agent-tools") + if integration_options: + command.extend(["--integration-options", integration_options]) + if preset: + command.extend(["--preset", preset]) + for extension in staged_extensions: + command.extend(["--extension", extension]) + if trust_extension_urls: + command.append("--trust-extension-urls") + + result = subprocess.run( + command, + cwd=Path.cwd(), + capture_output=True, + text=True, + check=False, + ) + if result.returncode: + details = (result.stderr or result.stdout).strip().replace("\n", " ") + raise RuntimeError(f"staged initialization failed: {details[:240]}") + + payload["actions"] = _build_preview_actions(initial_files, staged_root) + + for spec in url_extensions: + payload["actions"].append( + { + "action": "unresolved", + "path": spec, + "provenance": "extension:url", + } + ) + payload["actions"].sort(key=lambda action: action["path"]) + _emit_dry_run_preview(payload, json_output=json_output) + + def _confirm_extension_url_trust( url_specs: list[str], *, @@ -336,6 +524,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,7 +589,12 @@ 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 @@ -423,6 +626,7 @@ def init( raise typer.Exit(1) dir_existed_before = False + directory_conflict = False if here: project_name = Path.cwd().name project_path = Path.cwd() @@ -430,17 +634,21 @@ def init( existing_items = list(project_path.iterdir()) if existing_items: - console.print( - f"[yellow]Warning:[/yellow] Current directory is not empty ({len(existing_items)} items)" - ) - if force: - # Proceeding: the merge/overwrite warning is accurate here. + if not (dry_run and json_output): console.print( - "[yellow]Template files will be merged with existing content and may overwrite existing files[/yellow]" - ) - 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)" ) + 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: console.print( "[red]Error:[/red] Current directory is not empty and " @@ -492,17 +700,20 @@ def init( ) 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]" - ) + if not (dry_run and json_output): + console.print( + f"[cyan]--force supplied: merging into existing directory '[cyan]{safe_name}[/cyan]'[/cyan]" + ) else: error_panel = Panel( f"Directory already exists: '[cyan]{safe_name}[/cyan]'\n" @@ -568,9 +779,10 @@ 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) @@ -610,8 +822,24 @@ 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: + _preview_init( + project_path=project_path, + directory_conflict=directory_conflict, + 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") diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py new file mode 100644 index 0000000000..835b719e23 --- /dev/null +++ b/tests/test_init_dry_run.py @@ -0,0 +1,214 @@ +"""CLI contract tests for ``specify init --dry-run``.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from typer.testing import CliRunner + +from specify_cli import app +from specify_cli.commands.init import _snapshot_files + + +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" + } + assert not target.exists() + + +def test_forced_dry_run_reports_overwrite_without_changing_existing_file( + tmp_path: Path, +) -> None: + target = tmp_path / "existing-project" + 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), + "--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", ".github/skills/speckit-plan/SKILL.md")} + assert command.read_text(encoding="utf-8") == "user-owned content\n" + + +def test_non_forced_dry_run_reports_existing_target_conflict(tmp_path: Path) -> None: + target = tmp_path / "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 existing.read_text(encoding="utf-8") == "keep\n" + + +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) + assert { + (action["action"], action["path"]) + for action in payload["actions"] + } >= {("unresolved", extension_url)} + 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) + assert any(action["provenance"] == "extension" for action in payload["actions"]) + assert not target.exists() From 0fed0f7968b78589f16e25851554432c97a92da6 Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Tue, 1 Sep 2026 10:20:41 +0800 Subject: [PATCH 2/8] fix(init): keep dry-run staging isolated from user-home writes HermesIntegration.setup() was writing into the real user home during init --dry-run, which let preview runs touch global files. Ownership for materialized preset and extension commands also depended on destination-path heuristics, which misclassified agent-directory outputs and lost the true provenance signal. Dry-run staging now keeps home-scoped output in an isolated preview environment, and ownership is derived from the staged registries and markers instead of from destination paths. The manifest keeps the concrete source_id separate from the required provenance category. --- src/specify_cli/commands/init.py | 303 ++++++++++++++++++++++++++++--- tests/test_init_dry_run.py | 172 +++++++++++++++++- 2 files changed, 441 insertions(+), 34 deletions(-) diff --git a/src/specify_cli/commands/init.py b/src/specify_cli/commands/init.py index 0f4b5d70fd..8cba58863b 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -70,49 +70,258 @@ def _snapshot_files(root: Path) -> dict[str, str]: return files -def _preview_manifest_provenance(staged_root: Path) -> dict[str, str]: - """Map manifest-tracked staged paths to their installation source.""" - provenance: dict[str, str] = {} +def _snapshot_matching_files(root: Path, relative_paths: set[str]) -> dict[str, str]: + """Return digests for selected regular files below *root*.""" + files: dict[str, str] = {} + for relative_path in relative_paths: + path = root / relative_path + if not path.is_file() or path.is_symlink(): + continue + files[relative_path] = hashlib.sha256(path.read_bytes()).hexdigest() + return files + + +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"), + } + ) + 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 + + +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 provenance + 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("key", manifest_path.stem.removesuffix(".manifest"))) - source = "core" if key == "speckit" else f"integration:{key}" + 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", {}): - provenance[str(relative_path)] = source + ownership[str(relative_path)] = source except (OSError, TypeError, ValueError): continue - return provenance + 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 source_id: + return category, source_id + if marker.startswith(""): + value = marker.removeprefix("").strip() + if value.startswith(("preset:", "extension:")): + category, source_id = value.split(":", 1) + if source_id: + 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]] +) -> dict[str, PreviewOwnership]: + """Map staged generated artifacts using their embedded ownership markers.""" + ownership: dict[str, PreviewOwnership] = {} + for relative_path in _snapshot_files(staged_root): + 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_provenance(relative_path: str) -> str: +def _preview_default_ownership( + relative_path: str, default: PreviewOwnership +) -> PreviewOwnership: if relative_path.startswith(".specify/workflows/"): - return "workflow" + 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/"): - return "extension" + 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/"): - return "preset" + 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" - return "integration" + return "core", None + return default def _build_preview_actions( - initial_files: dict[str, str], staged_root: Path + initial_files: dict[str, str], + staged_root: Path, + *, + path_prefix: str = "", + ownership: dict[str, PreviewOwnership] | None = None, + default_ownership: PreviewOwnership = ("integration", None), ) -> list[dict[str, str]]: """Classify files produced by a staged initialization.""" staged_files = _snapshot_files(staged_root) - provenance = _preview_manifest_provenance(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 provenance if path 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): @@ -125,13 +334,17 @@ def _build_preview_actions( if initial_digest != staged_digest else "preserve" ) - actions.append( - { - "action": action, - "path": path, - "provenance": provenance.get(path, _preview_default_provenance(path)), - } + 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 @@ -150,10 +363,10 @@ def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None return for record in payload["actions"]: - console.print( - f"{record['action']:<10} {record['path']} " - f"[dim]({record['provenance']})[/dim]" - ) + source = record["provenance"] + if record.get("source_id"): + source = f"{source}:{record['source_id']}" + console.print(f"{record['action']:<10} {record['path']} [dim]({source})[/dim]") def _preview_init( @@ -181,11 +394,14 @@ def _preview_init( return initial_files = _snapshot_files(project_path) + real_home = Path.home() url_extensions = [spec for spec in extensions or [] if _ext_spec_is_url(spec)] staged_extensions = [spec for spec in extensions or [] if not _ext_spec_is_url(spec)] with tempfile.TemporaryDirectory(prefix="specify-init-preview-") as tmp_dir: staged_root = Path(tmp_dir) / "project" + staged_home = Path(tmp_dir) / "home" + staged_home.mkdir() if project_path.exists(): shutil.copytree(project_path, staged_root, symlinks=True) @@ -222,19 +438,48 @@ def _preview_init( capture_output=True, text=True, check=False, + env=_preview_subprocess_env(staged_home), ) if result.returncode: details = (result.stderr or result.stdout).strip().replace("\n", " ") raise RuntimeError(f"staged initialization failed: {details[:240]}") - payload["actions"] = _build_preview_actions(initial_files, staged_root) + registry_sources = _preview_registry_sources(staged_root) + project_ownership = _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) + ) + payload["actions"] = _build_preview_actions( + initial_files, + staged_root, + ownership=project_ownership, + default_ownership=("integration", selected_integration), + ) + staged_home_files = _snapshot_files(staged_home) + initial_home_files = _snapshot_matching_files(real_home, set(staged_home_files)) + home_ownership = registry_home_ownership + home_ownership.update(_preview_marker_ownership(staged_home, registry_sources)) + payload["actions"].extend( + _build_preview_actions( + initial_home_files, + staged_home, + path_prefix="~/", + ownership=home_ownership, + default_ownership=("integration", selected_integration), + ) + ) for spec in url_extensions: payload["actions"].append( { "action": "unresolved", "path": spec, - "provenance": "extension:url", + "provenance": "extension", + "source_id": spec, } ) payload["actions"].sort(key=lambda action: action["path"]) diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py index 835b719e23..032bdd05d0 100644 --- a/tests/test_init_dry_run.py +++ b/tests/test_init_dry_run.py @@ -5,11 +5,18 @@ import json from pathlib import Path +import pytest from typer.testing import CliRunner from specify_cli import app from specify_cli.commands.init import _snapshot_files +_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 test_dry_run_reports_new_project_files_without_creating_target(tmp_path: Path) -> None: target = tmp_path / "preview-project" @@ -58,6 +65,12 @@ def test_dry_run_json_is_machine_readable_and_has_no_target_writes(tmp_path: Pat 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" + assert { + action["provenance"] for action in payload["actions"] + } <= _PROVENANCE_CATEGORIES assert not target.exists() @@ -147,10 +160,13 @@ def test_dry_run_leaves_url_extension_unresolved_without_creating_target( assert result.exit_code == 0, result.output payload = json.loads(result.output) - assert { - (action["action"], action["path"]) - for action in payload["actions"] - } >= {("unresolved", extension_url)} + url_action = _action_for(payload, extension_url) + assert url_action == { + "action": "unresolved", + "path": extension_url, + "provenance": "extension", + "source_id": extension_url, + } assert not target.exists() @@ -210,5 +226,151 @@ def test_dry_run_includes_bundled_extension_artifacts(tmp_path: Path) -> None: assert result.exit_code == 0, result.output payload = json.loads(result.output) - assert any(action["provenance"] == "extension" for action in payload["actions"]) + 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() From c81ba58548ec30523a3e4d1e6435ed125e54c54b Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Tue, 1 Sep 2026 21:27:40 +0800 Subject: [PATCH 3/8] fix(init): make dry-run preview match real init plans Stage existing projects without --force, classify colliding artifacts as conflict, keep --json stdout parseable, report skipped already-installed artifacts, and remap in-project absolute symlinks onto the staged copy. --- src/specify_cli/commands/init.py | 171 ++++++++++++++++--- tests/test_init_dry_run.py | 273 ++++++++++++++++++++++++++++++- 2 files changed, 422 insertions(+), 22 deletions(-) diff --git a/src/specify_cli/commands/init.py b/src/specify_cli/commands/init.py index 8cba58863b..7e6460356b 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -107,6 +107,73 @@ def _preview_subprocess_env(staged_home: Path) -> dict[str, str]: return env +_INIT_PLAN_ENV = "SPECIFY_INIT_PLAN_PATH" + + +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 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 _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] @@ -312,6 +379,7 @@ def _build_preview_actions( path_prefix: str = "", ownership: dict[str, PreviewOwnership] | None = None, default_ownership: PreviewOwnership = ("integration", None), + directory_conflict: bool = False, ) -> list[dict[str, str]]: """Classify files produced by a staged initialization.""" staged_files = _snapshot_files(staged_root) @@ -327,13 +395,12 @@ def _build_preview_actions( for path in sorted(candidates): staged_digest = staged_files[path] initial_digest = initial_files.get(path) - action = ( - "create" - if initial_digest is None - else "overwrite" - if initial_digest != staged_digest - else "preserve" - ) + if 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) ) @@ -357,11 +424,8 @@ def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None console.print("\n[bold cyan]Initialization preview[/bold cyan]") if payload["conflict"]: console.print( - "[yellow]conflict[/yellow] target directory is non-empty; rerun with --force " - "to preview a forced merge" + "[yellow]conflict[/yellow] target directory exists; applying this plan requires --force" ) - return - for record in payload["actions"]: source = record["provenance"] if record.get("source_id"): @@ -369,6 +433,44 @@ def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None console.print(f"{record['action']:<10} {record['path']} [dim]({source})[/dim]") +def _remap_in_project_symlinks(project_root: Path, staged_root: Path) -> None: + """Retarget staged absolute symlinks that originally pointed inside *project_root*. + + ``copytree(..., symlinks=True)`` preserves absolute targets, so an in-tree + link keeps pointing at the live project. Remap those to the corresponding + staged path. Links that resolve outside *project_root* are left unchanged + so the initializer's containment check still rejects them. + """ + project_root = project_root.resolve() + staged_root = staged_root.resolve() + for dirpath, dirnames, filenames in os.walk(staged_root, followlinks=False): + for name in (*dirnames, *filenames): + path = Path(dirpath) / name + if not path.is_symlink(): + continue + raw_target = Path(os.fsdecode(os.readlink(path))) + if not raw_target.is_absolute(): + continue + try: + resolved_target = raw_target.resolve() + except (OSError, RuntimeError): + resolved_target = raw_target + try: + relative = resolved_target.relative_to(project_root) + except ValueError: + continue + remapped = staged_root / relative + was_dir = path.is_dir() + path.unlink() + path.symlink_to(remapped, target_is_directory=was_dir) + + +def _stage_project_copy(project_path: Path, staged_root: Path) -> None: + """Copy *project_path* into staging and remap in-project absolute symlinks.""" + shutil.copytree(project_path, staged_root, symlinks=True) + _remap_in_project_symlinks(project_path.resolve(), staged_root.resolve()) + + def _preview_init( *, project_path: Path, @@ -389,9 +491,6 @@ def _preview_init( "conflict": directory_conflict, "actions": [], } - if directory_conflict: - _emit_dry_run_preview(payload, json_output=json_output) - return initial_files = _snapshot_files(project_path) real_home = Path.home() @@ -403,7 +502,7 @@ def _preview_init( staged_home = Path(tmp_dir) / "home" staged_home.mkdir() if project_path.exists(): - shutil.copytree(project_path, staged_root, symlinks=True) + _stage_project_copy(project_path, staged_root) # Run the same public CLI path in a child process. Besides preventing # mutations of the target root, this isolates Rich's Live output from @@ -432,13 +531,16 @@ def _preview_init( 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=Path.cwd(), capture_output=True, text=True, check=False, - env=_preview_subprocess_env(staged_home), + env=env, ) if result.returncode: details = (result.stderr or result.stdout).strip().replace("\n", " ") @@ -458,6 +560,7 @@ def _preview_init( staged_root, ownership=project_ownership, default_ownership=("integration", selected_integration), + directory_conflict=directory_conflict, ) staged_home_files = _snapshot_files(staged_home) initial_home_files = _snapshot_matching_files(real_home, set(staged_home_files)) @@ -470,8 +573,12 @@ def _preview_init( path_prefix="~/", ownership=home_ownership, default_ownership=("integration", selected_integration), + directory_conflict=directory_conflict, ) ) + payload["actions"] = _merge_recorded_plan_actions( + payload["actions"], plan_path + ) for spec in url_extensions: payload["actions"].append( @@ -575,6 +682,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" @@ -592,6 +705,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" @@ -656,6 +775,11 @@ def ensure_constitution_from_template( if tracker: tracker.add("constitution", "Constitution setup") tracker.skip("constitution", "existing file preserved") + _record_init_plan_action( + "skip", + ".specify/memory/constitution.md", + "core", + ) return try: @@ -981,10 +1105,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()} @@ -1244,6 +1369,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 diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py index 032bdd05d0..fd194a4bf7 100644 --- a/tests/test_init_dry_run.py +++ b/tests/test_init_dry_run.py @@ -3,13 +3,19 @@ from __future__ import annotations import json +import os +import shutil from pathlib import Path import pytest from typer.testing import CliRunner from specify_cli import app -from specify_cli.commands.init import _snapshot_files +from specify_cli.commands.init import ( + _remap_in_project_symlinks, + _snapshot_files, + _stage_project_copy, +) _PROVENANCE_CATEGORIES = {"core", "integration", "preset", "workflow", "extension"} @@ -74,6 +80,31 @@ def test_dry_run_json_is_machine_readable_and_has_no_target_writes(tmp_path: Pat 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: @@ -109,6 +140,43 @@ def test_forced_dry_run_reports_overwrite_without_changing_existing_file( def test_non_forced_dry_run_reports_existing_target_conflict(tmp_path: Path) -> None: target = tmp_path / "nonempty-project" + command = target / ".github" / "skills" / "speckit-plan" / "SKILL.md" + command.parent.mkdir(parents=True) + command.write_text("user-owned content\n", encoding="utf-8") + 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 + actions = {(action["action"], action["path"]) for action in payload["actions"]} + assert ("conflict", ".github/skills/speckit-plan/SKILL.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 command.read_text(encoding="utf-8") == "user-owned content\n" + assert existing.read_text(encoding="utf-8") == "keep\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") @@ -131,10 +199,106 @@ def test_non_forced_dry_run_reports_existing_target_conflict(tmp_path: Path) -> assert result.exit_code == 0, result.output payload = json.loads(result.output) assert payload["conflict"] is True - assert payload["actions"] == [] + 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_leaves_url_extension_unresolved_without_creating_target( tmp_path: Path, ) -> None: @@ -374,3 +538,108 @@ def test_dry_run_isolates_and_reports_hermes_home_writes( assert hermes_action["provenance"] == "integration" assert hermes_action["source_id"] == "hermes" assert not target.exists() + + +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() + + +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 remapped == (staged / "store").resolve() + assert (staged / "link" / "file.txt").read_text(encoding="utf-8") == "ok\n" + assert Path(os.readlink(link)) == store.resolve() + + +def test_remap_leaves_external_absolute_symlinks(tmp_path: Path) -> None: + project = tmp_path / "proj" + project.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + link = project / "escape" + try: + link.symlink_to(outside.resolve()) + except (OSError, NotImplementedError): + pytest.skip("symlinks are not available") + + staged = tmp_path / "staged" + shutil.copytree(project, staged, symlinks=True) + _remap_in_project_symlinks(project.resolve(), staged.resolve()) + + assert Path(os.readlink(staged / "escape")) == outside.resolve() + + +def test_dry_run_rejects_external_absolute_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") + + with pytest.raises(RuntimeError, match="staged initialization failed"): + CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "kilocode", + "--script", + "sh", + ], + catch_exceptions=False, + ) From 5359248a6e2e9bfa6d70b364b1e061df7171b37f Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Tue, 1 Sep 2026 23:42:38 +0800 Subject: [PATCH 4/8] fix(init): keep dry-run plans accurate on Windows and chmod Quarantine external staged symlinks, strip Windows extended path prefixes, include permission bits in snapshot fingerprints, and resolve home-relative --extension specs before isolating HOME. --- src/specify_cli/commands/init.py | 173 ++++++++++++++++++----- tests/test_init_dry_run.py | 229 +++++++++++++++++++++++++++---- 2 files changed, 345 insertions(+), 57 deletions(-) diff --git a/src/specify_cli/commands/init.py b/src/specify_cli/commands/init.py index 7e6460356b..00a08628af 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -7,6 +7,7 @@ import os import shlex import shutil +import stat import subprocess import sys import tempfile @@ -56,8 +57,17 @@ 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 SHA-256 digests for regular files below *root*.""" + """Return fingerprints for regular files below *root*.""" if not root.exists(): return {} @@ -65,22 +75,28 @@ def _snapshot_files(root: Path) -> dict[str, str]: for path in root.rglob("*"): if not path.is_file() or path.is_symlink(): continue - digest = hashlib.sha256(path.read_bytes()).hexdigest() - files[path.relative_to(root).as_posix()] = digest + files[path.relative_to(root).as_posix()] = _file_fingerprint(path) return files def _snapshot_matching_files(root: Path, relative_paths: set[str]) -> dict[str, str]: - """Return digests for selected regular files below *root*.""" + """Return fingerprints for selected regular files below *root*.""" files: dict[str, str] = {} for relative_path in relative_paths: path = root / relative_path if not path.is_file() or path.is_symlink(): continue - files[relative_path] = hashlib.sha256(path.read_bytes()).hexdigest() + files[relative_path] = _file_fingerprint(path) return files +def _resolve_preview_child_extension(spec: str) -> str: + """Expand home-relative extension specs against the parent home.""" + if spec.startswith("~"): + return str(Path(spec).expanduser().resolve()) + return 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() @@ -95,6 +111,8 @@ def _preview_subprocess_env(staged_home: Path) -> dict[str, str]: "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) @@ -433,40 +451,125 @@ def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None console.print(f"{record['action']:<10} {record['path']} [dim]({source})[/dim]") +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: + _materialize_symlink(path) + return + try: + relative = os.path.relpath(os.fspath(target), os.fspath(project_root)) + except ValueError: + _materialize_symlink(path) + return + relative_path = Path(relative) + if relative_path.is_absolute() or ".." in relative_path.parts: + _materialize_symlink(path) + return + remapped = staged_root / relative_path + was_dir = path.is_dir() + path.unlink() + path.symlink_to(remapped, target_is_directory=was_dir) + + +def _materialize_symlink(path: Path) -> None: + """Replace a live external symlink with an empty directory placeholder. + + Nested targets are not copied, which avoids following a link to ``/`` or + ``$HOME``. ``mkdir`` and file writes then stay inside staging. + """ + path.unlink() + path.mkdir() + + def _remap_in_project_symlinks(project_root: Path, staged_root: Path) -> None: - """Retarget staged absolute symlinks that originally pointed inside *project_root*. + """Keep staged symlinks from pointing at the live project or external paths. - ``copytree(..., symlinks=True)`` preserves absolute targets, so an in-tree - link keeps pointing at the live project. Remap those to the corresponding - staged path. Links that resolve outside *project_root* are left unchanged - so the initializer's containment check still rejects them. + In-project absolute links are retargeted at the staged copy. Links that + resolve outside the project are replaced with placeholders so the child + initializer cannot write through them. """ - project_root = project_root.resolve() - staged_root = staged_root.resolve() - for dirpath, dirnames, filenames in os.walk(staged_root, followlinks=False): - for name in (*dirnames, *filenames): - path = Path(dirpath) / name - if not path.is_symlink(): - continue - raw_target = Path(os.fsdecode(os.readlink(path))) - if not raw_target.is_absolute(): + project_root = _normalize_fs_path(project_root) + staged_root = _normalize_fs_path(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): continue - try: - resolved_target = raw_target.resolve() - except (OSError, RuntimeError): - resolved_target = raw_target - try: - relative = resolved_target.relative_to(project_root) - except ValueError: + if resolved is not None and _is_within_root(resolved, project_root): + _remap_symlink_to_staged(path, project_root, staged_root) + changed = True continue - remapped = staged_root / relative - was_dir = path.is_dir() - path.unlink() - path.symlink_to(remapped, target_is_directory=was_dir) + _materialize_symlink(path) + changed = True + if not changed: + return + raise RuntimeError("staged symlink isolation did not converge") def _stage_project_copy(project_path: Path, staged_root: Path) -> None: - """Copy *project_path* into staging and remap in-project absolute symlinks.""" + """Copy *project_path* into staging and isolate live symlinks.""" shutil.copytree(project_path, staged_root, symlinks=True) _remap_in_project_symlinks(project_path.resolve(), staged_root.resolve()) @@ -527,7 +630,9 @@ def _preview_init( if preset: command.extend(["--preset", preset]) for extension in staged_extensions: - command.extend(["--extension", extension]) + command.extend( + ["--extension", _resolve_preview_child_extension(extension)] + ) if trust_extension_urls: command.append("--trust-extension-urls") @@ -539,11 +644,13 @@ def _preview_init( cwd=Path.cwd(), capture_output=True, text=True, + encoding="utf-8", + errors="replace", check=False, env=env, ) if result.returncode: - details = (result.stderr or result.stdout).strip().replace("\n", " ") + details = (result.stderr or result.stdout or "").strip().replace("\n", " ") raise RuntimeError(f"staged initialization failed: {details[:240]}") registry_sources = _preview_registry_sources(staged_root) diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py index fd194a4bf7..70925c847c 100644 --- a/tests/test_init_dry_run.py +++ b/tests/test_init_dry_run.py @@ -11,10 +11,13 @@ from typer.testing import CliRunner from specify_cli import app +from specify_cli._assets import _locate_bundled_extension from specify_cli.commands.init import ( - _remap_in_project_symlinks, + _is_within_root, + _normalize_fs_path, _snapshot_files, _stage_project_copy, + _strip_windows_extended_prefix, ) _PROVENANCE_CATEGORIES = {"core", "integration", "preset", "workflow", "extension"} @@ -24,6 +27,46 @@ 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) + + +@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" @@ -540,6 +583,42 @@ def test_dry_run_isolates_and_reports_hermes_home_writes( assert not target.exists() +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 == 0, result.output + json.loads(result.output) + 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" @@ -593,16 +672,17 @@ def test_remap_in_project_absolute_symlinks_points_at_staged_copy( _stage_project_copy(project, staged) remapped = Path(os.readlink(staged / "link")) - assert remapped == (staged / "store").resolve() + _assert_same_path(remapped, staged / "store") assert (staged / "link" / "file.txt").read_text(encoding="utf-8") == "ok\n" - assert Path(os.readlink(link)) == store.resolve() + _assert_same_path(Path(os.readlink(link)), store) -def test_remap_leaves_external_absolute_symlinks(tmp_path: Path) -> None: +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()) @@ -610,13 +690,19 @@ def test_remap_leaves_external_absolute_symlinks(tmp_path: Path) -> None: pytest.skip("symlinks are not available") staged = tmp_path / "staged" - shutil.copytree(project, staged, symlinks=True) - _remap_in_project_symlinks(project.resolve(), staged.resolve()) + _stage_project_copy(project, staged) - assert Path(os.readlink(staged / "escape")) == outside.resolve() + staged_escape = staged / "escape" + assert not staged_escape.is_symlink() + (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_rejects_external_absolute_command_symlink(tmp_path: Path) -> None: +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() @@ -627,19 +713,114 @@ def test_dry_run_rejects_external_absolute_command_symlink(tmp_path: Path) -> No except (OSError, NotImplementedError): pytest.skip("symlinks are not available") - with pytest.raises(RuntimeError, match="staged initialization failed"): - CliRunner().invoke( - app, - [ - "init", - str(target), - "--force", - "--dry-run", - "--json", - "--integration", - "kilocode", - "--script", - "sh", - ], - catch_exceptions=False, - ) + 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 payload["actions"] + assert list(outside.iterdir()) == [] + 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() From a2acf27c394fbf439d19a138749710ee6f756b24 Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Wed, 2 Sep 2026 09:24:20 +0800 Subject: [PATCH 5/8] fix(init): keep dry-run escape checks for external symlinks Retarget external staged links at an isolated dummy outside the project copy so setup() still rejects destinations that leave the tree, without writing through to the live target. --- src/specify_cli/commands/init.py | 150 ++++++++++++++++++++----------- tests/test_init_dry_run.py | 16 +++- 2 files changed, 109 insertions(+), 57 deletions(-) diff --git a/src/specify_cli/commands/init.py b/src/specify_cli/commands/init.py index 00a08628af..78c7006924 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -444,6 +444,8 @@ def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None console.print( "[yellow]conflict[/yellow] target directory exists; applying this plan requires --force" ) + if payload.get("error"): + console.print(f"[red]failed[/red] {payload['error']}") for record in payload["actions"]: source = record["provenance"] if record.get("source_id"): @@ -515,16 +517,16 @@ def _remap_symlink_to_staged( ) -> None: target = _symlink_target(path) if target is None: - _materialize_symlink(path) + _quarantine_symlink(path, staged_root) return try: relative = os.path.relpath(os.fspath(target), os.fspath(project_root)) except ValueError: - _materialize_symlink(path) + _quarantine_symlink(path, staged_root) return relative_path = Path(relative) if relative_path.is_absolute() or ".." in relative_path.parts: - _materialize_symlink(path) + _quarantine_symlink(path, staged_root) return remapped = staged_root / relative_path was_dir = path.is_dir() @@ -532,36 +534,53 @@ def _remap_symlink_to_staged( path.symlink_to(remapped, target_is_directory=was_dir) -def _materialize_symlink(path: Path) -> None: - """Replace a live external symlink with an empty directory placeholder. +def _quarantine_root(staged_root: Path) -> Path: + return _normalize_fs_path(staged_root.parent / "quarantine") - Nested targets are not copied, which avoids following a link to ``/`` or - ``$HOME``. ``mkdir`` and file writes then stay inside staging. + +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() - path.mkdir() + 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 replaced with placeholders so the child - initializer cannot write through them. + 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): + 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 - _materialize_symlink(path) + _quarantine_symlink(path, staged_root) changed = True if not changed: return @@ -574,6 +593,23 @@ def _stage_project_copy(project_path: Path, staged_root: Path) -> None: _remap_in_project_symlinks(project_path.resolve(), staged_root.resolve()) +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 + ) + 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, @@ -650,54 +686,60 @@ def _preview_init( env=env, ) if result.returncode: - details = (result.stderr or result.stdout or "").strip().replace("\n", " ") - raise RuntimeError(f"staged initialization failed: {details[:240]}") - - registry_sources = _preview_registry_sources(staged_root) - project_ownership = _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) - ) - payload["actions"] = _build_preview_actions( - initial_files, - staged_root, - ownership=project_ownership, - default_ownership=("integration", selected_integration), - directory_conflict=directory_conflict, - ) - staged_home_files = _snapshot_files(staged_home) - initial_home_files = _snapshot_matching_files(real_home, set(staged_home_files)) - home_ownership = registry_home_ownership - home_ownership.update(_preview_marker_ownership(staged_home, registry_sources)) - payload["actions"].extend( - _build_preview_actions( - initial_home_files, - staged_home, - path_prefix="~/", - ownership=home_ownership, + payload["error"] = _preview_child_failure_message(result) + else: + registry_sources = _preview_registry_sources(staged_root) + project_ownership = _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) + ) + payload["actions"] = _build_preview_actions( + initial_files, + staged_root, + ownership=project_ownership, default_ownership=("integration", selected_integration), directory_conflict=directory_conflict, ) - ) - payload["actions"] = _merge_recorded_plan_actions( - payload["actions"], plan_path - ) + staged_home_files = _snapshot_files(staged_home) + initial_home_files = _snapshot_matching_files( + real_home, set(staged_home_files) + ) + home_ownership = registry_home_ownership + home_ownership.update( + _preview_marker_ownership(staged_home, registry_sources) + ) + payload["actions"].extend( + _build_preview_actions( + initial_home_files, + staged_home, + path_prefix="~/", + ownership=home_ownership, + default_ownership=("integration", selected_integration), + directory_conflict=directory_conflict, + ) + ) + payload["actions"] = _merge_recorded_plan_actions( + payload["actions"], plan_path + ) - for spec in url_extensions: - payload["actions"].append( - { - "action": "unresolved", - "path": spec, - "provenance": "extension", - "source_id": spec, - } - ) + if not payload.get("error"): + for spec in url_extensions: + payload["actions"].append( + { + "action": "unresolved", + "path": spec, + "provenance": "extension", + "source_id": spec, + } + ) 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( diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py index 70925c847c..f5325a6ac1 100644 --- a/tests/test_init_dry_run.py +++ b/tests/test_init_dry_run.py @@ -693,7 +693,12 @@ def test_stage_copy_quarantines_external_absolute_symlinks(tmp_path: Path) -> No _stage_project_copy(project, staged) staged_escape = staged / "escape" - assert not staged_escape.is_symlink() + 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" @@ -729,10 +734,15 @@ def test_dry_run_does_not_write_through_external_command_symlink( catch_exceptions=False, ) - assert result.exit_code == 0, result.output + assert result.exit_code == 1, result.output payload = json.loads(result.output) - assert payload["actions"] + 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() From 3883a6edabcc44e2e35fb3a124d50b0d3c64320b Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Thu, 3 Sep 2026 08:44:59 +0800 Subject: [PATCH 6/8] fix(init): address dry-run review feedback - Validate typed marker source registration and provenance categories - Normalize Rich panel errors across narrow terminal layouts - Add regression coverage for ownership markers and wrapped errors --- src/specify_cli/commands/init.py | 9 +++++-- tests/test_init_dry_run.py | 40 ++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/src/specify_cli/commands/init.py b/src/specify_cli/commands/init.py index 78c7006924..346b2e0977 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -330,13 +330,13 @@ def _preview_content_ownership( if value.startswith(("preset:", "extension:")): category, source_id = value.split(":", 1) source_id = source_id.split(":", 1)[0] - if source_id: + 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 source_id: + if category in registry_sources.get(source_id, set()): return category, source_id if value.startswith("Source:"): bare_source_id = value.removeprefix("Source:").strip() @@ -600,6 +600,11 @@ def _preview_child_failure_message(result: subprocess.CompletedProcess[str]) -> 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) :] diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py index f5325a6ac1..327d6ef7c4 100644 --- a/tests/test_init_dry_run.py +++ b/tests/test_init_dry_run.py @@ -5,6 +5,7 @@ import json import os import shutil +import subprocess from pathlib import Path import pytest @@ -15,6 +16,8 @@ from specify_cli.commands.init import ( _is_within_root, _normalize_fs_path, + _preview_child_failure_message, + _preview_content_ownership, _snapshot_files, _stage_project_copy, _strip_windows_extended_prefix, @@ -36,6 +39,43 @@ def _assert_same_path(left: Path | str, right: Path | str) -> None: assert _normalize_fs_path(left_path) == _normalize_fs_path(right_path) +@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) + + @pytest.mark.parametrize( ("raw", "expected"), [ From 3e4cb86b47433230c092050f256d0ea7ce61bf04 Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Thu, 3 Sep 2026 21:18:04 +0800 Subject: [PATCH 7/8] fix(init): preserve dry-run semantics and report component failures - Preserve the caller's force mode in staged previews - Report optional preset and extension failures in JSON and human output --- src/specify_cli/commands/init.py | 99 ++++++++++++++++++++-- tests/test_init_dry_run.py | 138 +++++++++++++++++++++++++++---- 2 files changed, 216 insertions(+), 21 deletions(-) diff --git a/src/specify_cli/commands/init.py b/src/specify_cli/commands/init.py index 346b2e0977..0865cab4dd 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -126,6 +126,14 @@ def _preview_subprocess_env(staged_home: Path) -> dict[str, str]: _INIT_PLAN_ENV = "SPECIFY_INIT_PLAN_PATH" +_INIT_STAGING_CONFIRMATION_ENV = "SPECIFY_INIT_STAGING_CONFIRMATION" + + +def _staging_confirmation_is_accepted() -> bool: + """Return whether a staged preview child may skip its directory prompt.""" + return bool(os.environ.get(_INIT_PLAN_ENV)) and ( + os.environ.get(_INIT_STAGING_CONFIRMATION_ENV) == "1" + ) def _record_init_plan_action( @@ -152,6 +160,55 @@ def _record_init_plan_action( 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 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]]: @@ -446,6 +503,11 @@ def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None ) if payload.get("error"): console.print(f"[red]failed[/red] {payload['error']}") + for failure in payload["failures"]: + console.print( + f"failed {failure['component']}:{failure['source_id']} " + f"{failure['error']}" + ) for record in payload["actions"]: source = record["provenance"] if record.get("source_id"): @@ -619,6 +681,7 @@ def _preview_init( *, project_path: Path, directory_conflict: bool, + force: bool, script_type: str, selected_integration: str, ignore_agent_tools: bool, @@ -634,6 +697,7 @@ def _preview_init( "target": str(project_path), "conflict": directory_conflict, "actions": [], + "failures": [], } initial_files = _snapshot_files(project_path) @@ -648,22 +712,24 @@ def _preview_init( if project_path.exists(): _stage_project_copy(project_path, staged_root) - # Run the same public CLI path in a child process. Besides preventing + # 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 preview's human/JSON output contract. The staging-only + # confirmation signal avoids a prompt without changing force mode. command = [ sys.executable, "-c", "from specify_cli import main; main()", "init", str(staged_root), - "--force", "--non-interactive", "--integration", selected_integration, "--script", script_type, ] + if force: + command.append("--force") if ignore_agent_tools: command.append("--ignore-agent-tools") if integration_options: @@ -680,6 +746,7 @@ def _preview_init( plan_path = Path(tmp_dir) / "init-plan.jsonl" env = _preview_subprocess_env(staged_home) env[_INIT_PLAN_ENV] = str(plan_path) + env[_INIT_STAGING_CONFIRMATION_ENV] = "1" result = subprocess.run( command, cwd=Path.cwd(), @@ -730,6 +797,7 @@ def _preview_init( 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: @@ -1150,6 +1218,7 @@ def init( 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() @@ -1172,14 +1241,14 @@ def init( console.print( "[cyan]--force supplied: skipping confirmation and proceeding with merge[/cyan]" ) - elif non_interactive: + 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 @@ -1237,7 +1306,7 @@ def init( console.print( f"[cyan]--force supplied: merging into existing directory '[cyan]{safe_name}[/cyan]'[/cyan]" ) - else: + 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" @@ -1354,6 +1423,7 @@ def init( _preview_init( project_path=project_path, directory_conflict=directory_conflict, + force=force, script_type=selected_script, selected_integration=selected_ai, ignore_agent_tools=ignore_agent_tools, @@ -1597,6 +1667,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." ) @@ -1605,6 +1680,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." @@ -1623,6 +1703,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", @@ -1637,6 +1720,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", @@ -1672,6 +1756,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/tests/test_init_dry_run.py b/tests/test_init_dry_run.py index 327d6ef7c4..7a8827a6ad 100644 --- a/tests/test_init_dry_run.py +++ b/tests/test_init_dry_run.py @@ -192,9 +192,9 @@ def test_forced_dry_run_reports_overwrite_without_changing_existing_file( tmp_path: Path, ) -> None: target = tmp_path / "existing-project" - command = target / ".github" / "skills" / "speckit-plan" / "SKILL.md" - command.parent.mkdir(parents=True) - command.write_text("user-owned content\n", encoding="utf-8") + 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, @@ -214,26 +214,28 @@ def test_forced_dry_run_reports_overwrite_without_changing_existing_file( assert result.exit_code == 0, result.output payload = json.loads(result.output) - assert { - (action["action"], action["path"]) - for action in payload["actions"] - } >= {("overwrite", ".github/skills/speckit-plan/SKILL.md")} - assert command.read_text(encoding="utf-8") == "user-owned content\n" + 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_dry_run_reports_existing_target_conflict(tmp_path: Path) -> None: +def test_non_forced_here_dry_run_matches_confirmed_preserve_behavior( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: target = tmp_path / "nonempty-project" - command = target / ".github" / "skills" / "speckit-plan" / "SKILL.md" - command.parent.mkdir(parents=True) - command.write_text("user-owned content\n", encoding="utf-8") + 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", - str(target), + "--here", "--dry-run", "--json", "--integration", @@ -248,13 +250,30 @@ def test_non_forced_dry_run_reports_existing_target_conflict(tmp_path: Path) -> payload = json.loads(result.output) assert payload["conflict"] is True actions = {(action["action"], action["path"]) for action in payload["actions"]} - assert ("conflict", ".github/skills/speckit-plan/SKILL.md") in 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 command.read_text(encoding="utf-8") == "user-owned content\n" + 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, @@ -417,6 +436,95 @@ def test_dry_run_leaves_url_extension_unresolved_without_creating_target( 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" From 09b122c4ca2f8b316570c19c35e8314dfb7fa9d4 Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Thu, 3 Sep 2026 21:31:05 +0800 Subject: [PATCH 8/8] fix(init): align dry-run preview with init behavior - Preserve here and force semantics in staged previews - Surface core failures and URL resolution limits in preview output --- src/specify_cli/__init__.py | 7 ++- src/specify_cli/commands/init.py | 83 ++++++++++++++++++++++++-------- tests/test_init_dry_run.py | 3 +- 3 files changed, 70 insertions(+), 23 deletions(-) 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/commands/init.py b/src/specify_cli/commands/init.py index 0865cab4dd..80b73a63c6 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -90,9 +90,9 @@ def _snapshot_matching_files(root: Path, relative_paths: set[str]) -> dict[str, return files -def _resolve_preview_child_extension(spec: str) -> str: - """Expand home-relative extension specs against the parent home.""" - if spec.startswith("~"): +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 @@ -125,6 +125,16 @@ def _preview_subprocess_env(staged_home: Path) -> dict[str, str]: return env +def _seed_preview_home(staged_home: Path, real_home: Path) -> None: + """Copy the read-only catalog settings the initializer resolves from HOME.""" + for filename in ("extension-catalogs.yml", "preset-catalogs.yml"): + source = real_home / ".specify" / filename + if source.is_file() and not source.is_symlink(): + destination = staged_home / ".specify" / filename + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + + _INIT_PLAN_ENV = "SPECIFY_INIT_PLAN_PATH" _INIT_STAGING_CONFIRMATION_ENV = "SPECIFY_INIT_STAGING_CONFIRMATION" @@ -497,10 +507,12 @@ def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None return console.print("\n[bold cyan]Initialization preview[/bold cyan]") - if payload["conflict"]: + 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] {payload['error']}") for failure in payload["failures"]: @@ -651,7 +663,18 @@ def _remap_in_project_symlinks(project_root: Path, staged_root: Path) -> None: def _stage_project_copy(project_path: Path, staged_root: Path) -> None: """Copy *project_path* into staging and isolate live symlinks.""" - shutil.copytree(project_path, staged_root, symlinks=True) + def ignore_special_files(directory: str, names: list[str]) -> set[str]: + 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 + + shutil.copytree(project_path, staged_root, symlinks=True, ignore=ignore_special_files) _remap_in_project_symlinks(project_path.resolve(), staged_root.resolve()) @@ -680,8 +703,9 @@ def _preview_child_failure_message(result: subprocess.CompletedProcess[str]) -> def _preview_init( *, project_path: Path, - directory_conflict: bool, + gate: str, force: bool, + here: bool, script_type: str, selected_integration: str, ignore_agent_tools: bool, @@ -695,7 +719,8 @@ def _preview_init( payload: dict[str, Any] = { "dry_run": True, "target": str(project_path), - "conflict": directory_conflict, + "conflict": gate != "none", + "gate": gate, "actions": [], "failures": [], } @@ -709,6 +734,7 @@ def _preview_init( staged_root = Path(tmp_dir) / "project" staged_home = Path(tmp_dir) / "home" staged_home.mkdir() + _seed_preview_home(staged_home, real_home) if project_path.exists(): _stage_project_copy(project_path, staged_root) @@ -721,25 +747,25 @@ def _preview_init( "-c", "from specify_cli import main; main()", "init", - str(staged_root), "--non-interactive", "--integration", selected_integration, "--script", script_type, ] - if force: + if here: + command.append("--here") + else: + command.append(str(staged_root)) + if force or gate == "force_required": command.append("--force") - if ignore_agent_tools: - command.append("--ignore-agent-tools") + command.append("--ignore-agent-tools") if integration_options: command.extend(["--integration-options", integration_options]) if preset: - command.extend(["--preset", preset]) + command.extend(["--preset", _resolve_preview_child_path(preset)]) for extension in staged_extensions: - command.extend( - ["--extension", _resolve_preview_child_extension(extension)] - ) + command.extend(["--extension", _resolve_preview_child_path(extension)]) if trust_extension_urls: command.append("--trust-extension-urls") @@ -749,7 +775,7 @@ def _preview_init( env[_INIT_STAGING_CONFIRMATION_ENV] = "1" result = subprocess.run( command, - cwd=Path.cwd(), + cwd=staged_root if here else Path.cwd(), capture_output=True, text=True, encoding="utf-8", @@ -774,7 +800,7 @@ def _preview_init( staged_root, ownership=project_ownership, default_ownership=("integration", selected_integration), - directory_conflict=directory_conflict, + directory_conflict=False, ) staged_home_files = _snapshot_files(staged_home) initial_home_files = _snapshot_matching_files( @@ -791,7 +817,7 @@ def _preview_init( path_prefix="~/", ownership=home_ownership, default_ownership=("integration", selected_integration), - directory_conflict=directory_conflict, + directory_conflict=False, ) ) payload["actions"] = _merge_recorded_plan_actions( @@ -807,6 +833,7 @@ def _preview_init( "path": spec, "provenance": "extension", "source_id": spec, + "reason": "URL extensions are not fetched during dry-run", } ) payload["actions"].sort(key=lambda action: action["path"]) @@ -1012,6 +1039,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") @@ -1029,6 +1059,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: @@ -1420,10 +1451,18 @@ def init( 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, - directory_conflict=directory_conflict, + gate=gate, force=force, + here=here, script_type=selected_script, selected_integration=selected_ai, ignore_agent_tools=ignore_agent_tools, @@ -1627,6 +1666,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 = { @@ -1643,7 +1683,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: diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py index 7a8827a6ad..2c4ef8d7ff 100644 --- a/tests/test_init_dry_run.py +++ b/tests/test_init_dry_run.py @@ -336,7 +336,7 @@ def test_non_forced_dry_run_human_preview_lists_conflicting_artifacts( 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") + line.startswith("overwrite .github/skills/speckit-plan/SKILL.md") for line in lines ) assert any( @@ -432,6 +432,7 @@ def test_dry_run_leaves_url_extension_unresolved_without_creating_target( "path": extension_url, "provenance": "extension", "source_id": extension_url, + "reason": "URL extensions are not fetched during dry-run", } assert not target.exists()