diff --git a/docs/reference/core.md b/docs/reference/core.md index afd53c46d4..ad017a17b5 100644 --- a/docs/reference/core.md +++ b/docs/reference/core.md @@ -15,8 +15,12 @@ specify init [] | `--script sh\|ps\|py` | Script type: `sh` (bash/zsh), `ps` (PowerShell), or `py` (Python) | | `--here` | Initialize in the current directory instead of creating a new one | | `--force` | Force merge/overwrite when initializing in an existing directory | +| `--non-interactive` | Never prompt; use safe documented defaults where available | +| `--json` | Output machine-readable JSON; implies non-interactive behavior | | `--ignore-agent-tools` | Skip checks for AI coding agent CLI tools | | `--preset ` | Install a preset during initialization | +| `--extension ` | Install an optional bundled, local, catalog, or URL extension; repeatable | +| `--trust-extension-urls` | Explicitly authorize URL extension installs without a trust prompt | Creates a new Spec Kit project with the necessary directory structure, templates, scripts, and AI coding agent integration files. @@ -44,8 +48,18 @@ specify init my-project --integration copilot --script ps # Install a preset during initialization specify init my-project --integration copilot --preset compliance + +# Initialize non-interactively with machine-readable output +specify init my-project --json ``` +### JSON output + +`--json` outputs machine-readable JSON and always runs non-interactively. It +uses the same initializer as human mode, so it changes output rather than the +generated project. It does not imply `--force`, `--trust-extension-urls`, +`--ignore-agent-tools`, or overwrite/trust consent. + ### Environment Variables | Variable | Description | diff --git a/src/specify_cli/_command_init_json.py b/src/specify_cli/_command_init_json.py new file mode 100644 index 0000000000..d0730f34b1 --- /dev/null +++ b/src/specify_cli/_command_init_json.py @@ -0,0 +1,447 @@ +"""Machine-readable adapter for the existing ``specify init`` execution path.""" + +from __future__ import annotations + +import io +import json +import os +import sys +from contextlib import redirect_stderr, redirect_stdout +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, NoReturn + +import typer +from typer.core import TyperCommand +from typer.exceptions import TyperException + +from ._agent_config import ( + AGENT_CONFIG, + DEFAULT_INIT_INTEGRATION, + DEFAULT_INIT_INTEGRATION_ENV_VAR, + SCRIPT_TYPE_CHOICES, +) +from ._console import console, err_console +from ._utils import check_tool + + +@dataclass +class InitJsonFailure(Exception): + code: str + message: str + details: dict[str, Any] = field(default_factory=dict) + + +class InitJsonCommand(TyperCommand): + """Keep JSON-mode parser failures machine-readable.""" + + def make_context(self, info_name, args, parent=None, **extra): + end_of_options = args.index("--") if "--" in args else len(args) + json_output = "--json" in args[:end_of_options] + try: + return super().make_context(info_name, args, parent=parent, **extra) + except TyperException as exc: + if json_output: + _emit_failure( + InitJsonFailure( + "invalid_arguments", + _single_line(exc), + ) + ) + raise + + +def _single_line(value: object, *, limit: int = 500) -> str: + text = " ".join(str(value).split()) + return (text or value.__class__.__name__)[:limit] + + +def _emit(value: dict[str, Any], *, error: bool = False) -> None: + stream = sys.stderr if error else sys.stdout + payload = ( + json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + "\n" + ).encode("utf-8") + binary_stream = getattr(stream, "buffer", None) + if binary_stream is None: + stream.write(payload.decode("utf-8")) + stream.flush() + return + binary_stream.write(payload) + binary_stream.flush() + + +def _emit_failure(failure: InitJsonFailure) -> NoReturn: + _emit( + { + "error": { + "code": failure.code, + "message": failure.message, + "details": failure.details, + } + }, + error=True, + ) + raise typer.Exit(1) + + +def _extension_url(value: str) -> bool: + from urllib.parse import urlparse + + try: + return urlparse(value).scheme in {"http", "https"} + except ValueError: + return False + + +def _preflight( + *, + project_name: str | None, + script_type: str | None, + ignore_agent_tools: bool, + here: bool, + force: bool, + integration_key: str | None, + integration_options: str | None, + extensions: list[str] | None, + trust_extension_urls: bool, +) -> dict[str, Any]: + from ._download_security import is_https_or_localhost_http + from .command_init import ( + _InitIntegrationOptionsError, + _validate_init_integration_options, + ) + from .integrations import get_integration + from .integrations._commands import ( + _IntegrationOptionsError, + _parse_integration_options, + ) + + if project_name == ".": + here = True + project_name = None + if here and project_name: + raise InitJsonFailure( + "conflicting_target_options", + "Cannot specify both a project name and --here.", + {"project_name": project_name}, + ) + if not here and not project_name: + raise InitJsonFailure( + "target_required", + "Specify a project name, use '.', or pass --here.", + ) + + project_path = Path.cwd() if here else Path(project_name or "").resolve() + existed_before = project_path.exists() + if existed_before: + if not project_path.is_dir(): + raise InitJsonFailure( + "target_not_directory", + "The target exists but is not a directory.", + {"path": str(project_path)}, + ) + try: + existing_items = list(project_path.iterdir()) + except OSError as exc: + raise InitJsonFailure( + "target_unavailable", + "The target directory could not be inspected.", + {"path": str(project_path), "reason": _single_line(exc)}, + ) from exc + if not force: + if here and existing_items: + raise InitJsonFailure( + "target_not_empty", + "The current directory is not empty; pass --force to merge into it.", + {"path": str(project_path), "item_count": len(existing_items)}, + ) + if not here: + raise InitJsonFailure( + "target_exists", + "The target directory already exists; pass --force to merge into it.", + {"path": str(project_path), "item_count": len(existing_items)}, + ) + + requested_extensions = list(extensions or []) + unsupported_urls = [ + extension + for extension in requested_extensions + if _extension_url(extension) + and not is_https_or_localhost_http(extension) + ] + if unsupported_urls: + raise InitJsonFailure( + "invalid_extension_url", + "Extension URLs must use HTTPS; HTTP is allowed only for localhost.", + {"extensions": unsupported_urls}, + ) + untrusted_urls = [ + extension + for extension in requested_extensions + if _extension_url(extension) and not trust_extension_urls + ] + if untrusted_urls: + raise InitJsonFailure( + "extension_url_trust_required", + "External extension URLs require explicit trust before initialization.", + { + "extensions": untrusted_urls, + "required_flag": "--trust-extension-urls", + }, + ) + + warnings: list[dict[str, Any]] = [] + integration_defaulted = not integration_key + if integration_key: + selected_integration = integration_key + else: + override = (os.environ.get(DEFAULT_INIT_INTEGRATION_ENV_VAR) or "").strip() + if override and override not in AGENT_CONFIG: + warnings.append( + { + "code": "invalid_default_integration", + "message": ( + "The configured default integration was not recognized; " + "the built-in default was used." + ), + "details": { + "environment_variable": DEFAULT_INIT_INTEGRATION_ENV_VAR, + "value": override, + "default": DEFAULT_INIT_INTEGRATION, + }, + } + ) + selected_integration = DEFAULT_INIT_INTEGRATION + else: + selected_integration = override or DEFAULT_INIT_INTEGRATION + + integration = get_integration(selected_integration) + if integration is None or selected_integration not in AGENT_CONFIG: + raise InitJsonFailure( + "invalid_integration", + "The requested integration is not registered.", + { + "integration": selected_integration, + "available": sorted(AGENT_CONFIG), + }, + ) + if selected_integration == "generic" and not integration_options: + raise InitJsonFailure( + "invalid_integration_options", + "The generic integration requires --commands-dir.", + {"integration": "generic"}, + ) + + try: + parsed_options = ( + _parse_integration_options(integration, integration_options) + if integration_options + else {} + ) + integration.is_skills_mode( + parsed_options or None, + project_root=project_path, + ) + _validate_init_integration_options( + project_path, + integration, + parsed_options or None, + integration_options, + ) + except _IntegrationOptionsError as exc: + raise InitJsonFailure( + "invalid_integration_options", + "Integration options are invalid.", + { + "integration": selected_integration, + "reason": exc.diagnostic, + }, + ) from exc + except _InitIntegrationOptionsError as exc: + raise InitJsonFailure( + "invalid_integration_options", + exc.message, + { + "integration": selected_integration, + **exc.details, + }, + ) from exc + except (ValueError, typer.Exit) as exc: + raise InitJsonFailure( + "invalid_integration_options", + "Integration options are invalid.", + { + "integration": selected_integration, + "reason": _single_line(exc), + }, + ) from exc + + if not ignore_agent_tools: + agent_config = AGENT_CONFIG[selected_integration] + if agent_config.get("requires_cli") and not check_tool(selected_integration): + raise InitJsonFailure( + "missing_agent_tool", + "The selected integration requires an agent tool that was not found.", + { + "integration": selected_integration, + "install_url": agent_config.get("install_url"), + "override_flag": "--ignore-agent-tools", + }, + ) + + script_defaulted = not script_type + selected_script = script_type or ("ps" if os.name == "nt" else "sh") + if selected_script not in SCRIPT_TYPE_CHOICES: + raise InitJsonFailure( + "invalid_script_type", + "The requested script type is not supported.", + { + "script_type": selected_script, + "available": sorted(SCRIPT_TYPE_CHOICES), + }, + ) + + return { + "project_path": project_path, + "integration_defaulted": integration_defaulted, + "script_defaulted": script_defaulted, + "warnings": warnings, + } + + +def run_init_json( + *, + execute: Callable[[], dict[str, Any]], + project_name: str | None, + script_type: str | None, + ignore_agent_tools: bool, + here: bool, + force: bool, + integration: str | None, + integration_options: str | None, + extensions: list[str] | None, + trust_extension_urls: bool, +) -> None: + """Invoke the regular initializer and replace only its output rendering.""" + from .command_init import ( + _InitRollbackError, + _InitTargetClaimError, + _init_failure_context, + _init_json_mode, + _init_rollback_context, + ) + + result: dict[str, Any] | None = None + failure: InitJsonFailure | None = None + context: dict[str, Any] | None = None + stdout_capture = io.StringIO() + stderr_capture = io.StringIO() + failure_token = _init_failure_context.set(None) + rollback_token = _init_rollback_context.set(None) + json_mode_token = _init_json_mode.set(True) + try: + with ( + redirect_stdout(stdout_capture), + redirect_stderr(stderr_capture), + console.capture(), + err_console.capture(), + ): + context = _preflight( + project_name=project_name, + script_type=script_type, + ignore_agent_tools=ignore_agent_tools, + here=here, + force=force, + integration_key=integration, + integration_options=integration_options, + extensions=extensions, + trust_extension_urls=trust_extension_urls, + ) + result = execute() + except InitJsonFailure as exc: + failure = exc + except (typer.Exit, SystemExit) as exc: + internal_failure = _init_failure_context.get() + rollback = _init_rollback_context.get() + project_path = context["project_path"] if context else None + if isinstance(internal_failure, _InitTargetClaimError): + failure = InitJsonFailure( + internal_failure.code, + internal_failure.message, + internal_failure.details, + ) + elif isinstance(internal_failure, _InitRollbackError): + failure = InitJsonFailure( + "rollback_failed", + "Initialization failed and the newly created target could not be removed.", + { + "path": str(project_path), + "original_error": { + "code": "initialization_failed", + "message": "Project initialization failed.", + "details": {}, + }, + "cleanup_error": internal_failure.details, + }, + ) + elif internal_failure is not None and not isinstance( + internal_failure, + (OSError, ValueError), + ): + failure = InitJsonFailure( + "internal_error", + "Project initialization failed because of an unexpected internal error.", + {"exception_type": internal_failure.__class__.__name__}, + ) + else: + details: dict[str, Any] = {} + if rollback is not None and rollback.get("status") == "completed": + details["rollback"] = rollback + failure = InitJsonFailure( + "initialization_failed", + "Project initialization failed.", + details, + ) + exit_code = ( + exc.exit_code + if isinstance(exc, typer.Exit) + else exc.code + if isinstance(exc.code, int) + else 1 + ) + if exit_code == 0: + failure = InitJsonFailure( + "initialization_failed", + "Project initialization did not complete.", + ) + except Exception as exc: # noqa: BLE001 - sanitize the JSON boundary + failure = InitJsonFailure( + "internal_error", + "Project initialization failed because of an unexpected internal error.", + {"exception_type": exc.__class__.__name__}, + ) + finally: + _init_failure_context.reset(failure_token) + _init_rollback_context.reset(rollback_token) + _init_json_mode.reset(json_mode_token) + + if failure is not None: + _emit_failure(failure) + if result is None or context is None: + _emit_failure( + InitJsonFailure( + "internal_error", + "Project initialization did not produce a result.", + ) + ) + + result["integration"]["defaulted"] = context["integration_defaulted"] + result["script"]["defaulted"] = context["script_defaulted"] + result["warnings"] = [*context["warnings"], *result["warnings"]] + _emit(result) diff --git a/src/specify_cli/bundles/_commands.py b/src/specify_cli/bundles/_commands.py index 5619ecc04b..df335be909 100644 --- a/src/specify_cli/bundles/_commands.py +++ b/src/specify_cli/bundles/_commands.py @@ -96,6 +96,8 @@ def _run_init(integration: str, *, script_type: str, offline: bool = False) -> N ignore_agent_tools=True, here=True, force=True, + non_interactive=True, + json_output=False, skip_tls=False, debug=False, github_token=None, diff --git a/src/specify_cli/command_init.py b/src/specify_cli/command_init.py index c2630f0ef8..89918dfb27 100644 --- a/src/specify_cli/command_init.py +++ b/src/specify_cli/command_init.py @@ -5,8 +5,13 @@ import os import shlex import shutil +import stat import subprocess import sys +import tempfile +from contextlib import nullcontext +from contextvars import ContextVar +from dataclasses import dataclass from pathlib import Path from typing import Any @@ -29,6 +34,434 @@ from ._utils import check_tool +_init_failure_context: ContextVar[BaseException | None] = ContextVar( + "init_failure", + default=None, +) +_init_rollback_context: ContextVar[dict[str, Any] | None] = ContextVar( + "init_rollback", + default=None, +) +_init_json_mode: ContextVar[bool] = ContextVar("init_json_mode", default=False) + + +class _InitRollbackError(Exception): + def __init__(self, details: dict[str, Any]): + super().__init__(details["code"]) + self.details = details + + +class _InitTargetClaimError(Exception): + def __init__(self, code: str, message: str, details: dict[str, Any]): + super().__init__(message) + self.code = code + self.message = message + self.details = details + + +class _InitIntegrationOptionsError(Exception): + def __init__(self, message: str, details: dict[str, Any]): + super().__init__(message) + self.message = message + self.details = details + + +def _validate_init_integration_options( + project_path: Path, + integration: Any, + parsed_options: dict[str, Any] | None, + raw_options: str | None, +) -> None: + """Reject unsafe integration destinations and in-place layout changes.""" + from .integration_state import ( + default_integration_key, + installed_integration_keys, + integration_setting, + try_read_integration_json, + ) + + project_root = project_path.resolve() + requested_destination = None + if integration.key == "generic": + try: + commands_dir = integration._resolve_commands_dir( + parsed_options, + {"raw_options": raw_options}, + ) + except ValueError as exc: + raise _InitIntegrationOptionsError( + "The generic integration options are invalid.", + {"reason": str(exc)}, + ) from exc + requested_destination = (project_path / commands_dir).resolve() + if not requested_destination.is_relative_to(project_root): + raise _InitIntegrationOptionsError( + "The generic integration command directory must be inside the project.", + { + "destination": str(requested_destination), + "project_path": str(project_root), + }, + ) + + state, state_error = try_read_integration_json(project_path) + if state_error is not None: + raise _InitIntegrationOptionsError( + "The existing integration state could not be validated.", + { + "reason": state_error.detail or state_error.kind, + "recommendation": "Repair .specify/integration.json before reinitializing.", + }, + ) + if state is None: + return + + installed = installed_integration_keys(state) + current_key = default_integration_key(state) + if current_key and current_key != integration.key: + raise _InitIntegrationOptionsError( + "Reinitialization cannot replace the active integration.", + { + "current_integration": current_key, + "requested_integration": integration.key, + "recommendation": ( + f"Run 'specify integration switch {integration.key}' instead." + ), + }, + ) + incompatible_installed = [key for key in installed if key != integration.key] + if incompatible_installed: + raise _InitIntegrationOptionsError( + "Reinitialization cannot replace the installed integration set.", + { + "installed_integrations": installed, + "requested_integration": integration.key, + "recommendation": ( + "Use 'specify integration switch' or the integration " + "install/uninstall commands instead." + ), + }, + ) + + stored = integration_setting(state, integration.key) + if not stored: + return + stored_parsed = stored.get("parsed_options") + if not isinstance(stored_parsed, dict): + stored_parsed = None + stored_mode = integration.is_skills_mode( + stored_parsed, + project_root=project_path, + ) + requested_mode = integration.is_skills_mode( + parsed_options, + project_root=project_path, + ) + if stored_mode != requested_mode: + raise _InitIntegrationOptionsError( + "Reinitialization cannot change the integration command layout.", + { + "integration": integration.key, + "current_layout": "skills" if stored_mode else "commands", + "requested_layout": "skills" if requested_mode else "commands", + "recommendation": ( + f"Run 'specify integration upgrade {integration.key} " + "--integration-options \"...\"' instead." + ), + }, + ) + + if requested_destination is not None: + try: + stored_commands_dir = integration._resolve_commands_dir( + stored_parsed, + {"raw_options": stored.get("raw_options")}, + ) + except ValueError as exc: + raise _InitIntegrationOptionsError( + "The existing generic integration destination is invalid.", + { + "reason": str(exc), + "recommendation": ( + "Repair .specify/integration.json before reinitializing." + ), + }, + ) from exc + stored_destination = (project_path / stored_commands_dir).resolve() + if stored_destination != requested_destination: + raise _InitIntegrationOptionsError( + "Reinitialization cannot change the generic integration destination.", + { + "current_destination": str(stored_destination), + "requested_destination": str(requested_destination), + "recommendation": ( + "Run 'specify integration upgrade generic " + "--integration-options \"--commands-dir ...\"' instead." + ), + }, + ) + + +def _directory_identity(path: Path) -> tuple[int, int]: + state = path.stat(follow_symlinks=False) + if not stat.S_ISDIR(state.st_mode): + raise OSError(f"Target is no longer a directory: {path}") + return state.st_dev, state.st_ino + + +def _claim_new_target( + project_path: Path, + *, + should_claim: bool, +) -> tuple[int, int] | None: + if not should_claim: + return None + try: + project_path.mkdir(parents=True, exist_ok=False) + except FileExistsError as exc: + is_directory = project_path.is_dir() + raise _InitTargetClaimError( + "target_exists" if is_directory else "target_not_directory", + ( + "The target directory was created concurrently." + if is_directory + else "The target was replaced by a non-directory." + ), + {"path": str(project_path), "concurrent_creation": True}, + ) from exc + except OSError as exc: + raise _InitTargetClaimError( + "target_unavailable", + "The target directory could not be created.", + {"path": str(project_path), "reason": str(exc)}, + ) from exc + try: + return _directory_identity(project_path) + except OSError as exc: + raise _InitRollbackError( + { + "code": "target_identity_unavailable", + "exception_type": exc.__class__.__name__, + "reason": str(exc), + } + ) from exc + + +def _rollback_new_target( + project_path: Path, + target_identity: tuple[int, int] | None, +) -> None: + if target_identity is None: + return + try: + staging_root = Path( + tempfile.mkdtemp( + prefix=".specify-rollback-", + dir=project_path.parent, + ) + ) + except OSError as exc: + raise _InitRollbackError( + { + "code": "target_staging_failed", + "exception_type": exc.__class__.__name__, + "reason": str(exc), + } + ) from exc + + staged_target = staging_root / "target" + try: + project_path.rename(staged_target) + except OSError as exc: + cleanup_error = None + try: + staging_root.rmdir() + except OSError as cleanup_exc: + cleanup_error = str(cleanup_exc) + details = { + "code": "target_staging_failed", + "exception_type": exc.__class__.__name__, + "reason": str(exc), + } + if cleanup_error is not None: + details["staging_cleanup_error"] = cleanup_error + raise _InitRollbackError( + details + ) from exc + + def preserve_staged_target() -> dict[str, Any]: + details: dict[str, Any] = {"preserved_path": str(staged_target)} + if project_path.exists(): + details["restore_error"] = "The target path is occupied." + return details + try: + staged_target.rename(project_path) + except OSError as exc: + details["restore_error"] = str(exc) + return details + details["preserved_path"] = str(project_path) + try: + staging_root.rmdir() + except OSError as exc: + details["staging_cleanup_error"] = str(exc) + return details + + try: + staged_identity = _directory_identity(staged_target) + except OSError as exc: + details = preserve_staged_target() + raise _InitRollbackError( + { + "code": "target_identity_unavailable", + "exception_type": exc.__class__.__name__, + "reason": str(exc), + **details, + } + ) from exc + if staged_identity != target_identity: + details = preserve_staged_target() + raise _InitRollbackError( + { + "code": "target_identity_changed", + "reason": ( + "The target path no longer identified the directory created " + "by this invocation when rollback began." + ), + **details, + } + ) + try: + shutil.rmtree(staged_target) + except OSError as exc: + details = preserve_staged_target() + raise _InitRollbackError( + { + "code": "target_cleanup_failed", + "exception_type": exc.__class__.__name__, + "reason": str(exc), + **details, + } + ) from exc + try: + staging_root.rmdir() + except OSError as exc: + raise _InitRollbackError( + { + "code": "target_cleanup_failed", + "exception_type": exc.__class__.__name__, + "reason": str(exc), + "preserved_path": str(staging_root), + } + ) from exc + + +def _warning( + code: str, + message: str, + **details: Any, +) -> dict[str, Any]: + return {"code": code, "message": message, "details": details} + + +def _build_init_result( + *, + project_path: Path, + operation: str, + integration_key: str, + integration_defaulted: bool, + script_type: str, + script_defaulted: bool, + here: bool, + workflow: dict[str, Any], + constitution: dict[str, Any], + preset: dict[str, Any] | None, + extensions: list[dict[str, Any]], + warnings: list[dict[str, Any]], + tracker: StepTracker, +) -> dict[str, Any]: + permission_step = next( + (step for step in tracker.steps if step["key"] == "chmod"), + None, + ) + if permission_step is None or permission_step["status"] == "pending": + permissions = {"status": "not_applicable"} + elif permission_step["status"] == "error": + permissions = { + "status": "partial", + "detail": permission_step["detail"], + } + warnings.append( + _warning( + "script_permission_update_failed", + "Some executable script permissions could not be updated.", + detail=permission_step["detail"], + ) + ) + else: + permissions = { + "status": "completed", + "detail": permission_step["detail"], + } + + next_steps: list[dict[str, Any]] = [] + if not here: + next_steps.append( + { + "action": "change_directory", + "path": str(project_path), + } + ) + next_steps.extend( + [ + { + "action": "start_agent", + "integration": integration_key, + "working_directory": str(project_path), + }, + { + "action": "run_spec_kit", + "commands": [ + "constitution", + "specify", + "plan", + "tasks", + "implement", + "converge", + ], + "working_directory": str(project_path), + }, + ] + ) + return { + "project": { + "name": project_path.name, + "path": str(project_path), + "operation": operation, + }, + "integration": { + "key": integration_key, + "defaulted": integration_defaulted, + "status": "installed", + }, + "script": { + "type": script_type, + "defaulted": script_defaulted, + }, + "components": { + "shared_infrastructure": { + "status": "installed", + "script_type": script_type, + }, + "workflow": workflow, + "constitution": constitution, + "script_permissions": permissions, + "preset": preset, + "extensions": extensions, + }, + "warnings": warnings, + "next_steps": next_steps, + } + + def _stdin_is_interactive() -> bool: return sys.stdin.isatty() @@ -98,11 +531,21 @@ def _confirm_extension_url_trust( return approvals -def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_version: str) -> str: +@dataclass(frozen=True) +class _InitExtensionInstallResult: + status: str + message: str + + +def _install_extension_during_init_result( + project_path: Path, + ext_spec: str, + speckit_version: str, +) -> _InitExtensionInstallResult: """Install a single extension during ``specify init``. Handles bundled extension names, local directory paths, and HTTPS URLs. - Returns a short status message on success. + Returns machine status separately from the human-facing message. Raises ``ValueError`` on failure so the caller can convert it to a tracker error without aborting the entire init. """ @@ -126,7 +569,10 @@ def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_ve ) except ExtensionError as exc: raise ValueError(str(exc)) from exc - return f"{manifest.name} v{manifest.version} installed" + return _InitExtensionInstallResult( + "installed", + f"{manifest.name} v{manifest.version} installed", + ) # --- Local path --- if ext_spec.startswith(("./", "../", "/", "~/", ".\\", "..\\")) or Path(ext_spec).is_absolute(): @@ -136,15 +582,24 @@ def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_ve if not (source_path / "extension.yml").exists(): raise ValueError(f"No extension.yml found in {source_path}") manifest = manager.install_from_directory(source_path, speckit_version) - return f"{manifest.name} v{manifest.version} installed" + return _InitExtensionInstallResult( + "installed", + f"{manifest.name} v{manifest.version} installed", + ) # --- Bundled extension name or catalog ID --- bundled_path = _locate_bundled_extension(ext_spec) if bundled_path is not None: if manager.registry.is_installed(ext_spec): - return "already installed" + return _InitExtensionInstallResult( + "already_installed", + "already installed", + ) manifest = manager.install_from_directory(bundled_path, speckit_version) - return f"{manifest.name} v{manifest.version} installed" + return _InitExtensionInstallResult( + "installed", + f"{manifest.name} v{manifest.version} installed", + ) # Fall back to catalog catalog = ExtensionCatalog(project_path) @@ -159,9 +614,15 @@ 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): - return "already installed" + return _InitExtensionInstallResult( + "already_installed", + "already installed", + ) manifest = manager.install_from_directory(bundled_path, speckit_version) - return f"{manifest.name} v{manifest.version} installed" + return _InitExtensionInstallResult( + "installed", + f"{manifest.name} v{manifest.version} installed", + ) if ext_info.get("bundled") and not ext_info.get("download_url"): from .extensions import REINSTALL_COMMAND @@ -186,7 +647,23 @@ def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_ve ) finally: zip_path.unlink(missing_ok=True) - return f"{manifest.name} v{manifest.version} installed" + return _InitExtensionInstallResult( + "installed", + f"{manifest.name} v{manifest.version} installed", + ) + + +def _install_extension_during_init( + project_path: Path, + ext_spec: str, + speckit_version: str, +) -> str: + """Compatibility wrapper returning the existing human status message.""" + return _install_extension_during_init_result( + project_path, + ext_spec, + speckit_version, + ).message def _shell_quote_arg(value: str) -> str: @@ -257,7 +734,9 @@ def ensure_constitution_from_template( def register(app: typer.Typer) -> None: - @app.command() + from ._command_init_json import InitJsonCommand + + @app.command(cls=InitJsonCommand) def init( project_name: str = typer.Argument( None, @@ -291,6 +770,14 @@ def init( "PTY but cannot send arrow-key input." ), ), + json_output: bool = typer.Option( + False, + "--json", + help=( + "Emit one machine-readable JSON result. Implies non-interactive " + "behavior but does not imply force or trust/permission flags." + ), + ), skip_tls: bool = typer.Option( False, "--skip-tls", @@ -379,6 +866,39 @@ def init( specify init my-project --extension ./my-extensions/custom-ext # Local path extension specify init my-project --extension https://example.com/extensions/my-ext.zip --trust-extension-urls # URL extension (non-interactive) """ + if json_output is True: + from ._command_init_json import run_init_json + + return run_init_json( + execute=lambda: init( + project_name=project_name, + script_type=script_type, + ignore_agent_tools=ignore_agent_tools, + here=here, + force=force, + non_interactive=True, + json_output=False, + skip_tls=skip_tls, + debug=debug, + github_token=github_token, + offline=offline, + preset=preset, + integration=integration, + integration_options=integration_options, + extensions=extensions, + trust_extension_urls=trust_extension_urls, + ), + project_name=project_name, + script_type=script_type, + ignore_agent_tools=ignore_agent_tools, + here=here, + force=force, + integration=integration, + integration_options=integration_options, + extensions=extensions, + trust_extension_urls=trust_extension_urls, + ) + # Lazy imports to avoid circular dependency — __init__.py imports this module from . import ( _install_shared_infra_or_exit, @@ -395,7 +915,8 @@ def init( _write_integration_json, ) - show_banner() + if not _init_json_mode.get(): + show_banner() from .integrations import INTEGRATION_REGISTRY, get_integration @@ -491,6 +1012,17 @@ def init( if project_path.exists(): safe_name = _escape_markup(str(project_name)) if not project_path.is_dir(): + if _init_json_mode.get(): + _init_failure_context.set( + _InitTargetClaimError( + "target_not_directory", + "The target path exists but is not a directory.", + { + "path": str(project_path), + "concurrent_creation": True, + }, + ) + ) console.print( f"[red]Error:[/red] '{safe_name}' exists but is not a directory." ) @@ -508,6 +1040,17 @@ def init( f"[cyan]--force supplied: merging into existing directory '[cyan]{safe_name}[/cyan]'[/cyan]" ) else: + if _init_json_mode.get(): + _init_failure_context.set( + _InitTargetClaimError( + "target_exists", + "The target directory already exists.", + { + "path": str(project_path), + "concurrent_creation": True, + }, + ) + ) error_panel = Panel( f"Directory already exists: '[cyan]{safe_name}[/cyan]'\n" "Please choose a different project name or remove the existing directory.\n" @@ -520,6 +1063,25 @@ def init( console.print(error_panel) raise typer.Exit(1) + operation = ( + "reinitialized" + if (project_path / ".specify").is_dir() + else "merged" + if dir_existed_before + else "created" + ) + init_warnings: list[dict[str, Any]] = [] + workflow_outcome: dict[str, Any] = { + "id": "speckit", + "status": "skipped", + "reason": "not_processed", + } + preset_outcome: dict[str, Any] | None = None + extension_outcomes: list[dict[str, Any]] = [] + constitution_existed = ( + project_path / ".specify" / "memory" / "constitution.md" + ).exists() + if integration: if integration not in AGENT_CONFIG: console.print( @@ -572,9 +1134,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 _init_json_mode.get(): + 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) @@ -614,8 +1177,11 @@ 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 _init_json_mode.get(): + console.print( + f"[cyan]Selected coding agent integration:[/cyan] {selected_ai}" + ) + console.print(f"[cyan]Selected script type:[/cyan] {selected_script}") tracker = StepTracker("Initialize Specify Project") @@ -644,6 +1210,32 @@ def init( tracker.add("final", "Finalize") + integration_parsed_options: dict[str, Any] = {} + if integration_options: + extra = _parse_integration_options( + resolved_integration, + integration_options, + ) + if extra: + integration_parsed_options.update(extra) + try: + _validate_init_integration_options( + project_path, + resolved_integration, + integration_parsed_options or None, + integration_options, + ) + except _InitIntegrationOptionsError as exc: + console.print(f"[red]Error:[/red] {exc.message}") + recommendation = exc.details.get("recommendation") + if recommendation: + console.print(f"[dim]{recommendation}[/dim]") + raise typer.Exit(1) from exc + resolved_integration.is_skills_mode( + integration_parsed_options or None, + project_root=project_path, + ) + # Resolve trust for URL-based extensions BEFORE entering the Live # display: the confirmation prompt cannot be shown/answered underneath # the Rich Live spinner. URL installs are default-deny unless the user @@ -652,20 +1244,57 @@ def init( if extensions: url_specs = [e for e in extensions if _ext_spec_is_url(e)] if url_specs: + from ._download_security import is_https_or_localhost_http + + unsupported_urls = [ + spec for spec in url_specs if not is_https_or_localhost_http(spec) + ] + if unsupported_urls: + console.print( + "[red]Error:[/red] Extension URLs must use HTTPS; " + "HTTP is allowed only for localhost." + ) + raise typer.Exit(1) extension_url_approvals = _confirm_extension_url_trust( url_specs, trust_override=trust_extension_urls, allow_prompt=_prompts_allowed(non_interactive), ) + try: + target_identity = _claim_new_target( + project_path, + should_claim=not here and not dir_existed_before, + ) + if target_identity is not None: + _init_rollback_context.set( + { + "status": "claimed", + "path": str(project_path), + } + ) + except (_InitRollbackError, _InitTargetClaimError) as exc: + _init_failure_context.set(exc) + console.print(f"[red]Error:[/red] Could not create target: {exc}") + raise typer.Exit(1) from exc + # Disable transient mode on Windows: PowerShell 5.1's legacy console # hangs when Rich tries to restore cursor state via VT escape sequences. _transient = sys.platform != "win32" - with Live( - tracker.render(), console=console, refresh_per_second=8, transient=_transient - ) as live: - tracker.attach_refresh(lambda: live.update(tracker.render())) + live_context = ( + nullcontext() + if _init_json_mode.get() + else Live( + tracker.render(), + console=console, + refresh_per_second=8, + transient=_transient, + ) + ) + with live_context as live: + if live is not None: + tracker.attach_refresh(lambda: live.update(tracker.render())) try: from .integrations.manifest import IntegrationManifest @@ -676,14 +1305,6 @@ def init( version=get_speckit_version(), ) - integration_parsed_options: dict[str, Any] = {} - if integration_options: - extra = _parse_integration_options( - resolved_integration, integration_options - ) - if extra: - integration_parsed_options.update(extra) - from .events import resolve_events events_map = resolve_events( resolved_integration.key, @@ -701,13 +1322,43 @@ def init( ) manifest.save() + init_opts = { + "ai": selected_ai, + "integration": resolved_integration.key, + "here": here, + "script": selected_script, + "feature_numbering": "sequential", + "speckit_version": get_speckit_version(), + } + if resolved_integration.is_skills_mode( + integration_parsed_options or None, project_root=project_path + ): + init_opts["ai_skills"] = True + save_init_options(project_path, init_opts) + + integration_settings = _with_integration_setting( + {}, + resolved_integration.key, + resolved_integration, + script_type=selected_script, + raw_options=integration_options, + parsed_options=integration_parsed_options or None, + project_root=project_path, + ) + _write_integration_json( + project_path, + resolved_integration.key, + [resolved_integration.key], + integration_settings, + ) + if force: from .integrations._helpers import ( _register_extensions_for_agent, _register_presets_for_agent, ) - _register_extensions_for_agent( + extension_registration_error = _register_extensions_for_agent( project_path, resolved_integration.key, force=True, @@ -716,7 +1367,18 @@ def init( " may need re-registration." ), ) - _register_presets_for_agent( + if extension_registration_error is not None: + init_warnings.append( + _warning( + "extension_reregistration_failed", + "The project was re-initialized, but installed extensions may need re-registration.", + integration=resolved_integration.key, + reason=extension_registration_error.replace( + "\n", " " + ).strip(), + ) + ) + preset_registration_error = _register_presets_for_agent( project_path, resolved_integration.key, continuing=( @@ -724,22 +1386,17 @@ def init( " may need re-registration." ), ) - - integration_settings = _with_integration_setting( - {}, - resolved_integration.key, - resolved_integration, - script_type=selected_script, - raw_options=integration_options, - parsed_options=integration_parsed_options or None, - project_root=project_path, - ) - _write_integration_json( - project_path, - resolved_integration.key, - [resolved_integration.key], - integration_settings, - ) + if preset_registration_error is not None: + init_warnings.append( + _warning( + "preset_reregistration_failed", + "The project was re-initialized, but installed presets may need re-registration.", + integration=resolved_integration.key, + reason=preset_registration_error.replace( + "\n", " " + ).strip(), + ) + ) tracker.complete( "integration", @@ -775,6 +1432,10 @@ def init( wf_registry = WorkflowRegistry(project_path) if wf_registry.is_installed("speckit"): tracker.complete("workflow", "already installed") + workflow_outcome = { + "id": "speckit", + "status": "already_installed", + } else: import shutil as _shutil @@ -799,25 +1460,41 @@ def init( }, ) tracker.complete("workflow", "speckit installed") + workflow_outcome = { + "id": "speckit", + "status": "installed", + "version": definition.version, + } else: tracker.skip("workflow", "bundled workflow not found") + workflow_outcome = { + "id": "speckit", + "status": "skipped", + "reason": "bundled_workflow_not_found", + } + init_warnings.append( + _warning( + "bundled_workflow_not_found", + "The bundled speckit workflow was not available.", + workflow="speckit", + ) + ) except Exception as wf_err: sanitized_wf = str(wf_err).replace("\n", " ").strip() tracker.error("workflow", f"install failed: {sanitized_wf[:120]}") - - init_opts = { - "ai": selected_ai, - "integration": resolved_integration.key, - "here": here, - "script": selected_script, - "feature_numbering": "sequential", - "speckit_version": get_speckit_version(), - } - if resolved_integration.is_skills_mode( - integration_parsed_options or None, project_root=project_path - ): - init_opts["ai_skills"] = True - save_init_options(project_path, init_opts) + workflow_outcome = { + "id": "speckit", + "status": "failed", + "reason": sanitized_wf, + } + init_warnings.append( + _warning( + "workflow_install_failed", + "The project was initialized without the optional bundled workflow.", + workflow="speckit", + reason=sanitized_wf, + ) + ) ensure_executable_scripts(project_path, tracker=tracker) @@ -830,15 +1507,27 @@ def init( local_path = Path(preset).resolve() if local_path.is_dir() and (local_path / "preset.yml").exists(): - preset_manager.install_from_directory( + preset_manifest = preset_manager.install_from_directory( local_path, speckit_ver ) + preset_outcome = { + "requested": preset, + "id": preset_manifest.id, + "status": "installed", + "source": "local", + } else: bundled_path = _locate_bundled_preset(preset) if bundled_path: - preset_manager.install_from_directory( + preset_manifest = preset_manager.install_from_directory( bundled_path, speckit_ver ) + preset_outcome = { + "requested": preset, + "id": preset_manifest.id, + "status": "installed", + "source": "bundled", + } else: preset_catalog = PresetCatalog(project_path) pack_info = preset_catalog.get_pack_info(preset) @@ -846,6 +1535,18 @@ def init( console.print( f"[yellow]Warning:[/yellow] Preset '{preset}' not found in catalog. Skipping." ) + preset_outcome = { + "requested": preset, + "status": "skipped", + "reason": "not_found", + } + init_warnings.append( + _warning( + "preset_not_found", + "The requested optional preset was not found and was skipped.", + preset=preset, + ) + ) elif pack_info.get("bundled") and not pack_info.get( "download_url" ): @@ -861,15 +1562,33 @@ def init( console.print( f"Try reinstalling: {REINSTALL_COMMAND}" ) + preset_outcome = { + "requested": preset, + "status": "failed", + "reason": "bundled_preset_not_found", + } + init_warnings.append( + _warning( + "bundled_preset_not_found", + "The requested bundled preset was missing from the installed package.", + preset=preset, + ) + ) else: zip_path = None try: zip_path = preset_catalog.download_pack(preset) - preset_manager.install_from_zip( + preset_manifest = preset_manager.install_from_zip( zip_path, speckit_ver, catalog_name=pack_info.get("_catalog_name"), ) + preset_outcome = { + "requested": preset, + "id": preset_manifest.id, + "status": "installed", + "source": "catalog", + } except PresetError as preset_err: _print_cli_warning( "install", @@ -878,12 +1597,33 @@ def init( preset_err, continuing="Continuing without the optional preset.", ) + reason = str(preset_err).replace("\n", " ").strip() + preset_outcome = { + "requested": preset, + "status": "failed", + "reason": reason, + } + init_warnings.append( + _warning( + "preset_install_failed", + "The project was initialized without the optional preset.", + preset=preset, + reason=reason, + ) + ) finally: if zip_path is not None: try: zip_path.unlink(missing_ok=True) - except OSError: - pass + except OSError as cleanup_err: + init_warnings.append( + _warning( + "preset_download_cleanup_failed", + "The preset was processed, but its temporary download could not be removed.", + preset=preset, + reason=str(cleanup_err), + ) + ) except Exception as preset_err: _print_cli_warning( "install", @@ -892,10 +1632,24 @@ def init( preset_err, continuing="Continuing without the optional preset.", ) + reason = str(preset_err).replace("\n", " ").strip() + preset_outcome = { + "requested": preset, + "status": "failed", + "reason": reason, + } + init_warnings.append( + _warning( + "preset_install_failed", + "The project was initialized without the optional preset.", + preset=preset, + reason=reason, + ) + ) # Install extensions specified via --extension if extensions: - from .extensions._commands import _refresh_events_and_warn + from .events import EventRefreshError, refresh_integration_events speckit_ver = get_speckit_version() any_extension_installed = False @@ -911,35 +1665,161 @@ def init( "skipped: untrusted URL not confirmed " "(use --trust-extension-urls)", ) + extension_outcomes.append( + { + "requested": ext_spec, + "status": "skipped", + "reason": "extension_url_trust_required", + } + ) continue try: - status_msg = _install_extension_during_init( + install_result = _install_extension_during_init_result( project_path, ext_spec, speckit_ver ) - tracker.complete(f"extension-{i}", status_msg) - any_extension_installed = True + tracker.complete( + f"extension-{i}", + install_result.message, + ) + extension_outcomes.append( + { + "requested": ext_spec, + "status": install_result.status, + "message": install_result.message, + } + ) + if install_result.status == "installed": + any_extension_installed = True except Exception as ext_err: sanitized_ext = str(ext_err).replace("\n", " ").strip() tracker.error( f"extension-{i}", f"failed: {_escape_markup(sanitized_ext[:120])}", ) + extension_outcomes.append( + { + "requested": ext_spec, + "status": "failed", + "reason": sanitized_ext, + } + ) + init_warnings.append( + _warning( + "extension_install_failed", + "The project was initialized without a requested optional extension.", + extension=ext_spec, + reason=sanitized_ext, + ) + ) # Refresh native event configuration once after the batch so # that an extension declaring ``events:`` has its hooks # activated, mirroring the ``extension add`` path. if any_extension_installed: - _refresh_events_and_warn(project_path) + try: + refresh_integration_events(project_path) + except EventRefreshError as refresh_err: + console.print( + "[yellow]Warning:[/yellow] Extensions were installed, " + "but integration event configuration refresh failed." + ) + init_warnings.append( + _warning( + "extension_event_refresh_failed", + "Extensions were installed, but one or more integration event configurations could not be refreshed.", + failures=[ + { + "integration": key, + "reason": detail, + } + for key, detail in refresh_err.failures + ], + ) + ) # Seed the constitution AFTER preset installation so that a # preset-provided constitution-template (resolved via the # priority stack) wins over the core template. ensure_constitution_from_template(project_path, tracker=tracker) + constitution_path = ( + project_path / ".specify" / "memory" / "constitution.md" + ) + constitution_step = next( + step + for step in tracker.steps + if step["key"] == "constitution" + ) + if constitution_existed: + constitution_outcome = { + "status": "preserved", + "path": str(constitution_path), + } + elif constitution_path.exists(): + constitution_outcome = { + "status": "created", + "path": str(constitution_path), + "source": ( + "composed" + if "composed" in constitution_step["detail"] + else "copied" + ), + } + else: + constitution_outcome = { + "status": "failed", + "path": str(constitution_path), + "reason": constitution_step["detail"] or "template_not_found", + } + init_warnings.append( + _warning( + "constitution_initialization_failed", + "The project was initialized, but its constitution could not be created.", + path=str(constitution_path), + reason=constitution_outcome["reason"], + ) + ) tracker.complete("final", "project ready") + init_result = _build_init_result( + project_path=project_path, + operation=operation, + integration_key=resolved_integration.key, + integration_defaulted=not bool(integration), + script_type=selected_script, + script_defaulted=not bool(script_type), + here=here, + workflow=workflow_outcome, + constitution=constitution_outcome, + preset=preset_outcome, + extensions=extension_outcomes, + warnings=init_warnings, + tracker=tracker, + ) except (typer.Exit, SystemExit): + try: + _rollback_new_target(project_path, target_identity) + if target_identity is not None: + _init_rollback_context.set( + { + "status": "completed", + "path": str(project_path), + } + ) + except _InitRollbackError as cleanup_err: + _init_rollback_context.set( + { + "status": "failed", + "path": str(project_path), + } + ) + _init_failure_context.set(cleanup_err) + console.print( + "[red]Error:[/red] Initialization failed and the new " + f"target could not be removed: {cleanup_err.details['code']}" + ) raise except Exception as e: + _init_failure_context.set(e) tracker.error("final", str(e)) console.print( Panel( @@ -966,13 +1846,35 @@ def init( border_style="magenta", ) ) - if not here and project_path.exists() and not dir_existed_before: - shutil.rmtree(project_path) + try: + _rollback_new_target(project_path, target_identity) + if target_identity is not None: + _init_rollback_context.set( + { + "status": "completed", + "path": str(project_path), + } + ) + except _InitRollbackError as cleanup_err: + _init_rollback_context.set( + { + "status": "failed", + "path": str(project_path), + } + ) + _init_failure_context.set(cleanup_err) + console.print( + "[red]Error:[/red] Initialization failed and the new " + f"target could not be removed: {cleanup_err.details['code']}" + ) raise typer.Exit(1) finally: pass - if _transient: + if _init_json_mode.get(): + return init_result + + if _transient and not _init_json_mode.get(): console.print(tracker.render()) console.print("\n[bold green]Project ready.[/bold green]") @@ -1178,3 +2080,5 @@ def _display_cmd(name: str) -> str: ) console.print() console.print(enhancements_panel) + + return init_result diff --git a/src/specify_cli/integrations/_commands.py b/src/specify_cli/integrations/_commands.py index ef90e576e9..f17d4a7175 100644 --- a/src/specify_cli/integrations/_commands.py +++ b/src/specify_cli/integrations/_commands.py @@ -12,6 +12,7 @@ # Re-export helpers used by command_init.py and tests from ._helpers import ( # noqa: F401 + _IntegrationOptionsError, _cli_error_detail, _cli_phase_label, _parse_integration_options, diff --git a/src/specify_cli/integrations/_helpers.py b/src/specify_cli/integrations/_helpers.py index 38e87294ec..b4aa7c0f42 100644 --- a/src/specify_cli/integrations/_helpers.py +++ b/src/specify_cli/integrations/_helpers.py @@ -184,6 +184,14 @@ def _resolve_integration_script_type( # Integration options # --------------------------------------------------------------------------- +class _IntegrationOptionsError(typer.Exit): + """Exit carrying the parser diagnostic for machine-readable callers.""" + + def __init__(self, diagnostic: str): + super().__init__(1) + self.diagnostic = diagnostic + + def _parse_integration_options(integration: Any, raw_options: str) -> dict[str, Any] | None: """Parse --integration-options string into a dict matching the integration's declared options. @@ -198,8 +206,9 @@ def _parse_integration_options(integration: Any, raw_options: str) -> dict[str, # makes shlex raise "No closing quotation". Translate it into the same # clean exit-1 UX as every other bad-input path below rather than # letting a raw traceback escape. - console.print(f"[red]Error:[/red] Could not parse integration options: {exc}.") - raise typer.Exit(1) + diagnostic = f"Could not parse integration options: {exc}." + console.print(f"[red]Error:[/red] {diagnostic}") + raise _IntegrationOptionsError(diagnostic) declared_options = list(integration.options()) declared = {opt.name.lstrip("-"): opt for opt in declared_options} allowed = ", ".join(sorted(opt.name for opt in declared_options)) @@ -208,9 +217,11 @@ def _parse_integration_options(integration: Any, raw_options: str) -> dict[str, token = tokens[i] if not token.startswith("-"): console.print(f"[red]Error:[/red] Unexpected integration option value '{escape(token)}'.") + diagnostic = f"Unexpected integration option value '{token}'." if allowed: console.print(f"Allowed options: {allowed}") - raise typer.Exit(1) + diagnostic += f" Allowed options: {allowed}." + raise _IntegrationOptionsError(diagnostic) name = token.lstrip("-") value: str | None = None # Handle --name=value syntax @@ -219,14 +230,19 @@ def _parse_integration_options(integration: Any, raw_options: str) -> dict[str, opt = declared.get(name) if not opt: console.print(f"[red]Error:[/red] Unknown integration option '{escape(token)}'.") + diagnostic = f"Unknown integration option '{token}'." if allowed: console.print(f"Allowed options: {allowed}") - raise typer.Exit(1) + diagnostic += f" Allowed options: {allowed}." + raise _IntegrationOptionsError(diagnostic) key = name.replace("-", "_") if opt.is_flag: if value is not None: - console.print(f"[red]Error:[/red] Option '{opt.name}' is a flag and does not accept a value.") - raise typer.Exit(1) + diagnostic = ( + f"Option '{opt.name}' is a flag and does not accept a value." + ) + console.print(f"[red]Error:[/red] {diagnostic}") + raise _IntegrationOptionsError(diagnostic) parsed[key] = True i += 1 elif value is not None: @@ -236,8 +252,9 @@ def _parse_integration_options(integration: Any, raw_options: str) -> dict[str, parsed[key] = tokens[i + 1] i += 2 else: - console.print(f"[red]Error:[/red] Option '{opt.name}' requires a value.") - raise typer.Exit(1) + diagnostic = f"Option '{opt.name}' requires a value." + console.print(f"[red]Error:[/red] {diagnostic}") + raise _IntegrationOptionsError(diagnostic) return parsed or None @@ -370,13 +387,14 @@ def _best_effort_extension_op( *, phase: str, continuing: str, -) -> None: +) -> str | None: """Run a best-effort ``ExtensionManager`` operation for ``agent_key``. ``op`` receives the ``ExtensionManager`` and ``agent_key``. Any failure is surfaced as a warning via ``_print_cli_warning`` and never aborts the surrounding integration operation. ``continuing`` describes what already - succeeded so the warning makes the partial outcome clear. + succeeded so the warning makes the partial outcome clear. Returns the + failure reason for callers that also expose structured outcomes. """ try: from ..extensions import ExtensionManager @@ -387,6 +405,8 @@ def _best_effort_extension_op( from .. import _print_cli_warning _print_cli_warning(phase, "integration", agent_key, ext_err, continuing=continuing) + return str(ext_err) + return None def _register_extensions_for_agent( @@ -395,7 +415,7 @@ def _register_extensions_for_agent( *, continuing: str, force: bool = False, -) -> None: +) -> str | None: """Register all enabled extensions' commands/skills for ``agent_key``. ``use`` / ``switch`` re-register enabled extensions for the agent they @@ -413,11 +433,12 @@ def _register_extensions_for_agent( extension content is layered on top of the core-template files that ``setup()`` just regenerated (fixes the skip-guard bug for skills mode). - Best-effort: never aborts the surrounding integration operation. Callers - invoke it *after* the use/upgrade/switch transaction has committed so a - failure here cannot trigger a rollback. + Best-effort: never aborts the surrounding integration operation and returns + the failure reason when registration fails. Callers invoke it *after* the + use/upgrade/switch transaction has committed so a failure here cannot + trigger a rollback. """ - _best_effort_extension_op( + return _best_effort_extension_op( project_root, agent_key, lambda mgr, key: mgr.register_enabled_extensions_for_agent(key, force=force), @@ -452,7 +473,7 @@ def _register_presets_for_agent( agent_key: str, *, continuing: str, -) -> None: +) -> str | None: """Register all enabled presets' command overrides/skills for ``agent_key``. Presets follow the same single-active rule as extensions (#2948): @@ -460,7 +481,8 @@ def _register_presets_for_agent( activate (rescaffold), so a preset installed while a different integration was active is not left targeting that inactive integration. - Best-effort: never aborts the surrounding integration operation. + Best-effort: never aborts the surrounding integration operation and returns + the failure reason when registration fails. """ try: from ..presets import PresetManager @@ -477,6 +499,8 @@ def _register_presets_for_agent( preset_err, continuing=continuing, ) + return str(preset_err) + return None def _resync_manifest_after_registration( diff --git a/src/specify_cli/integrations/bob/__init__.py b/src/specify_cli/integrations/bob/__init__.py index 821f03dd66..38d10dc058 100644 --- a/src/specify_cli/integrations/bob/__init__.py +++ b/src/specify_cli/integrations/bob/__init__.py @@ -24,8 +24,6 @@ from pathlib import Path from typing import Any -import typer - from ..base import ( IntegrationBase, IntegrationOption, @@ -45,12 +43,13 @@ def _validate_mode_options(parsed_options: dict[str, Any] | None) -> None: opts = parsed_options or {} if opts.get("skills") and opts.get("legacy_commands"): from ..._console import console + from .._helpers import _IntegrationOptionsError - console.print( - "[red]Error:[/red] --skills and --legacy-commands are mutually " - "exclusive; pass only one." + diagnostic = ( + "--skills and --legacy-commands are mutually exclusive; pass only one." ) - raise typer.Exit(1) + console.print(f"[red]Error:[/red] {diagnostic}") + raise _IntegrationOptionsError(diagnostic) def _warn_legacy_commands_deprecated() -> None: diff --git a/src/specify_cli/integrations/copilot/__init__.py b/src/specify_cli/integrations/copilot/__init__.py index 0ea251d0af..864186f6ea 100644 --- a/src/specify_cli/integrations/copilot/__init__.py +++ b/src/specify_cli/integrations/copilot/__init__.py @@ -27,8 +27,6 @@ from pathlib import Path from typing import Any -import typer - from ..base import IntegrationBase, IntegrationOption, SkillsIntegration from ..manifest import IntegrationManifest @@ -96,12 +94,13 @@ def _validate_mode_options(parsed_options: dict[str, Any] | None) -> None: opts = parsed_options or {} if opts.get("skills") and opts.get("commands"): from ..._console import console + from .._helpers import _IntegrationOptionsError - console.print( - "[red]Error:[/red] --skills and --commands are mutually exclusive; " - "pass only one." + diagnostic = ( + "--skills and --commands are mutually exclusive; pass only one." ) - raise typer.Exit(1) + console.print(f"[red]Error:[/red] {diagnostic}") + raise _IntegrationOptionsError(diagnostic) class _CopilotSkillsHelper(SkillsIntegration): diff --git a/tests/integrations/test_cli.py b/tests/integrations/test_cli.py index ed6864253b..542a26d104 100644 --- a/tests/integrations/test_cli.py +++ b/tests/integrations/test_cli.py @@ -1982,18 +1982,17 @@ def _zip_bytes_from_dir(source_dir): return buf.getvalue() def test_url_extension_rejects_non_https(self, tmp_path): - """A non-HTTPS URL is rejected before any download; init is not aborted.""" + """A non-HTTPS URL is rejected before target creation or download.""" project, result = self._run_init( tmp_path, ["--extension", "http://example.com/ext.zip", "--trust-extension-urls"], project_name="ext-http", ) - assert result.exit_code == 0, "init should not abort on a rejected URL" + assert result.exit_code == 1 normalized = _normalize_cli_output(result.output) - assert "failed" in normalized.lower() - # No extension directory should have been created for the bad URL. - assert not (project / ".specify" / "extensions" / "ext").exists() + assert "must use https" in normalized.lower() + assert not (project / ".specify").exists() def test_url_extension_skipped_without_trust(self, tmp_path): """Non-interactive URL install without --trust-extension-urls is denied.""" diff --git a/tests/specify_cli/bundles/test_command_init.py b/tests/specify_cli/bundles/test_command_init.py index adf60e0035..5c91b1eb0e 100644 --- a/tests/specify_cli/bundles/test_command_init.py +++ b/tests/specify_cli/bundles/test_command_init.py @@ -38,3 +38,17 @@ def test_override_symlinked_specify_errors_bundle_init_no_fallback( assert result.exit_code != 0 assert "symlinked .specify" in result.output assert not (elsewhere / ".specify").exists() + + +def test_bundle_init_direct_callback_keeps_human_output(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + + result = runner.invoke( + app, + ["bundle", "init", "--integration", "copilot", "--offline"], + ) + + assert result.exit_code == 0, result.output + assert "Initializing a Spec Kit project" in result.output + assert "Spec Kit project ready" in result.output + assert '"project":{' not in result.output diff --git a/tests/specify_cli/integrations/test_registration.py b/tests/specify_cli/integrations/test_registration.py index 6ef4256f74..fb95b771d5 100644 --- a/tests/specify_cli/integrations/test_registration.py +++ b/tests/specify_cli/integrations/test_registration.py @@ -106,6 +106,9 @@ def test_unbalanced_quote_exits_cleanly(self, capsys): with pytest.raises(typer.Exit) as excinfo: _parse_integration_options(integration, '--commands-dir "foo') assert excinfo.value.exit_code == 1 + assert excinfo.value.diagnostic == ( + "Could not parse integration options: No closing quotation." + ) assert "Error: Could not parse integration options: No closing quotation." in capsys.readouterr().out def test_bad_option_token_with_rich_markup_exits_cleanly(self): diff --git a/tests/specify_cli/test_command_init_json.py b/tests/specify_cli/test_command_init_json.py new file mode 100644 index 0000000000..ca9251e496 --- /dev/null +++ b/tests/specify_cli/test_command_init_json.py @@ -0,0 +1,1564 @@ +"""JSON output and parity tests for ``specify init``.""" + +from __future__ import annotations + +import json +import os +import stat +from pathlib import Path +from typing import Any + +import pytest +import typer +from typer.testing import CliRunner, Result + +from specify_cli import app + + +def _invoke(args: list[str], *, cwd: Path) -> Result: + previous = Path.cwd() + os.chdir(cwd) + try: + return CliRunner().invoke( + app, + ["init", *args], + catch_exceptions=False, + ) + finally: + os.chdir(previous) + + +def _success(result: Result) -> dict[str, Any]: + assert result.exit_code == 0, result.stderr or result.stdout + assert result.stderr == "" + assert result.stdout.endswith("\n") + assert result.stdout.count("\n") == 1 + assert "\x1b[" not in result.stdout + payload = json.loads(result.stdout) + assert isinstance(payload, dict) + assert "error" not in payload + return payload + + +def _failure(result: Result, code: str) -> dict[str, Any]: + assert result.exit_code == 1, result.stdout or result.stderr + assert result.stdout == "" + assert result.stderr.endswith("\n") + assert result.stderr.count("\n") == 1 + assert "\x1b[" not in result.stderr + payload = json.loads(result.stderr) + assert payload["error"]["code"] == code + assert set(payload["error"]) == {"code", "message", "details"} + return payload["error"] + + +def _normalize_value(value: Any, root: Path) -> Any: + if isinstance(value, dict): + return { + key: _normalize_value(item, root) + for key, item in value.items() + if key not in {"installed_at", "updated_at", "created_at"} + } + if isinstance(value, list): + return [_normalize_value(item, root) for item in value] + if isinstance(value, str): + return value.replace(str(root), "") + return value + + +def _project_snapshot(root: Path) -> dict[str, tuple[Any, ...]]: + snapshot: dict[str, tuple[Any, ...]] = {} + for path in sorted(root.rglob("*")): + relative = path.relative_to(root).as_posix() + mode = stat.S_IMODE(path.lstat().st_mode) + if path.is_symlink(): + snapshot[relative] = ("symlink", mode, os.readlink(path)) + elif path.is_dir(): + snapshot[relative] = ("directory", mode) + else: + content = path.read_bytes() + try: + normalized = _normalize_value( + json.loads(content.decode("utf-8")), + root, + ) + content = json.dumps( + normalized, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + except (UnicodeDecodeError, json.JSONDecodeError): + try: + content = content.decode("utf-8").replace( + str(root), + "", + ).encode("utf-8") + except UnicodeDecodeError: + pass + snapshot[relative] = ("file", mode, content) + return snapshot + + +def test_json_init_new_directory_uses_safe_defaults(tmp_path: Path): + project = tmp_path / "prøject" + + payload = _success( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ) + ) + + assert payload["project"] == { + "name": "prøject", + "path": str(project.resolve()), + "operation": "created", + } + assert payload["integration"] == { + "key": "copilot", + "defaulted": True, + "status": "installed", + } + assert payload["script"] == { + "type": "ps" if os.name == "nt" else "sh", + "defaulted": True, + } + assert payload["components"]["shared_infrastructure"]["status"] == "installed" + assert payload["components"]["workflow"]["status"] == "installed" + assert payload["components"]["constitution"]["status"] == "created" + assert payload["components"]["preset"] is None + assert payload["components"]["extensions"] == [] + assert (project / ".specify" / "init-options.json").is_file() + + +def test_json_init_here_honors_explicit_integration_and_script(tmp_path: Path): + payload = _success( + _invoke( + [ + "--here", + "--json", + "--integration", + "copilot", + "--script", + "py", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert payload["project"]["operation"] == "merged" + assert payload["project"]["path"] == str(tmp_path.resolve()) + assert payload["integration"]["defaulted"] is False + assert payload["script"] == {"type": "py", "defaulted": False} + assert all(step["action"] != "change_directory" for step in payload["next_steps"]) + + +def test_json_init_never_uses_prompt_or_rich_ui( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli.command_init as init_command + + monkeypatch.setattr(init_command, "_stdin_is_interactive", lambda: True) + + def fail_ui(*_args: Any, **_kwargs: Any) -> None: + raise AssertionError("JSON mode must not invoke interactive or Rich UI") + + monkeypatch.setattr(init_command, "select_with_arrows", fail_ui) + monkeypatch.setattr(init_command, "show_banner", fail_ui) + monkeypatch.setattr(init_command, "Live", fail_ui) + monkeypatch.setattr(init_command, "Panel", fail_ui) + monkeypatch.setattr(typer, "confirm", fail_ui) + + payload = _success( + _invoke( + ["project", "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ) + ) + + assert payload["integration"]["defaulted"] is True + assert payload["script"]["defaulted"] is True + + +def test_json_init_force_merges_nonempty_target(tmp_path: Path): + project = tmp_path / "existing" + project.mkdir() + marker = project / "keep.txt" + marker.write_text("keep", encoding="utf-8") + + _failure( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "target_exists", + ) + payload = _success( + _invoke( + [ + str(project), + "--json", + "--force", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert payload["project"]["operation"] == "merged" + assert marker.read_text(encoding="utf-8") == "keep" + + +@pytest.mark.parametrize( + ("args", "code"), + [ + (["--json"], "target_required"), + (["project", "--here", "--json"], "conflicting_target_options"), + ( + ["project", "--json", "--integration", "missing"], + "invalid_integration", + ), + ( + [ + "project", + "--json", + "--integration", + "generic", + ], + "invalid_integration_options", + ), + ( + [ + "project", + "--json", + "--integration", + "copilot", + "--integration-options=--skills --commands", + ], + "invalid_integration_options", + ), + ( + [ + "project", + "--json", + "--integration", + "copilot", + '--integration-options=--skills "', + ], + "invalid_integration_options", + ), + ( + [ + "project", + "--json", + "--integration", + "copilot", + "--integration-options=--missing", + ], + "invalid_integration_options", + ), + ( + [ + "project", + "--json", + "--integration", + "copilot", + "--integration-options=--skills=true", + ], + "invalid_integration_options", + ), + ( + [ + "project", + "--json", + "--integration", + "generic", + "--integration-options=--commands-dir", + ], + "invalid_integration_options", + ), + ( + [ + "project", + "--json", + "--integration", + "generic", + "--integration-options=--commands-dir ../outside", + ], + "invalid_integration_options", + ), + ( + [ + "project", + "--json", + "--integration", + "bob", + "--integration-options=--skills --legacy-commands", + ], + "invalid_integration_options", + ), + ( + [ + "project", + "--json", + "--extension", + "http://example.com/extension.zip", + "--trust-extension-urls", + ], + "invalid_extension_url", + ), + (["project", "--json", "--script", "fish"], "invalid_script_type"), + ], +) +def test_json_validation_failures_do_not_create_target( + tmp_path: Path, + args: list[str], + code: str, +): + _failure(_invoke(args, cwd=tmp_path), code) + assert not (tmp_path / "project").exists() + + +@pytest.mark.parametrize( + ("integration", "options", "diagnostic"), + [ + ( + "copilot", + '--skills "', + "Could not parse integration options: No closing quotation.", + ), + ( + "copilot", + "--missing", + "Unknown integration option '--missing'.", + ), + ( + "copilot", + "--skills=true", + "Option '--skills' is a flag and does not accept a value.", + ), + ( + "generic", + "--commands-dir", + "Option '--commands-dir' requires a value.", + ), + ( + "copilot", + "--skills stray", + "Unexpected integration option value 'stray'.", + ), + ( + "copilot", + "--skills --commands", + "--skills and --commands are mutually exclusive; pass only one.", + ), + ( + "bob", + "--skills --legacy-commands", + "--skills and --legacy-commands are mutually exclusive; pass only one.", + ), + ], +) +def test_json_integration_option_failures_preserve_parser_diagnostic( + tmp_path: Path, + integration: str, + options: str, + diagnostic: str, +): + error = _failure( + _invoke( + [ + "project", + "--json", + "--integration", + integration, + f"--integration-options={options}", + ], + cwd=tmp_path, + ), + "invalid_integration_options", + ) + + assert diagnostic in error["details"]["reason"] + assert not (tmp_path / "project").exists() + + +def test_json_init_reports_missing_required_agent_tool( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli._command_init_json as init_json + + monkeypatch.setattr(init_json, "check_tool", lambda _tool: False) + + error = _failure( + _invoke( + ["project", "--json", "--integration", "claude"], + cwd=tmp_path, + ), + "missing_agent_tool", + ) + + assert error["details"]["override_flag"] == "--ignore-agent-tools" + assert not (tmp_path / "project").exists() + + +def test_json_init_rejects_untrusted_url_before_mutation(tmp_path: Path): + error = _failure( + _invoke( + [ + "project", + "--json", + "--extension", + "https://example.com/extension.zip", + ], + cwd=tmp_path, + ), + "extension_url_trust_required", + ) + + assert error["details"]["required_flag"] == "--trust-extension-urls" + assert not (tmp_path / "project").exists() + + +def test_json_init_exposes_optional_failures( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli.command_init as init_command + from specify_cli.presets import PresetManager + + monkeypatch.setattr( + PresetManager, + "install_from_directory", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + OSError("preset install failed") + ), + ) + monkeypatch.setattr( + init_command, + "_install_extension_during_init_result", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + ValueError("extension install failed") + ), + ) + + payload = _success( + _invoke( + [ + "project", + "--json", + "--ignore-agent-tools", + "--preset", + "lean", + "--extension", + "git", + ], + cwd=tmp_path, + ) + ) + + assert payload["components"]["preset"]["status"] == "failed" + assert payload["components"]["extensions"][0]["status"] == "failed" + assert {warning["code"] for warning in payload["warnings"]} >= { + "preset_install_failed", + "extension_install_failed", + } + + +def test_json_init_rolls_back_new_target_after_fatal_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli + + monkeypatch.setattr( + specify_cli, + "_install_shared_infra_or_exit", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + typer.Exit(1) + ), + ) + project = tmp_path / "project" + + error = _failure( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "initialization_failed", + ) + + assert error["details"]["rollback"]["status"] == "completed" + assert not project.exists() + + +def test_json_init_structures_system_exit_and_reports_actual_rollback( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli + + monkeypatch.setattr( + specify_cli, + "_install_shared_infra_or_exit", + lambda *_args, **_kwargs: (_ for _ in ()).throw(SystemExit(7)), + ) + project = tmp_path / "project" + + error = _failure( + _invoke( + ["project", "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "initialization_failed", + ) + + assert error["details"]["rollback"] == { + "status": "completed", + "path": str(project), + } + assert not project.exists() + + +def test_json_init_preserves_preexisting_target_after_fatal_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli + + project = tmp_path / "project" + project.mkdir() + marker = project / "keep.txt" + marker.write_text("keep", encoding="utf-8") + monkeypatch.setattr( + specify_cli, + "_install_shared_infra_or_exit", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + typer.Exit(1) + ), + ) + + error = _failure( + _invoke( + [ + str(project), + "--json", + "--force", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ), + "initialization_failed", + ) + + assert "rollback" not in error["details"] + assert marker.read_text(encoding="utf-8") == "keep" + + +def test_json_init_sanitizes_unexpected_exception( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + from specify_cli.integrations import get_integration + + integration = get_integration("copilot") + assert integration is not None + monkeypatch.setattr( + integration, + "setup", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("sensitive detail") + ), + ) + + error = _failure( + _invoke( + ["project", "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "internal_error", + ) + + assert error["details"] == {"exception_type": "RuntimeError"} + assert "sensitive detail" not in json.dumps(error) + assert not (tmp_path / "project").exists() + + +def test_json_init_reports_cleanup_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli + import specify_cli.command_init as init_command + + monkeypatch.setattr( + specify_cli, + "_install_shared_infra_or_exit", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + typer.Exit(1) + ), + ) + monkeypatch.setattr( + init_command.shutil, + "rmtree", + lambda _path: (_ for _ in ()).throw(OSError("cleanup failed")), + ) + + error = _failure( + _invoke( + ["project", "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "rollback_failed", + ) + + assert error["details"]["cleanup_error"]["code"] == "target_cleanup_failed" + assert (tmp_path / "project").is_dir() + + +def test_json_init_preserves_target_created_during_claim_race( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + project = tmp_path / "project" + marker = project / "keep.txt" + original_mkdir = Path.mkdir + + def racing_mkdir(path: Path, *args: Any, **kwargs: Any) -> None: + if path == project and kwargs.get("exist_ok") is False: + original_mkdir(path, parents=True) + marker.write_text("keep", encoding="utf-8") + original_mkdir(path, *args, **kwargs) + + monkeypatch.setattr(Path, "mkdir", racing_mkdir) + + error = _failure( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "target_exists", + ) + + assert error["details"]["concurrent_creation"] is True + assert marker.read_text(encoding="utf-8") == "keep" + + +def test_json_init_does_not_infer_rollback_from_stale_preflight( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli._command_init_json as init_json + + project = tmp_path / "project" + marker = project / "keep.txt" + original_preflight = init_json._preflight + + def racing_preflight(**kwargs: Any) -> dict[str, Any]: + context = original_preflight(**kwargs) + project.mkdir() + marker.write_text("keep", encoding="utf-8") + return context + + monkeypatch.setattr(init_json, "_preflight", racing_preflight) + + error = _failure( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "target_exists", + ) + + assert "rollback" not in error["details"] + assert marker.read_text(encoding="utf-8") == "keep" + + +def test_json_init_reports_unverifiable_claim_without_cleanup( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli.command_init as init_command + + project = tmp_path / "project" + monkeypatch.setattr( + init_command, + "_directory_identity", + lambda _path: (_ for _ in ()).throw(OSError("identity failed")), + ) + monkeypatch.setattr( + init_command.shutil, + "rmtree", + lambda _path: (_ for _ in ()).throw( + AssertionError("unverified target must not be removed") + ), + ) + + error = _failure( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "rollback_failed", + ) + + assert error["details"]["cleanup_error"]["code"] == ( + "target_identity_unavailable" + ) + assert project.is_dir() + + +def test_json_init_preserves_replacement_target( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + from specify_cli.integrations import get_integration + + project = tmp_path / "project" + claimed = tmp_path / "claimed" + marker = project / "keep.txt" + integration = get_integration("copilot") + assert integration is not None + + def replace_target(*_args: Any, **_kwargs: Any) -> None: + project.rename(claimed) + project.mkdir() + marker.write_text("keep", encoding="utf-8") + raise RuntimeError("initialization failed") + + monkeypatch.setattr(integration, "setup", replace_target) + + error = _failure( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "rollback_failed", + ) + + assert error["details"]["cleanup_error"]["code"] == "target_identity_changed" + assert marker.read_text(encoding="utf-8") == "keep" + assert claimed.is_dir() + + +def test_json_rollback_does_not_delete_public_path_replacement( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli.command_init as init_command + from specify_cli.integrations import get_integration + + project = tmp_path / "project" + marker = project / "keep.txt" + integration = get_integration("copilot") + assert integration is not None + original_directory_identity = init_command._directory_identity + + def replace_public_path_after_staging(path: Path) -> tuple[int, int]: + identity = original_directory_identity(path) + if path.name == "target" and ".specify-rollback-" in path.parent.name: + project.mkdir() + marker.write_text("replacement", encoding="utf-8") + return identity + + monkeypatch.setattr( + init_command, + "_directory_identity", + replace_public_path_after_staging, + ) + monkeypatch.setattr( + integration, + "setup", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("initialization failed") + ), + ) + + _failure( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "internal_error", + ) + + assert marker.read_text(encoding="utf-8") == "replacement" + + +@pytest.mark.parametrize( + "args", + [ + ["project", "--json", "--unknown-option"], + ["project", "--json", "--script"], + ["project", "--json", "--integration"], + ], +) +def test_json_parser_failures_are_structured( + tmp_path: Path, + args: list[str], +): + _failure(_invoke(args, cwd=tmp_path), "invalid_arguments") + assert not (tmp_path / "project").exists() + + +@pytest.mark.parametrize( + "args", + [ + ["--ignore-agent-tools"], + [ + "--integration", + "copilot", + "--script", + "py", + "--ignore-agent-tools", + ], + [ + "--integration", + "copilot", + "--preset", + "lean", + "--extension", + "git", + "--ignore-agent-tools", + ], + [ + "--integration", + "generic", + "--integration-options=--commands-dir .agent/commands", + ], + [ + "--integration", + "copilot", + "--integration-options=--commands", + "--ignore-agent-tools", + ], + [ + "--integration", + "copilot", + "--integration-options=--skills", + "--ignore-agent-tools", + ], + [ + "--integration", + "copilot", + "--script", + "sh", + "--ignore-agent-tools", + ], + ], +) +def test_json_and_human_init_create_identical_projects( + tmp_path: Path, + args: list[str], +): + human = tmp_path / "human" + machine = tmp_path / "machine" + + human_result = _invoke( + [str(human), "--non-interactive", *args], + cwd=tmp_path, + ) + assert human_result.exit_code == 0, human_result.output + _success( + _invoke( + [str(machine), "--json", *args], + cwd=tmp_path, + ) + ) + + assert _project_snapshot(human) == _project_snapshot(machine) + + +def test_json_and_human_reinitialization_create_identical_projects( + tmp_path: Path, +): + human = tmp_path / "human" + machine = tmp_path / "machine" + initial_args = [ + "--integration", + "copilot", + "--extension", + "git", + "--ignore-agent-tools", + ] + switch_args = [ + "--force", + "--integration", + "copilot", + "--ignore-agent-tools", + ] + + assert _invoke( + [str(human), "--non-interactive", *initial_args], + cwd=tmp_path, + ).exit_code == 0 + _success(_invoke([str(machine), "--json", *initial_args], cwd=tmp_path)) + assert _invoke( + [str(human), "--non-interactive", *switch_args], + cwd=tmp_path, + ).exit_code == 0 + _success(_invoke([str(machine), "--json", *switch_args], cwd=tmp_path)) + + assert _project_snapshot(human) == _project_snapshot(machine) + + +@pytest.mark.parametrize( + "switch_args", + [ + [ + "--integration", + "generic", + "--integration-options=--commands-dir .agent/commands", + ], + [ + "--integration", + "copilot", + "--integration-options=--commands", + ], + ], +) +def test_json_rejects_incompatible_reinitialization_before_mutation( + tmp_path: Path, + switch_args: list[str], +): + project = tmp_path / "project" + _success( + _invoke( + [ + str(project), + "--json", + "--integration", + "copilot", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + before = _project_snapshot(project) + + error = _failure( + _invoke( + [ + str(project), + "--json", + "--force", + *switch_args, + "--ignore-agent-tools", + ], + cwd=tmp_path, + ), + "invalid_integration_options", + ) + + assert "integration" in error["details"] + assert _project_snapshot(project) == before + + +def test_json_rejects_generic_destination_change_before_mutation(tmp_path: Path): + project = tmp_path / "project" + _success( + _invoke( + [ + str(project), + "--json", + "--integration", + "generic", + "--integration-options=--commands-dir .agent/commands", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + before = _project_snapshot(project) + + _failure( + _invoke( + [ + str(project), + "--json", + "--force", + "--integration", + "generic", + "--integration-options=--commands-dir .other/commands", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ), + "invalid_integration_options", + ) + + assert _project_snapshot(project) == before + + +def test_human_rejects_incompatible_reinitialization_before_mutation( + tmp_path: Path, +): + project = tmp_path / "project" + first = _invoke( + [ + str(project), + "--non-interactive", + "--integration", + "copilot", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + assert first.exit_code == 0, first.output + before = _project_snapshot(project) + + result = _invoke( + [ + str(project), + "--non-interactive", + "--force", + "--integration", + "generic", + "--integration-options=--commands-dir .agent/commands", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + + assert result.exit_code == 1 + assert "integration switch generic" in result.output + assert _project_snapshot(project) == before + + +def test_force_reinitialization_persists_state_before_registration( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli + from specify_cli.integrations import _helpers + + project = tmp_path / "project" + assert ( + _invoke( + [ + str(project), + "--non-interactive", + "--integration", + "copilot", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ).exit_code + == 0 + ) + stale_options = specify_cli.load_init_options(project) + stale_options.update( + { + "ai": "generic", + "integration": "generic", + } + ) + stale_options.pop("ai_skills", None) + specify_cli.save_init_options(project, stale_options) + + observed: list[tuple[str, dict[str, Any]]] = [] + + def capture_state(project_root: Path, agent_key: str, **_kwargs: Any) -> None: + observed.append((agent_key, specify_cli.load_init_options(project_root))) + + monkeypatch.setattr(_helpers, "_register_extensions_for_agent", capture_state) + monkeypatch.setattr(_helpers, "_register_presets_for_agent", capture_state) + + result = _invoke( + [ + str(project), + "--non-interactive", + "--force", + "--integration", + "copilot", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + + assert result.exit_code == 0, result.output + assert len(observed) == 2 + for agent_key, options in observed: + assert agent_key == "copilot" + assert options["ai"] == "copilot" + assert options["integration"] == "copilot" + assert options["ai_skills"] is True + + +def test_json_empty_selections_use_and_report_defaults(tmp_path: Path): + payload = _success( + _invoke( + [ + "project", + "--json", + "--integration", + "", + "--script", + "", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert payload["integration"]["defaulted"] is True + assert payload["script"]["defaulted"] is True + + +def test_json_here_rejects_nonempty_target_without_force(tmp_path: Path): + marker = tmp_path / "keep.txt" + marker.write_text("keep", encoding="utf-8") + + _failure( + _invoke( + ["--here", "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ), + "target_not_empty", + ) + + assert marker.read_text(encoding="utf-8") == "keep" + assert not (tmp_path / ".specify").exists() + + +def test_json_rejects_target_file(tmp_path: Path): + target = tmp_path / "project" + target.write_text("keep", encoding="utf-8") + + _failure( + _invoke([str(target), "--json"], cwd=tmp_path), + "target_not_directory", + ) + + assert target.read_text(encoding="utf-8") == "keep" + + +def test_json_rejects_empty_named_target_without_force(tmp_path: Path): + target = tmp_path / "project" + target.mkdir() + + _failure( + _invoke([str(target), "--json"], cwd=tmp_path), + "target_exists", + ) + + assert not (target / ".specify").exists() + + +def test_json_reports_reinitialization_operation(tmp_path: Path): + project = tmp_path / "project" + _success( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ) + ) + + payload = _success( + _invoke( + [ + str(project), + "--json", + "--force", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert payload["project"]["operation"] == "reinitialized" + + +def test_json_invalid_default_integration_is_a_warning( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv("SPECKIT_INTEGRATION_DEFAULT", "missing") + + payload = _success( + _invoke( + ["project", "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ) + ) + + assert payload["integration"]["key"] == "copilot" + assert payload["warnings"][0]["code"] == "invalid_default_integration" + + +def test_json_reports_missing_bundled_workflow( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli.command_init as init_command + + monkeypatch.setattr(init_command, "_locate_bundled_workflow", lambda _id: None) + + payload = _success( + _invoke( + ["project", "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ) + ) + + assert payload["components"]["workflow"]["status"] == "skipped" + assert { + warning["code"] for warning in payload["warnings"] + } >= {"bundled_workflow_not_found"} + + +def test_json_reports_bundled_workflow_install_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + from specify_cli.workflows.engine import WorkflowDefinition + + monkeypatch.setattr( + WorkflowDefinition, + "from_yaml", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + OSError("workflow parse failed") + ), + ) + + payload = _success( + _invoke( + ["project", "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ) + ) + + assert payload["components"]["workflow"] == { + "id": "speckit", + "status": "failed", + "reason": "workflow parse failed", + } + warning = next( + warning + for warning in payload["warnings"] + if warning["code"] == "workflow_install_failed" + ) + assert warning["details"] == { + "workflow": "speckit", + "reason": "workflow parse failed", + } + + +def test_json_reports_missing_optional_preset( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + from specify_cli.presets import PresetCatalog + + monkeypatch.setattr( + PresetCatalog, + "get_pack_info", + lambda *_args, **_kwargs: None, + ) + + payload = _success( + _invoke( + [ + "project", + "--json", + "--preset", + "missing", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert payload["components"]["preset"]["status"] == "skipped" + assert {warning["code"] for warning in payload["warnings"]} >= { + "preset_not_found" + } + + +def test_json_reports_already_installed_extension(tmp_path: Path): + project = tmp_path / "project" + args = [ + str(project), + "--json", + "--extension", + "git", + "--ignore-agent-tools", + ] + _success(_invoke(args, cwd=tmp_path)) + + payload = _success(_invoke([*args, "--force"], cwd=tmp_path)) + + assert payload["components"]["extensions"][0]["status"] == "already_installed" + + +def test_json_extension_status_does_not_depend_on_display_message( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli.command_init as init_command + import specify_cli.events as events + + monkeypatch.setattr( + init_command, + "_install_extension_during_init_result", + lambda *_args, **_kwargs: init_command._InitExtensionInstallResult( + status="already_installed", + message="present from an earlier installation", + ), + ) + monkeypatch.setattr( + events, + "refresh_integration_events", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("already-installed extension must not refresh events") + ), + ) + + payload = _success( + _invoke( + [ + "project", + "--json", + "--extension", + "git", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert payload["components"]["extensions"][0] == { + "requested": "git", + "status": "already_installed", + "message": "present from an earlier installation", + } + + +def test_json_exposes_extension_event_refresh_failures( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli.events as events + from specify_cli.events import EventRefreshError + + monkeypatch.setattr( + events, + "refresh_integration_events", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + EventRefreshError( + [ + ("copilot", "hook write failed"), + ("claude", "settings unavailable"), + ] + ) + ), + ) + + payload = _success( + _invoke( + [ + "project", + "--json", + "--extension", + "git", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert payload["components"]["extensions"][0]["status"] == "installed" + warning = next( + warning + for warning in payload["warnings"] + if warning["code"] == "extension_event_refresh_failed" + ) + assert warning["details"]["failures"] == [ + { + "integration": "copilot", + "reason": "hook write failed", + }, + { + "integration": "claude", + "reason": "settings unavailable", + }, + ] + + +def test_json_exposes_force_reregistration_failures( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + from specify_cli.integrations import _helpers + + project = tmp_path / "project" + _success( + _invoke( + [str(project), "--json", "--ignore-agent-tools"], + cwd=tmp_path, + ) + ) + monkeypatch.setattr( + _helpers, + "_register_extensions_for_agent", + lambda *_args, **_kwargs: "extension registration failed", + ) + monkeypatch.setattr( + _helpers, + "_register_presets_for_agent", + lambda *_args, **_kwargs: "preset registration failed", + ) + + payload = _success( + _invoke( + [ + str(project), + "--json", + "--force", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + warnings = {warning["code"]: warning for warning in payload["warnings"]} + assert warnings["extension_reregistration_failed"]["details"]["reason"] == ( + "extension registration failed" + ) + assert warnings["preset_reregistration_failed"]["details"]["reason"] == ( + "preset registration failed" + ) + + +def test_json_preserves_existing_constitution(tmp_path: Path): + project = tmp_path / "project" + args = [str(project), "--json", "--ignore-agent-tools"] + _success(_invoke(args, cwd=tmp_path)) + constitution = project / ".specify" / "memory" / "constitution.md" + constitution.write_text("custom constitution", encoding="utf-8") + + payload = _success(_invoke([*args, "--force"], cwd=tmp_path)) + + assert payload["components"]["constitution"]["status"] == "preserved" + assert constitution.read_text(encoding="utf-8") == "custom constitution" + + +def test_json_explicit_url_trust_never_prompts( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + import specify_cli.command_init as init_command + + monkeypatch.setattr( + typer, + "confirm", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("trusted JSON URL must not prompt") + ), + ) + monkeypatch.setattr( + init_command, + "_install_extension_during_init_result", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + ValueError("download unavailable") + ), + ) + + payload = _success( + _invoke( + [ + "project", + "--json", + "--extension", + "https://example.com/extension.zip", + "--trust-extension-urls", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert payload["components"]["extensions"][0]["status"] == "failed" + + +def test_json_and_human_here_mode_create_identical_projects(tmp_path: Path): + human = tmp_path / "human" + machine = tmp_path / "machine" + human.mkdir() + machine.mkdir() + (human / "keep.txt").write_text("keep", encoding="utf-8") + (machine / "keep.txt").write_text("keep", encoding="utf-8") + + assert _invoke( + [ + "--here", + "--force", + "--non-interactive", + "--ignore-agent-tools", + ], + cwd=human, + ).exit_code == 0 + _success( + _invoke( + [ + "--here", + "--force", + "--json", + "--ignore-agent-tools", + ], + cwd=machine, + ) + ) + + assert _project_snapshot(human) == _project_snapshot(machine) + + +def test_json_and_human_force_merge_create_identical_projects(tmp_path: Path): + human = tmp_path / "human" + machine = tmp_path / "machine" + human.mkdir() + machine.mkdir() + (human / "keep.txt").write_text("keep", encoding="utf-8") + (machine / "keep.txt").write_text("keep", encoding="utf-8") + + assert _invoke( + [ + str(human), + "--force", + "--non-interactive", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ).exit_code == 0 + _success( + _invoke( + [ + str(machine), + "--force", + "--json", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + ) + + assert _project_snapshot(human) == _project_snapshot(machine) + + +def test_noninteractive_human_output_remains_human_readable(tmp_path: Path): + result = _invoke( + [ + "human-project", + "--non-interactive", + "--ignore-agent-tools", + ], + cwd=tmp_path, + ) + + assert result.exit_code == 0, result.output + assert "Project ready" in result.stdout + with pytest.raises(json.JSONDecodeError): + json.loads(result.stdout)