"""Agent runtime events for integrations. Provides: - ``resolve_events`` — layered event resolution (CLI flag → YAML override → extension-declared → built-in). - ``collect_extension_events`` — scan installed extension.yml files for ``events:``. - ``install_integration_events`` / ``remove_integration_events`` — entry points called from ``IntegrationBase.setup()`` / ``teardown()``. """ from __future__ import annotations import json import logging import os import re import shlex import shutil import sys import subprocess import platform from pathlib import Path from typing import TYPE_CHECKING, Any import yaml if TYPE_CHECKING: from .integrations.base import IntegrationBase from .integrations.manifest import IntegrationManifest logger = logging.getLogger(__name__) # -- Constants ------------------------------------------------------------- EVENTS_DISPATCHER_DIR = Path(".specify") EVENTS_DISPATCHER_FILENAME = "events.py" # POSIX-form (forward-slash) relative path so it matches manifest keys, which # are always stored in POSIX form (record_file/record_existing normalize via # .as_posix()). On Windows, str(Path(".specify")/"events.py") yields # ".specify\\events.py", which never matched a manifest key, so the shared- # dispatcher manifest-claim drop was skipped and uninstall(force=True) deleted # the dispatcher another integration still depended on. EVENTS_DISPATCHER_REL = (EVENTS_DISPATCHER_DIR / EVENTS_DISPATCHER_FILENAME).as_posix() YAML_OVERRIDE_FILENAME = Path(".specify") / "integration-events.yml" _SPECKIT_MARKER = "__speckit_event__" # Buffer (seconds) added to the native hook timeout so the agent's outer cap # fires after the dispatcher's inner subprocess timeout, letting the inner # kill its child cleanly instead of being killed mid-flight (which orphans # the grandchild script process). The dispatcher receives the raw seconds # (no buffer); the native config field gets seconds + buffer (R2). EVENT_TIMEOUT_BUFFER = 5 # Canonical event names (snake_case) CANONICAL_EVENTS = frozenset({ "session_start", "pre_tool_use", "post_tool_use", "session_end", "user_prompt_submit", "stop", }) # -- Events Dispatcher template --------------------------------------------- _EVENTS_DISPATCHER_TEMPLATE = '''#!/usr/bin/env python3 """Specify CLI Event Dispatcher — dispatches agent runtime events. Generated by: specify integration install/upgrade Do not edit manually. Self-contained: it prefers `specify_cli` when the package is importable (durable pip/pipx/uv-tool install) and falls back to an inline stdlib-only resolver when Spec Kit is not installed at runtime — e.g. a one-time `uvx` init whose environment is discarded after `specify init` finishes (R1). In both cases it resolves the event's command template and runs its script directly, without requiring a persistent `specify` executable on PATH. """ import json import os import re import shlex import shutil import subprocess import sys from pathlib import Path def _find_command_template(command_name, project_root): """Locate the command's .md template. Returns (path, ext_id|None).""" exts_dir = project_root / ".specify" / "extensions" disabled_ids = set() registry_file = exts_dir / ".registry" if registry_file.is_file(): try: reg_data = json.loads(registry_file.read_text(encoding="utf-8")) for ext_id, meta in reg_data.get("extensions", {}).items(): if isinstance(meta, dict) and meta.get("enabled") is False: disabled_ids.add(ext_id) except Exception: pass stem = command_name.replace("speckit.", "").replace("spec.", "") # 1. Manifest-driven resolution from extension.yml in enabled extensions (Suppressed #1) if exts_dir.is_dir(): for ext_dir in sorted(exts_dir.iterdir()): if not ext_dir.is_dir() or ext_dir.name in disabled_ids: continue ext_yml = ext_dir / "extension.yml" if ext_yml.is_file(): try: yml_text = ext_yml.read_text(encoding="utf-8") cur_name = None cur_file = None in_provides = False in_commands = False for line in yml_text.splitlines(): stripped = line.strip() if stripped == "provides:": in_provides = True continue if in_provides and stripped == "commands:": in_commands = True continue if in_commands and stripped and not line[0].isspace(): in_provides = False in_commands = False continue if in_commands: if "name:" in line: cur_name = line.split("name:", 1)[1].strip().strip('"').strip("'") if "file:" in line: cur_file = line.split("file:", 1)[1].strip().strip('"').strip("'") if cur_name and cur_file: if cur_name == command_name: candidate = ext_dir / cur_file if candidate.exists(): return candidate, ext_dir.name cur_name = None cur_file = None except Exception: pass # 2. On-disk extension commands by file stem (non-disabled extensions) if exts_dir.is_dir(): for ext_dir in sorted(exts_dir.iterdir()): if not ext_dir.is_dir() or ext_dir.name in disabled_ids: continue cmds_dir = ext_dir / "commands" if cmds_dir.is_dir(): for f in cmds_dir.glob("*.md"): if f.stem == command_name or f.stem == stem: return f, ext_dir.name # 3. Core templates in the project core = project_root / ".specify" / "templates" / "commands" if core.is_dir(): candidate = core / (stem + ".md") if candidate.exists(): return candidate, None return None, None def _script_variant(project_root): """Return the project's persisted script type ('sh'|'ps'|'py').""" default = "ps" if os.name == "nt" else "sh" init_opts = project_root / ".specify" / "init-options.json" try: data = json.loads(init_opts.read_text(encoding="utf-8")) script = data.get("script") if script in ("sh", "ps", "py"): return script except Exception: pass return default def _extract_scripts(template_path): """Parse the scripts: block from a command template's frontmatter.""" try: content = template_path.read_text(encoding="utf-8") except Exception: return {} m = re.match(r"^---\\n(.*?)\\n---", content, re.DOTALL) if not m: return {} scripts = {} in_scripts = False for line in m.group(1).splitlines(): if line.rstrip() == "scripts:": in_scripts = True continue if in_scripts and line and not line[0].isspace(): break if in_scripts and ":" in line: k, _, v = line.partition(":") scripts[k.strip()] = v.strip() return scripts def _resolve_argv(template_path, project_root, ext_id): """Resolve the command's script to a runnable argv (stdlib only).""" scripts = _extract_scripts(template_path) if not scripts: return None requested = _script_variant(project_root) order = (requested,) if requested in scripts else () fallbacks = (requested, "ps" if requested != "ps" else "sh", "py", "sh") seen = set() for cand in order + fallbacks: if cand in seen: continue seen.add(cand) if cand in scripts: variant = cand break else: return None script_cmd = scripts.get(variant, "").strip() if not script_cmd: return None base = (project_root / ".specify" / "extensions" / ext_id) if ext_id else (project_root / ".specify") try: tokens = shlex.split(script_cmd, posix=(os.name != "nt")) except ValueError: return None if not tokens: return None script_abs = base / tokens[0] if not script_abs.exists(): return None rest = tokens[1:] if variant == "py": # .py files aren't directly executable; run under the dispatcher's own # Python (sys.executable), which is always available here. return [sys.executable or "python3", str(script_abs), *rest] if variant == "ps": launcher = shutil.which("pwsh") or shutil.which("powershell") if not launcher: return None return [launcher, "-File", str(script_abs), *rest] # sh: direct on POSIX; a bash/sh launcher on Windows. if os.name == "nt": launcher = shutil.which("bash") or shutil.which("sh") if launcher: return [launcher, str(script_abs), *rest] return None return [str(script_abs), *rest] def _run_inline(command_name, payload, project_root, timeout, envelope="plain", native_event=""): """Resolve and run the event command with stdlib only (no specify_cli).""" template_path, ext_id = _find_command_template(command_name, project_root) if not template_path: return 0 # command not found: fail open (no-op) for lifecycle events argv = _resolve_argv(template_path, project_root, ext_id) if not argv: return 0 try: result = subprocess.run( argv, input=payload, capture_output=True, text=True, timeout=timeout, cwd=str(project_root), ) if result.stdout: _emit(result.stdout, envelope, native_event) if result.returncode != 0: if result.stderr: sys.stderr.write(result.stderr) return result.returncode return 0 except subprocess.TimeoutExpired: print(f"Event {command_name} timed out", file=sys.stderr) return 2 except Exception as e: print(f"Event {command_name} error: {e}", file=sys.stderr) return 2 def _emit(output, envelope, native_event=""): """Write handler output to stdout in the agent's context-injection shape. Not every agent injects a hook's plain-text stdout as model context: Gemini/Tabnine/Qwen/Devin are JSON-only protocols (plain text becomes user-facing noise, never context), Copilot discards non-JSON stdout, and Cursor parses stdout as JSON. The native hook command passes the envelope as the dispatcher's 5th argument (see events_context_envelope on the integration classes), and the native event name as the 6th argument so hookSpecificOutput can include hookEventName: hookSpecificOutput → {"hookSpecificOutput": {"hookEventName": ..., "additionalContext": ...}} additionalContext → {"additionalContext": ...} (top-level, Copilot) additional_context → {"additional_context": ...} (top-level, Cursor) suppress → emit nothing (strict-JSON agents on events whose output can't be used) plain (default) → passthrough (Claude/Codex inject plain stdout) Empty output emits nothing under any envelope (an empty additionalContext is useless noise). """ if not output: return if envelope == "suppress": return if envelope == "hookSpecificOutput": payload = {"additionalContext": output} if native_event: payload["hookEventName"] = native_event sys.stdout.write(json.dumps({"hookSpecificOutput": payload}) + "\\n") return if envelope == "additionalContext": sys.stdout.write(json.dumps({"additionalContext": output}) + "\\n") return if envelope == "additional_context": sys.stdout.write(json.dumps({"additional_context": output}) + "\\n") return sys.stdout.write(output) def main(): if len(sys.argv) < 3: sys.exit(0) command_name = sys.argv[1] # event_name is accepted for argv-compat with the native hook command but # is not needed for resolution (the command template drives everything). _event_name = sys.argv[2] # Optional 4th arg: per-handler timeout in seconds (S4). timeout = 120 if len(sys.argv) >= 4: try: timeout = int(sys.argv[3]) except (TypeError, ValueError): timeout = 120 # Optional 5th arg: context-injection envelope for stdout (C13): plain # (default), hookSpecificOutput, additionalContext, additional_context, # or suppress. Unknown values fall back to plain passthrough. envelope = sys.argv[4] if len(sys.argv) >= 5 else "plain" if envelope not in ("plain", "hookSpecificOutput", "additionalContext", "additional_context", "suppress"): envelope = "plain" # Optional 6th arg: native event name for hookSpecificOutput's # hookEventName field (required by Qwen's hooks spec; included by # Gemini/Tabnine/Devin which derive from the same protocol). native_event = sys.argv[5] if len(sys.argv) >= 6 else "" payload = sys.stdin.read() if not sys.stdin.isatty() else "{}" project_root = Path(__file__).parent.parent.resolve() # Preferred path: specify_cli is importable (durable install) — delegate to # the full resolver, which also handles extension manifests whose file stem # differs from the command name and the project's custom script selection. try: from specify_cli.events import resolve_and_run_event_command sys.exit( resolve_and_run_event_command( command_name, _event_name, payload, project_root, timeout=timeout, envelope=envelope, native_event=native_event ) ) except (ImportError, TypeError): pass # Fallback: self-contained stdlib resolver (one-time/temporary installs). sys.exit(_run_inline(command_name, payload, project_root, timeout, envelope, native_event)) if __name__ == "__main__": main() ''' # -- TS plugin template (opencode) ---------------------------------------- _TS_PLUGIN_TEMPLATE = '''import {{ execFileSync }} from 'child_process'; import * as path from 'path'; // The dispatcher + interpreter are resolved per-project at plugin load from // the `directory` OpenCode passes to the plugin factory (C8), not // process.cwd() — OpenCode may be launched from a parent directory or host // another workspace, in which case process.cwd() points at the wrong project. let DISPATCHER = ''; let INTERPRETER = ''; function canImportSpecifyCli(py: string): boolean {{ // R2: a project-local venv commonly lacks Spec Kit (installed globally or // via uv tool). Probe the interpreter can import specify_cli before // selecting it, so an unrelated venv doesn't shadow the PATH fallback. try {{ execFileSync(py, ['-c', 'import specify_cli'], {{ stdio: ['ignore', 'ignore', 'ignore'], timeout: 10000, }}); return true; }} catch (e) {{ return false; }} }} function resolveDispatcher(directory: string): void {{ DISPATCHER = path.join(directory, '.specify', 'events.py'); // Prefer a project-local venv interpreter that can import specify_cli (R2), // then fall back to a platform-appropriate PATH interpreter (S2: python on // Windows, where python3 is commonly absent; python3 on POSIX). const venvPy = path.join(directory, '.venv', 'bin', 'python'); const venvWin = path.join(directory, '.venv', 'Scripts', 'python.exe'); INTERPRETER = ( (require('fs').existsSync(venvPy) && canImportSpecifyCli(venvPy) && venvPy) || (require('fs').existsSync(venvWin) && canImportSpecifyCli(venvWin) && venvWin) || (process.platform === 'win32' ? 'python' : 'python3') ) as string; }} function runEvent(command: string, event: string, input: any, output: any, timeoutSec: number): string {{ if (!DISPATCHER) return ''; try {{ // execFileSync with an argv array invokes the interpreter directly — no // shell — so command/event strings with metacharacters can't break out // of the dispatcher argument (C9). The dispatcher arg is seconds; the // execFileSync timeout is ms with a buffer so the outer cap fires after // the dispatcher's inner subprocess (S3). stdout is captured and // returned so context-injection hooks (experimental.chat.system.transform, // chat.message) can push it into their outputs; stderr stays inherited so // dispatcher errors remain visible (C11). return execFileSync(INTERPRETER, [DISPATCHER, command, event, String(timeoutSec)], {{ input: JSON.stringify({{ input, output }}), stdio: ['pipe', 'pipe', 'inherit'], encoding: 'utf-8', timeout: (timeoutSec + {buffer}) * 1000, }}); }} catch (e) {{ // Propagate to OpenCode's hook machinery so only this hook is rejected, // not the entire host process. process.exit() would kill the agent. throw new Error(`specify event ${{command}} (${{event}}) failed: ${{(e as Error).message}}`); }} }} // Cache session_start handler output per sessionID so non-idempotent // handlers (setup, telemetry, file-mutating scripts) run once per session // instead of on every LLM request (experimental.chat.system.transform // fires per LLM turn). Evicted on session.deleted. const sessionStartCache = new Map(); {event_entries} export default (async ({{ client, project, directory, $ }}) => {{ resolveDispatcher(directory); return {{ {plugin_returns} }}; }}); ''' # -- Command runner logic (core) -------------------------------------------- def _find_command_template(command_name: str, project_root: Path) -> tuple[Path | None, str | None]: # 1. Resolve via installed extension manifests (authoritative). The # registry stores per-agent ``registered_commands`` name-lists, not a # ``{name, file}`` map, so the command→file mapping lives only in each # extension's ``extension.yml`` ``provides.commands`` (S8). Match the # command name to its declared ``file`` so commands whose file stem # differs from the command name (e.g. ``speckit.selftest.extension`` → # ``commands/selftest.md``) resolve correctly. exts_dir = project_root / ".specify" / "extensions" # S1: build the set of explicitly-disabled extension IDs so dispatch skips # disabled extensions (a stale hook would otherwise keep executing a # disabled extension's command). Applied to both the manifest loop and the # on-disk fallback below. disabled_ids = _disabled_extension_ids(project_root) try: from .extensions import ExtensionManager manager = ExtensionManager(project_root) for ext_id in sorted(manager.registry.keys()): if ext_id in disabled_ids: continue manifest = manager.get_extension(ext_id) if manifest is None: continue for cmd in manifest.commands: if not isinstance(cmd, dict): continue if cmd.get("name") == command_name and cmd.get("file"): candidate = exts_dir / ext_id / cmd["file"] if candidate.exists(): return candidate, ext_id except Exception: # Fall through to the on-disk scan if the registry/manifests can't be # read; event dispatch should degrade gracefully, not crash. pass # 2. Scan extension directories by file stem (covers extensions present on # disk but not resolvable via the manifest above). S1: skip disabled # extensions here too so the disk fallback can't re-enable them. if exts_dir.is_dir(): for ext_dir in sorted(exts_dir.iterdir()): if ext_dir.name in disabled_ids: continue cmds_dir = ext_dir / "commands" if cmds_dir.is_dir(): for f in cmds_dir.glob("*.md"): if f.stem == command_name: return f, ext_dir.name # 3. Check core templates in the project core = project_root / ".specify" / "templates" / "commands" if core.is_dir(): stem = command_name.replace("speckit.", "").replace("spec.", "") candidate = core / f"{stem}.md" if candidate.exists(): return candidate, None # 4. Fallback to package-bundled templates via the canonical asset # resolvers (wheel: core_pack/commands; source: repo-root # templates/commands). The previous bespoke inspect.getfile() math # pointed at core_pack/templates/commands, which never exists in a # wheel build (force-include maps templates/commands -> core_pack/commands). from ._assets import _locate_core_pack, _repo_root core_pack = _locate_core_pack() candidate_dirs = [ core_pack / "commands" if core_pack is not None else None, _repo_root() / "templates" / "commands", ] stem = command_name.replace("speckit.", "").replace("spec.", "") for candidate_dir in candidate_dirs: if candidate_dir is None or not candidate_dir.is_dir(): continue candidate = candidate_dir / f"{stem}.md" if candidate.exists(): return candidate, None return None, None def _resolve_event_command_argv( template_path: Path, project_root: Path, ext_id: str | None ) -> list[str] | None: """Resolve a command template's ``scripts:`` entry to a runnable argv. ``scripts:`` values are command strings (e.g. ``scripts/bash/setup-plan.sh --json``), not bare paths, so joining the whole value into a ``Path`` made ``exists()`` false and real commands silently no-op'd. This resolves the stored variant (honoring the project's sh/ps/py selection), splits the command string safely into argv, and prepends the appropriate interpreter (Python for ``.py``, the platform shell otherwise). Returns ``None`` if no runnable script is declared. """ from .integrations.base import IntegrationBase content = template_path.read_text(encoding="utf-8") m = re.match(r'^---\n(.*?)\n---', content, re.DOTALL) if not m: return None fm = m.group(1) try: fm_data = yaml.safe_load(fm) or {} except Exception: return None if not isinstance(fm_data, dict): return None scripts = fm_data.get("scripts", {}) if not isinstance(scripts, dict): return None # Determine the requested variant from the project's persisted selection, # falling back to the platform default — same logic MarkdownIntegration # uses for command scaffolding. requested = _load_project_script_type(project_root) try: variant = IntegrationBase.select_script_variant(requested, scripts) except ValueError: return None script_cmd = scripts.get(variant) if not isinstance(script_cmd, str) or not script_cmd.strip(): return None # Base under which the script's leading path component is anchored — # .specify/ (core) or .specify/extensions// (extension). All variants # share this anchoring so a `scripts/...` token resolves correctly (S2: # the py branch previously invoked build_python_invocation() on the raw # command string, leaving `scripts/...` anchored at the project root). if ext_id: base = project_root / ".specify" / "extensions" / ext_id else: base = project_root / ".specify" try: tokens = shlex.split(script_cmd, posix=(os.name != "nt")) except ValueError: # Mirror the generated dispatcher's _resolve_argv: a scripts: value # shlex cannot tokenize (e.g. an unclosed quote) declares no runnable # script, so degrade to "no argv" instead of raising. return None if not tokens: return None script_abs = base / tokens[0] if not script_abs.exists(): return None rest_args = tokens[1:] if variant == "py": # .py files aren't directly executable on Windows; prefix the resolved # interpreter. argv is passed to subprocess.run(shell=False), so no # shell quoting is needed. interpreter = IntegrationBase.resolve_python_interpreter(project_root) return [interpreter, str(script_abs), *rest_args] if variant == "ps": # PowerShell scripts cannot be executed directly by # subprocess.run(shell=False); invoke via `pwsh -File` (PowerShell 7+), # falling back to `powershell -File` (Windows PowerShell) when pwsh is # absent (S6). The default Windows script type would otherwise fail. launcher = shutil.which("pwsh") or shutil.which("powershell") or "pwsh" return [launcher, "-File", str(script_abs), *rest_args] # sh: the script is chmod'd executable during install on POSIX. On Windows # subprocess.run(shell=False) can't execute a .sh directly, so prefix a # bash/sh launcher when one is available (mirroring the ps branch's # pwsh -File handling, S5). if os.name == "nt": launcher = shutil.which("bash") or shutil.which("sh") if launcher: return [launcher, str(script_abs), *rest_args] return [str(script_abs), *rest_args] def _load_project_script_type(project_root: Path) -> str: """Return the project's persisted script type ('sh'|'ps'|'py'). Falls back to the platform default when init-options are absent or unreadable so event dispatch still works in a partially-initialized project. """ default = "ps" if platform.system().lower().startswith("win") else "sh" try: from ._init_options import load_init_options opts = load_init_options(project_root) if isinstance(opts, dict): script = opts.get("script") if isinstance(script, str) and script in ("sh", "ps", "py"): return script except Exception: pass return default def resolve_and_run_event_command( command_name: str, event_name: str, payload: str, project_root: Path, *, timeout: int = 120, envelope: str = "plain", native_event: str = "", ) -> int: """Core entry point to resolve and execute an event-driven command. *timeout* is the per-handler timeout in seconds, passed through from the native hook config via the dispatcher (S4) so a handler configured above the previous fixed 120s cap can run for its full duration. *envelope* selects how the handler's stdout is emitted for the agent's context-injection protocol (C13): ``plain`` passthrough (Claude/Codex inject plain stdout), ``hookSpecificOutput``/``additionalContext``/ ``additional_context`` JSON wrappers (Gemini/Tabnine/Qwen/Devin, Copilot, Cursor respectively), or ``suppress`` (strict-JSON agents on events whose output can't be used). *native_event* is the agent's native hookEventName (e.g. ``"SessionStart"``), required inside ``hookSpecificOutput`` by Qwen's hooks spec (and included by the Claude Code hooks spec Gemini/Tabnine/Devin derive from). Only used when *envelope* is ``hookSpecificOutput``. """ template_path, ext_id = _find_command_template(command_name, project_root) if not template_path: logger.warning("Event command '%s' not found", command_name) return 0 argv = _resolve_event_command_argv(template_path, project_root, ext_id) if not argv: logger.warning("No script found for event command '%s'", command_name) return 0 try: result = subprocess.run( argv, input=payload, capture_output=True, text=True, timeout=timeout, cwd=str(project_root), ) if result.stdout: _emit_event_stdout(result.stdout, envelope, native_event) if result.returncode != 0: if result.stderr: sys.stderr.write(result.stderr) return result.returncode return 0 except subprocess.TimeoutExpired: sys.stderr.write(f"Event command {command_name} timed out\n") return 2 except Exception as e: sys.stderr.write(f"Event command {command_name} error: {e}\n") return 2 def _emit_event_stdout(output: str, envelope: str, native_event: str = "") -> None: """Write handler stdout in the agent's context-injection shape (C13). Mirrors the ``_emit`` helper inside the generated dispatcher template; keep both in sync. Empty output emits nothing under any envelope. *native_event* is the agent's native hookEventName, required inside ``hookSpecificOutput`` by Qwen's hooks spec (and included by the Claude Code hooks spec Gemini/Tabnine/Devin derive from). """ if not output: return if envelope == "suppress": return if envelope == "hookSpecificOutput": payload = {"additionalContext": output} if native_event: payload["hookEventName"] = native_event sys.stdout.write(json.dumps({"hookSpecificOutput": payload}) + "\n") return if envelope == "additionalContext": sys.stdout.write(json.dumps({"additionalContext": output}) + "\n") return if envelope == "additional_context": sys.stdout.write(json.dumps({"additional_context": output}) + "\n") return sys.stdout.write(output) # -- Sourcing events map (CLI/Orchestration domain) ------------------------- # Resolved events map: each canonical event name maps to an *ordered list* of # handler configs. Built-in defaults and per-extension declarations both # contribute, so two extensions declaring ``session_start`` both run (finding # #2) instead of the last one silently winning. ResolvedEvents = dict[str, list[dict[str, Any]]] def _normalize_handlers(value: Any) -> list[dict[str, Any]]: """Coerce a single handler config or a list of them into a validated list. Accepts both the legacy single-mapping shape (``{command: ...}``) and the explicit list shape (``[{command: ...}, ...]``). Drops any entry that is not a mapping or lacks a ``command`` with a warning, so a malformed user override never reaches installation and crashes on ``cfg.get(...)`` (#21). """ if isinstance(value, dict): value = [value] if not isinstance(value, list): return [] handlers: list[dict[str, Any]] = [] for entry in value: if not isinstance(entry, dict): logger.warning("Skipping malformed event handler (expected a mapping): %r", entry) continue handlers.append(entry) return handlers def _validate_resolved_event(event_name: str, handlers: list[dict[str, Any]]) -> None: """Validate a resolved event's handlers, raising a user-facing error. Raised for structural problems the user must fix (unknown event name, handler missing a ``command``, or ``command`` not a non-empty string per #17). Malformed-but-skipable entries are already dropped by ``_normalize_handlers``. """ from .extensions import ValidationError if event_name not in CANONICAL_EVENTS: raise ValidationError( f"Unknown event '{event_name}': must be one of {sorted(CANONICAL_EVENTS)}" ) for handler in handlers: command = handler.get("command") if not isinstance(command, str) or not command.strip(): raise ValidationError( f"Event '{event_name}' handler missing required non-empty 'command' string" ) # C10: matcher must be a string (or absent). A non-string matcher such # as `matcher: []` passes extension validation but later crashes # by_matcher.setdefault(matcher, ...) with TypeError: unhashable type. matcher = handler.get("matcher") if matcher is not None and not isinstance(matcher, str): raise ValidationError( f"Event '{event_name}' handler has invalid 'matcher': " "must be a string" ) timeout = handler.get("timeout") if timeout is not None: if not isinstance(timeout, int) or isinstance(timeout, bool) or timeout <= 0: raise ValidationError( f"Event '{event_name}' handler has invalid 'timeout': must be a positive integer" ) def resolve_events( integration_key: str, integration_config: dict[str, Any] | None, project_root: Path, parsed_options: dict[str, Any] | None, ) -> ResolvedEvents: """Resolve the final event set for an integration. Returns a mapping of canonical event name → ordered list of handler configs. Layers (lowest → highest precedence): 1. CLI gate ``--events false`` → empty map (caller still removes prior native hooks; see ``install_integration_events``). 2. Built-in defaults from ``integration_config["events"]`` (single-config per event, wrapped as one-element lists). 3. Extension-declared ``events:`` — appended per extension so multiple extensions can declare the same event (#2). 4. User YAML override (``.specify/integration-events.yml``) — replaces the accumulated set entirely when the integration key is present. Validated (#21) before returning; a malformed override is warned about and ignored rather than crashing downstream. """ # Layer 1: CLI flag gate if parsed_options: events_flag = str(parsed_options.get("events", "true")).lower() if events_flag in ("false", "0", "no", "off"): return {} events: ResolvedEvents = {} # Layer 2: built-in defaults from integration config if integration_config and isinstance(integration_config.get("events"), dict): for ev, cfg in integration_config["events"].items(): handlers = _normalize_handlers(cfg) if handlers: events.setdefault(ev, []).extend(handlers) # Layer 3: extension-declared events (accumulated, not overwriting) for ev, handlers in collect_extension_events(project_root).items(): events.setdefault(ev, []).extend(handlers) # Layer 4: user YAML override (replaces entirely if key present) override_file = project_root / YAML_OVERRIDE_FILENAME if override_file.exists(): try: override = yaml.safe_load(override_file.read_text(encoding="utf-8")) or {} except (OSError, UnicodeError, yaml.YAMLError): logger.warning( "Could not read or parse %s; ignoring override", override_file ) override = {} integrations = override.get("integrations", {}) if isinstance(override, dict) else {} if isinstance(integrations, dict) and integration_key in integrations: key_data = integrations[integration_key] if not isinstance(key_data, dict): # C6: a non-mapping integration entry (e.g. `claude: bad`) must # not be treated as a valid explicit disable. Warn and abandon # the override, keeping the accumulated built-in + extension # layers. Only an explicitly present, mapping-valued `events` # field replaces the prior layers. logger.warning( "Override %s: entry for '%s' is not a mapping; ignoring override", override_file, integration_key, ) else: key_events = key_data.get("events", {}) if not isinstance(key_events, dict): logger.warning( "Override %s: 'events' for '%s' is not a mapping; ignoring override", override_file, integration_key, ) else: # Validate every entry before adopting the override. A single # invalid entry abandons the whole override and keeps the # accumulated built-in + extension layers (#10): previously a # typo reset resolved_override to {} and then assigned that # empty map to events, silently disabling all hooks despite # the "ignored" warning. Only a fully-valid override (including # an explicit `events: {}`) replaces the prior layers. resolved_override: ResolvedEvents = {} override_valid = True for ev, raw in key_events.items(): handlers = _normalize_handlers(raw) if not handlers: # C4: a malformed handler (e.g. `stop: []` or # `stop: bad-value`) normalizes to no handlers. # Abandon the whole override (keep prior layers) # rather than skipping the entry — otherwise an # override whose only entry is malformed silently # disabled every built-in and extension hook. An # explicit `events: {}` (no entries) remains a # valid disable. logger.warning( "Override %s: event '%s' has no valid handler; ignoring entire override", override_file, ev, ) override_valid = False break try: _validate_resolved_event(ev, handlers) except Exception as exc: logger.warning( "Override %s: invalid event '%s': %s; ignoring entire override", override_file, ev, exc, ) override_valid = False break resolved_override[ev] = handlers if override_valid: events = resolved_override # else: keep the accumulated built-in + extension layers. return events def _disabled_extension_ids(project_root: Path) -> set[str]: """Return the set of explicitly-disabled extension IDs. Extensions not tracked in the registry are treated as enabled (backward compat). Used by ``collect_extension_events`` and ``_find_command_template`` so a disabled extension's events and commands are never emitted or executed (S1) — otherwise a stale native hook would keep running a disabled extension after its config file was preserved (e.g. a JSONC parse failure that skipped native cleanup). """ from .extensions import ExtensionRegistry exts_dir = project_root / ".specify" / "extensions" disabled_ids: set[str] = set() if not exts_dir.is_dir(): return disabled_ids try: registry = ExtensionRegistry(exts_dir) for ext_id, meta in registry.list_by_priority(include_disabled=True): if not isinstance(meta, dict) or not meta.get("enabled", True): disabled_ids.add(ext_id) except Exception: pass return disabled_ids def collect_extension_events(project_root: Path) -> ResolvedEvents: """Scan all installed extensions for ``events:`` declarations. Returns a mapping of event name → list of handler configs. Multiple extensions declaring the same event each contribute a handler (in extension-directory sort order), so callers can emit all of them (#2). Honors the extension registry's ``enabled`` flag (#1): an explicitly disabled extension's events are skipped so disabling an extension actually deactivates its runtime hooks. Extensions absent from the registry (e.g. a partially-staged install) are still included to preserve the on-disk scan behavior. Events are read from a validated ``ExtensionManifest`` (R1) rather than the raw ``extension.yml`` YAML, so the command-reference canonicalization applied during install validation (C11, e.g. ``my-ext.boot`` → ``speckit.my-ext.boot``) is reflected — otherwise refresh would emit the obsolete name and ``_find_command_template`` could not match it, leaving the hook silently inert. """ from .extensions import ExtensionManager events: ResolvedEvents = {} exts_dir = project_root / ".specify" / "extensions" if not exts_dir.is_dir(): return events manager = ExtensionManager(project_root) # Build the set of explicitly-disabled extension IDs. Extensions not # tracked in the registry are treated as enabled (backward compat). disabled_ids = _disabled_extension_ids(project_root) # Union of extension IDs to consider: registry-tracked IDs (validated # manifests, canonicalized refs) plus on-disk dirs not yet in the registry # (partially-staged installs). The latter fall back to the raw YAML since # no validated manifest is available, preserving the on-disk scan behavior. registry_ids = set() try: registry_ids = set(manager.registry.keys()) except Exception: pass on_disk_ids = { d.name for d in exts_dir.iterdir() if d.is_dir() and (d / "extension.yml").exists() } for ext_id in sorted(registry_ids | on_disk_ids): if ext_id in disabled_ids: continue # Prefer the validated manifest (canonicalized command refs, R1); # fall back to the raw YAML for an on-disk extension not yet # registered (a malformed extension shouldn't abort collection). runtime: dict[str, Any] = {} if ext_id in registry_ids: try: manifest = manager.get_extension(ext_id) except Exception: manifest = None if manifest is not None: runtime = manifest.data.get("events", {}) or {} if not runtime: ext_yml = exts_dir / ext_id / "extension.yml" if not ext_yml.exists(): continue try: data = yaml.safe_load(ext_yml.read_text(encoding="utf-8")) or {} except (UnicodeDecodeError, yaml.YAMLError): continue if not isinstance(data, dict): continue runtime = data.get("events", {}) or {} if not isinstance(runtime, dict): continue for event, config in runtime.items(): handlers = _normalize_handlers(config) if handlers: events.setdefault(event, []).extend(handlers) return events # -- Writing/Merging Config (Integration domain) --------------------------- def _resolve_interpreter(project_root: Path) -> str: """Resolve a portable Python interpreter for native hook commands (#16). Delegates to ``IntegrationBase.resolve_python_interpreter`` so generated commands honor the project venv and never hard-code ``python3`` (which is commonly absent on Windows even when ``py.exe``/``python.exe`` exist). """ from .integrations.base import IntegrationBase return IntegrationBase.resolve_python_interpreter(project_root) def _resolve_interpreter_for_target(target_os: str) -> str: """Resolve a Python interpreter for a target OS, independent of the host (#S4). Copilot's native config carries both a ``bash`` (POSIX) and a ``powershell`` (Windows) variant in the same checked-in file. Resolving both with the *host* interpreter writes a Linux venv path into the PowerShell hook (or vice-versa), so the config fails on the other OS. Each variant instead gets a portable interpreter for its target shell; the dispatcher script's own ``_find_specify()`` does per-OS venv resolution at runtime. """ if target_os == "windows": # Windows: ``python`` is the most portable on PATH; the py launcher # (``py -3``) is the recommended fallback when ``python`` is absent. return "python" # POSIX (bash): ``python3`` is universally available. return "python3" def _native_timeout(integration: IntegrationBase, timeout_seconds: Any) -> int: """Return the timeout in the unit the integration's native config expects. Claude/Cursor/Codex/Copilot measure timeouts in seconds; Gemini measures in milliseconds (#7). An integration declares its unit via ``events_timeout_unit`` (``"s"`` default, ``"ms"`` for Gemini). """ try: seconds = int(timeout_seconds) except (TypeError, ValueError): seconds = 60 if getattr(integration, "events_timeout_unit", "s") == "ms": return seconds * 1000 return seconds def _shell_quote(value: str, target_os: str) -> str: """Quote *value* as one argument for the target shell (R2). ``host`` and ``posix`` targets use ``shlex.quote`` (POSIX shells). Safe tokens — ``python3``, ``speckit.ext.cmd`` — pass through bare, so a single-``command``-string hook (Claude/Gemini/etc.) stays invocable on every platform. ``windows`` targets use a PowerShell single-quoted literal with embedded quotes doubled, for Copilot's dedicated ``powershell`` field. Prevents a component containing spaces (e.g. a venv interpreter path under a directory with spaces) or shell metacharacters (a malformed extension/override ``command``) from breaking the hook or being interpreted by the native shell instead of passed as one dispatcher argument. """ if target_os == "windows": return "'" + value.replace("'", "''") + "'" # "host" and "posix" both use POSIX quoting. On Windows the single- # command-string formats (Claude/Gemini/Qwen/Devin/Tabnine) are run via # Git Bash or the agent's POSIX-ish shell, so POSIX quoting is correct and # avoids emitting 'python' (which PowerShell wouldn't invoke without &). return shlex.quote(value) def _dispatcher_command( integration: IntegrationBase, project_root: Path, command_name: str, event_name: str, *, target_os: str = "host", timeout_seconds: Any = None, ) -> str: """Build the single shell command string that invokes the dispatcher (#6). Claude/Gemini/Qwen/Devin/Tabnine accept one ``command`` string (not a ``command``+``args`` split), so each adapter renders a complete invocation: `` []``. The interpreter is resolved portably (#16); Claude's dispatcher path is prefixed with ``${CLAUDE_PROJECT_DIR}/`` (Claude expands it before shell execution). ``target_os`` selects an OS-appropriate interpreter for adapters that emit both POSIX and Windows variants into one checked-in file (Copilot): ``host`` uses the host-resolved interpreter (venv-aware), while ``posix``/``windows`` emit portable interpreters so the config works on either OS (#S4). Each component is shell-quoted for the target shell (R2) so an interpreter path with spaces or a command/event containing shell metacharacters is passed as a single argument rather than reinterpreted by the native shell. The Claude dispatcher is double-quoted (``"${CLAUDE_PROJECT_DIR}/..."``) so the variable still expands but a project path with spaces doesn't word-split (C2). For the explicit ``windows`` target (Copilot's powershell field) the quoted interpreter is prefixed with PowerShell's call operator ``&`` so the quoted command is actually invoked (C1). When *timeout_seconds* is given, the resolved timeout (in the integration's native unit) is appended as a 4th argument so the dispatcher and inner runner honor the per-handler timeout instead of a fixed 120s cap that would kill a handler configured for longer (S4). When omitted, a default of 60s is emitted so the positional argument order (command event timeout envelope native_event) stays aligned — otherwise the envelope would land in the timeout slot and the dispatcher would silently fall back to plain stdout. When the integration declares a context-injection envelope for this canonical event (``events_context_envelope``, C13), the envelope token is appended as a 5th argument so the dispatcher wraps stdout in the JSON shape the agent's hook protocol requires. Plain-passthrough agents (Claude/Codex) declare no envelope and get no extra argument. """ if target_os == "host": interpreter = _resolve_interpreter(project_root) else: interpreter = _resolve_interpreter_for_target(target_os) q_interp = _shell_quote(interpreter, target_os) q_command = _shell_quote(command_name, target_os) q_event = _shell_quote(event_name, target_os) if integration.key == "claude": # C2: double-quote so ${CLAUDE_PROJECT_DIR} still expands (double # quotes allow variable expansion in POSIX shells) but a project path # containing spaces doesn't word-split. dispatcher = '"${CLAUDE_PROJECT_DIR}/' + EVENTS_DISPATCHER_REL + '"' else: dispatcher = _shell_quote(EVENTS_DISPATCHER_REL, target_os) # C1: PowerShell won't invoke a single-quoted command without the call # operator. Prefix & for the explicit windows target only. prefix = "& " if target_os == "windows" else "" base = f"{prefix}{q_interp} {dispatcher} {q_command} {q_event}" # Always emit the timeout (4th positional arg) so the dispatcher's argv # parsing stays aligned when an envelope (5th) or native_event (6th) # follows. Without it the envelope would land in the timeout slot and # the dispatcher would fall back to plain stdout (R3). resolved_timeout = 60 if timeout_seconds is None else int(timeout_seconds) # R2: the dispatcher interprets this argument as seconds, so pass the # raw seconds — NOT _native_timeout(...) (which converts to ms for # Gemini/Qwen/Tabnine and would yield 60000 seconds). The buffer is # applied to the native hook timeout field (in the adapter formatters) # so the agent's outer cap fires after the inner subprocess timeout. base += f" {_shell_quote(str(resolved_timeout), target_os)}" envelope = _context_envelope_for(integration, event_name) if envelope: base += f" {_shell_quote(envelope, target_os)}" # hookSpecificOutput requires the native hookEventName inside the # envelope (Qwen's hooks spec marks it mandatory; the Claude Code # hooks spec that Gemini/Tabnine/Devin derive from includes it). # Append the native event name as a 6th dispatcher argument so the # dispatcher can populate hookEventName in the JSON output. if envelope == "hookSpecificOutput": native_event = getattr(integration, "CANONICAL_TO_NATIVE", {}).get(event_name, "") if native_event: base += f" {_shell_quote(native_event, target_os)}" return base def _context_envelope_for(integration: IntegrationBase, canonical_event: str) -> str | None: """Resolve the context-injection envelope for an integration + event (C13). The event key wins; ``"*"`` is the fallback. Returns ``None`` when the integration declares no envelope for the event (plain stdout passthrough). """ mapping = getattr(integration, "events_context_envelope", None) or {} if canonical_event in mapping: return mapping[canonical_event] return mapping.get("*") def install_integration_events( integration: IntegrationBase, project_root: Path, manifest: IntegrationManifest, events: ResolvedEvents, ) -> list[Path]: """Generate dispatcher, merge native config, return created files. ``events`` maps each canonical event to an ordered list of handler configs (#2); every handler is emitted as a separate native hook entry so two extensions declaring ``session_start`` both run. """ canonical_to_native = getattr(integration, "CANONICAL_TO_NATIVE", {}) if not canonical_to_native: return [] # Filter to only supported events, preserving all handlers per event. filtered: ResolvedEvents = {} for ev, handlers in events.items(): if not isinstance(handlers, list): continue if ev in canonical_to_native: filtered[ev] = handlers else: print( f"\u26a0\ufe0f {integration.key} does not support '{ev}' events; skipping", file=sys.stderr, ) # #3: an empty resolved map (--events false, or override disabling events) # must still strip prior Specify hooks from this integration's native # config rather than leaving them active. S3: also run the shared- # dispatcher refcount cleanup so an --events false upgrade of the last # event integration doesn't orphan .specify/events.py permanently (the # new manifest no longer claims it and stale cleanup excludes it). if not filtered: _remove_native_event_hooks(integration, project_root, manifest) _cleanup_shared_dispatcher(integration, project_root, manifest) return [] created: list[Path] = [] # 1. Generate events.py dispatcher script (#12: validate destination first) dispatcher_dir = project_root / EVENTS_DISPATCHER_DIR dispatcher_path = dispatcher_dir / EVENTS_DISPATCHER_FILENAME _ensure_safe_destination(dispatcher_path) dispatcher_dir.mkdir(parents=True, exist_ok=True) dispatcher_path.write_text(_EVENTS_DISPATCHER_TEMPLATE, encoding="utf-8") dispatcher_path.chmod(0o755) manifest.record_file( str(dispatcher_path.relative_to(project_root)), dispatcher_path.read_bytes(), ) created.append(dispatcher_path) # 2. Format-specific merge/write fmt = getattr(integration, "events_format", "json-nested") config_file = getattr(integration, "events_config_file", None) if not config_file: return created config_path = project_root / config_file if fmt == "ts-plugin": # Opencode TS plugin custom merge plugin_rel = ".opencode/plugin/speckit-events.ts" plugin_path = project_root / plugin_rel _ensure_safe_destination(plugin_path) plugin_path.parent.mkdir(parents=True, exist_ok=True) plugin_path.write_text( _build_opencode_plugin(filtered, canonical_to_native), encoding="utf-8", ) manifest.record_file( plugin_rel, plugin_path.read_bytes(), ) created.append(plugin_path) # Merge plugin path into opencode.json. S5: only track the config # file when the merge actually wrote; a skipped merge (JSONC/malformed) # must not be tracked or manifest.uninstall() would later delete the # user's untouched file. if _merge_opencode_plugin_ref(config_path, f"./{plugin_rel}"): rel = str(config_path.relative_to(project_root)) if rel not in manifest.files: manifest.record_existing(rel) created.append(config_path) elif fmt == "copilot-json": # Copilot dedicated .github/hooks/speckit.json. Each handler becomes # its own entry in the native event's list (#2). The bash and # powershell variants get independent OS-targeted interpreters (#S4) # so a config generated on Linux doesn't write a POSIX venv path into # the PowerShell hook (and vice-versa). Entries carry the ownership # marker so a pre-existing user-authored file is merged (owned entries # replaced) rather than overwritten (#8), and teardown removes only # owned entries. copilot_hooks: dict[str, list[dict[str, Any]]] = {} for ev, handlers in filtered.items(): native = canonical_to_native[ev] entries: list[dict[str, Any]] = [] for cfg in handlers: command = cfg.get("command", "") bash_cmd = _dispatcher_command( integration, project_root, command, ev, target_os="posix", timeout_seconds=cfg.get("timeout", 60), ) ps_cmd = _dispatcher_command( integration, project_root, command, ev, target_os="windows", timeout_seconds=cfg.get("timeout", 60), ) entries.append( { "type": "command", "bash": bash_cmd, "powershell": ps_cmd, "timeoutSec": _native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER), _SPECKIT_MARKER: True, } ) copilot_hooks[native] = entries # S5: only track when the merge wrote (skips on JSONC/malformed). if _merge_copilot_json(config_path, copilot_hooks): rel = str(config_path.relative_to(project_root)) if rel not in manifest.files: manifest.record_existing(rel) created.append(config_path) elif fmt == "toml": # Codex config.toml custom merge. One [[hooks..hooks]] block # per handler so multiple handlers per event all emit (#2). lines: list[str] = [] for ev, handlers in filtered.items(): native = canonical_to_native[ev] for cfg in handlers: command = cfg.get("command", "") dispatcher_cmd = _dispatcher_command(integration, project_root, command, ev, timeout_seconds=cfg.get("timeout", 60)) lines.append(f'[[hooks.{native}]]') lines.append(f'matcher = {_toml_quote(str(cfg.get("matcher", "*")))}') lines.append('') lines.append(f'[[hooks.{native}.hooks]]') lines.append('type = "command"') lines.append(f'command = {_toml_quote(dispatcher_cmd)}') lines.append(f'timeout = {_native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER)}') lines.append('speckit_marker = true') lines.append('') # S5: only track when the merge wrote (skips on unreadable file). if _merge_toml_fragment(config_path, "\n".join(lines)): rel = str(config_path.relative_to(project_root)) if rel not in manifest.files: manifest.record_existing(rel) created.append(config_path) elif fmt == "json-flat": # Cursor hooks.json custom merge. Flat command-string entries, one # per handler (#2), single resolved command string (#6/#16). cursor_hooks: dict[str, list[dict[str, Any]]] = {} for ev, handlers in filtered.items(): native = canonical_to_native[ev] entries: list[dict[str, Any]] = [] for cfg in handlers: command = cfg.get("command", "") dispatcher_cmd = _dispatcher_command(integration, project_root, command, ev, timeout_seconds=cfg.get("timeout", 60)) entries.append( { "command": dispatcher_cmd, "type": "command", "timeout": _native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER), "matcher": cfg.get("matcher", "*"), _SPECKIT_MARKER: True, } ) cursor_hooks[native] = entries # #7: Cursor's .cursor/hooks.json schema requires top-level # "version": 1; ensure it (preserving a user's value if present). # S5: only track when the merge wrote (skips on JSONC/malformed). if _merge_json_fragment(config_path, cursor_hooks, version=1): rel = str(config_path.relative_to(project_root)) if rel not in manifest.files: manifest.record_existing(rel) created.append(config_path) elif fmt == "json-nested": # Claude/Qwen/Gemini/Devin/Tabnine nested config JSON merge. # Native schema is a single ``command`` string per hook (not # command+args), so each handler renders one complete dispatcher # invocation (#6). Gemini timeouts are converted to ms (#7). # Handlers are grouped by distinct matcher so each matcher gets its # own matcher-group (S3); previously all handlers were placed under # the first handler's matcher, so two extensions registering the same # event with different matchers both ran for the first matcher and # neither for the later. nested_hooks: dict[str, list[dict[str, Any]]] = {} for ev, handlers in filtered.items(): native = canonical_to_native[ev] by_matcher: dict[str, list[dict[str, Any]]] = {} for cfg in handlers: matcher = cfg.get("matcher", "*") command = cfg.get("command", "") dispatcher_cmd = _dispatcher_command(integration, project_root, command, ev, timeout_seconds=cfg.get("timeout", 60)) by_matcher.setdefault(matcher, []).append( { "type": "command", "command": dispatcher_cmd, "timeout": _native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER), _SPECKIT_MARKER: True, } ) nested_hooks[native] = [ {"matcher": matcher, "hooks": inner} for matcher, inner in by_matcher.items() ] # S5: only track when the merge wrote (skips on JSONC/malformed). if _merge_json_fragment(config_path, nested_hooks): rel = str(config_path.relative_to(project_root)) if rel not in manifest.files: manifest.record_existing(rel) created.append(config_path) elif fmt == "json-root-nested": # Devin hooks.v1.json: a root event map ({"PreToolUse": [...]}) with # no top-level "hooks" wrapper (U2). Same matcher-grouping and single # command string as json-nested, but written to the root. root_hooks: dict[str, list[dict[str, Any]]] = {} for ev, handlers in filtered.items(): native = canonical_to_native[ev] by_matcher: dict[str, list[dict[str, Any]]] = {} for cfg in handlers: matcher = cfg.get("matcher", "*") command = cfg.get("command", "") dispatcher_cmd = _dispatcher_command(integration, project_root, command, ev, timeout_seconds=cfg.get("timeout", 60)) by_matcher.setdefault(matcher, []).append( { "type": "command", "command": dispatcher_cmd, "timeout": _native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER), _SPECKIT_MARKER: True, } ) root_hooks[native] = [ {"matcher": matcher, "hooks": inner} for matcher, inner in by_matcher.items() ] if _merge_json_root(config_path, root_hooks): rel = str(config_path.relative_to(project_root)) if rel not in manifest.files: manifest.record_existing(rel) created.append(config_path) return created def _remove_native_event_hooks( integration: IntegrationBase, project_root: Path, manifest: IntegrationManifest, ) -> None: """Remove Specify-authored hooks from *this* integration's native config. Used both by full teardown and by the empty-resolved-map install path (#3). Does NOT touch the shared dispatcher (another integration may still reference it — #10). """ fmt = getattr(integration, "events_format", None) config_file = getattr(integration, "events_config_file", None) if not config_file: return config_path = project_root / config_file if not config_path.exists(): return _ensure_safe_destination(config_path) if fmt == "copilot-json": _remove_copilot_entries(config_path) elif fmt == "toml": _remove_toml_entries(config_path) elif fmt in ("json-nested", "json-flat"): _remove_json_entries(config_path) elif fmt == "json-root-nested": _remove_json_root_entries(config_path) elif fmt == "ts-plugin": _remove_opencode_entries(config_path) # Always drop this integration's manifest claim on the native config, # whether the file was deleted or retained with user content (S9). If we # kept a retained file tracked, teardown()'s manifest.uninstall(force=True) # would delete the entire user-owned settings file. After cleanup the file # is either gone or contains only user content, so this integration must # no longer claim it for teardown purposes. manifest.remove(config_file) def _other_event_integrations_reference_dispatcher( project_root: Path, excluding_key: str ) -> bool: """Return True if another installed event-capable integration still references the shared ``.specify/events.py`` dispatcher (#10). Inspects each installed integration's manifest (excluding *excluding_key*) for the dispatcher path so uninstalling one multi-install event-capable integration doesn't delete the dispatcher the others still rely on. """ from .integrations._helpers import _read_integration_json from .integrations.manifest import IntegrationManifest from .integration_state import installed_integration_keys state = _read_integration_json(project_root) for key in installed_integration_keys(state): if key == excluding_key: continue try: manifest = IntegrationManifest.load(key, project_root) except Exception: continue if EVENTS_DISPATCHER_REL in manifest.files: return True return False def _cleanup_shared_dispatcher( integration: IntegrationBase, project_root: Path, manifest: IntegrationManifest ) -> None: """Drop this integration's manifest claim on the shared dispatcher and delete the file only when no other installed event-capable integration still references it (#10, S3). The manifest.remove() runs in both branches (S1): if we retain the file but leave it tracked, the subsequent manifest.uninstall() in teardown() sees the matching hash and deletes the file another integration still depends on. Used by full teardown and by the empty-resolved-map install path so an ``--events false`` upgrade of the last event integration doesn't orphan ``.specify/events.py`` permanently (S3). """ dispatcher_rel = EVENTS_DISPATCHER_REL # Drop this integration's manifest claim if present. The remove() is # conditional (S1): an upgrade passes a *fresh* manifest that may never # have claimed the dispatcher, so the key may be absent — that's a no-op. if dispatcher_rel in manifest.files: manifest.remove(dispatcher_rel) # S2: run the no-other-references deletion independently of whether the # new manifest currently contains the key. An ``integration upgrade # --events false`` passes a fresh manifest that never recorded the # dispatcher, so gating the deletion on its presence orphans the file the # old on-disk manifest owned — and stale cleanup excludes it (C3). If no # other installed event-capable integration references the dispatcher, # delete it; otherwise leave it for them. if not _other_event_integrations_reference_dispatcher(project_root, integration.key): dispatcher_path = project_root / dispatcher_rel if dispatcher_path.exists(): _ensure_safe_destination(dispatcher_path) dispatcher_path.unlink(missing_ok=True) def remove_integration_events( integration: IntegrationBase, project_root: Path, manifest: IntegrationManifest ) -> None: """Remove Specify-authored event entries from native config. The shared ``.specify/events.py`` dispatcher is deleted only when no other installed event-capable integration still references it (#10); otherwise it is left in place so multi-install setups don't lose the dispatcher mid-stream. """ _remove_native_event_hooks(integration, project_root, manifest) _cleanup_shared_dispatcher(integration, project_root, manifest) # Clean up opencode TS plugin (owned solely by the opencode integration). if integration.key == "opencode": plugin_rel = ".opencode/plugin/speckit-events.ts" if plugin_rel in manifest.files: plugin_path = project_root / plugin_rel if plugin_path.exists(): _ensure_safe_destination(plugin_path) plugin_path.unlink(missing_ok=True) manifest.remove(plugin_rel) def events_stale_exclusions(integration_key: str) -> set[str]: """Return project-relative paths to protect from stale cleanup.""" from .integrations import get_integration integration = get_integration(integration_key) if not integration: return set() exclusions = set() config_file = getattr(integration, "events_config_file", None) if config_file: exclusions.add(config_file) if integration_key == "opencode": exclusions.add(".opencode/plugin/speckit-events.ts") # C3: the shared dispatcher is written into every event-capable # integration's manifest but is reference-counted across them. An upgrade # with --events false omits events.py from the new manifest, so the generic # stale pass would delete it without the refcount check, breaking any other # installed event-capable integration. Protect it here; its deletion is # left to remove_integration_events(), which checks the refcount. exclusions.add(EVENTS_DISPATCHER_REL) return exclusions class EventRefreshError(RuntimeError): """Raised when refreshing one or more integrations' event config failed. Aggregates per-integration failures so a lifecycle command (extension add/remove/enable/disable) can surface that an extension was not fully deactivated — a stale native hook may still be active (R3). """ def __init__(self, failures: list[tuple[str, str]]) -> None: self.failures = failures details = "; ".join(f"{key}: {detail}" for key, detail in failures) super().__init__( f"event refresh failed for {len(failures)} integration(s): {details}" ) def refresh_integration_events(project_root: Path) -> None: """Re-resolve and re-emit native event config for every installed event-capable integration (#1). Called after extension state changes (install/uninstall/enable/disable) so that extension-declared events are regenerated in each installed integration's native config — otherwise the documented install-after- ``specify init`` flow is inert and disabled/removed extension events stay active. Each integration is refreshed independently; a failure for one is logged and accumulated but does not abort the others. If any integration failed, :class:`EventRefreshError` is raised at the end so the lifecycle command can't claim the extension was fully deactivated while a stale native hook may still be active (R3). """ from .integrations import get_integration from .integrations._helpers import _read_integration_json, _resolve_integration_options from .integrations.manifest import IntegrationManifest from .integration_state import installed_integration_keys state = _read_integration_json(project_root) failures: list[tuple[str, str]] = [] for key in installed_integration_keys(state): integration = get_integration(key) if integration is None or not integration.supports_events(): continue try: manifest = IntegrationManifest.load(key, project_root) except Exception as exc: logger.warning("Could not load manifest for '%s'; skipping event refresh: %s", key, exc) failures.append((key, f"manifest load: {exc}")) continue try: # C12: resolve first, then call install_integration_events once. # The previous flow ran _remove_native_event_hooks *before* # resolution, so any later failure (invalid destination, write # error, formatter error) destroyed the working native config # before the new one was written. install_integration_events # already removes stale Specify-marked entries and handles an # empty map (stripping prior hooks), so the destructive pre-step # is both unsafe and redundant. # S7: resolve this integration's persisted parsed_options so a # stored --events false is honored across extension lifecycle # changes; passing None would re-enable events the user disabled. _, parsed_options = _resolve_integration_options(integration, state, key, None) events_map = resolve_events( key, integration.config, project_root, parsed_options ) # install_integration_events handles both the populated case # (writes new config, stripping stale owned entries) and the empty # case (strips prior hooks for --events false / disabled override). install_integration_events(integration, project_root, manifest, events_map) manifest.save() except Exception as exc: logger.warning("Failed to refresh events for '%s': %s", key, exc) failures.append((key, str(exc))) if failures: raise EventRefreshError(failures) # -- Manifest validation --------------------------------------------------- def validate_events(data: dict[str, Any]) -> None: """Validate ``events`` field in extension manifest data.""" from .extensions import ValidationError events = data.get("events") if "events" in data and not isinstance(events, dict): raise ValidationError("Invalid events: expected a mapping") if events: for event_name, event_config in events.items(): if not isinstance(event_config, dict): raise ValidationError( f"Invalid event '{event_name}': expected a mapping" ) command = event_config.get("command") # #17: command must be a non-empty string. A truthy non-string # (e.g. command: [foo]) would pass a bare truthiness check and # later render into invalid native configuration. if not isinstance(command, str) or not command.strip(): raise ValidationError( f"Event '{event_name}' missing required 'command' string" ) if event_name not in CANONICAL_EVENTS: raise ValidationError( f"Unknown event '{event_name}': " f"must be one of {sorted(CANONICAL_EVENTS)}" ) # C10: matcher must be a string (or absent). A non-string matcher # such as `matcher: []` would later crash by_matcher.setdefault. matcher = event_config.get("matcher") if matcher is not None and not isinstance(matcher, str): raise ValidationError( f"Event '{event_name}' has invalid 'matcher': must be a string" ) timeout = event_config.get("timeout") if timeout is not None: if not isinstance(timeout, int) or isinstance(timeout, bool) or timeout <= 0: raise ValidationError( f"Event '{event_name}' has invalid 'timeout': must be a positive integer" ) def has_events(data: dict[str, Any]) -> bool: """Return True if ``events`` is present and non-empty.""" return bool(data.get("events")) # -- Helper merging functions ---------------------------------------------- def _toml_quote(value: str) -> str: """Render *value* as a TOML basic string via the shared escaper.""" from ._toml_string import escape_toml_basic return escape_toml_basic(value) def _build_opencode_plugin( filtered_events: ResolvedEvents, canonical_to_native: dict[str, str], ) -> str: """Render the opencode TS plugin for the resolved event set. Each canonical event may carry multiple handlers (#2); all handlers for a native event are invoked from one generated function. The dispatcher and interpreter are resolved per-project at plugin load from the ``directory`` OpenCode passes (C8); the dispatcher is launched with ``execFileSync`` and an argv array (C9). Both the ``input`` and ``output`` callback arguments are forwarded to ``runEvent`` (C7) so pre_tool_use can inspect tool arguments and post_tool_use can inspect the result. Context-injection natives get dedicated hook bodies: for ``experimental.chat.system.transform`` the handlers' concatenated stdout is pushed into ``output.system`` (system-prompt injection, re-applied per LLM request so the context survives compaction); for ``chat.message`` it is pushed as a synthetic text part on the user message (C11). Other natives keep their side-effect behavior (tool.execute.* args mutation; session.* lifecycle events via the generic ``event`` hook). """ event_entries: list[str] = [] plugin_returns: list[str] = [] event_handlers: list[str] = [] for ev, handlers in filtered_events.items(): native = canonical_to_native[ev] # S1: serialize every interpolated value as a JSON string literal so a # quote/backslash/backtick in a command or matcher can't break the # generated TypeScript or inject code. json.dumps produces a valid # TS/JS string literal (double-quoted, fully escaped). ev_lit = json.dumps(ev) native_lit = json.dumps(native) is_injection = native in ("experimental.chat.system.transform", "chat.message") # Build the body: one runEvent() call per handler wrapped in try/catch, # forwarding both input and output (C7). An optional tool-name matcher # guard applies to tool.execute.* hooks. All handlers execute before # any aggregate error is thrown. Injection hooks additionally collect # each handler's stdout and return the concatenation. body_lines: list[str] = [" const errors: string[] = [];"] if is_injection: body_lines.append(" const contexts: string[] = [];") for cfg in handlers: command = str(cfg.get("command", "")) command_lit = json.dumps(command) matcher = cfg.get("matcher", "*") # S3: thread the per-handler timeout (seconds) to runEvent so the # execFileSync cap and dispatcher arg match the configuration # instead of a fixed 60000ms / 120s. timeout_sec = int(cfg.get("timeout", 60)) if native.startswith("tool.execute."): if matcher and matcher != "*": tools = [t.strip().strip('"') for t in matcher.split("|")] checks = " || ".join( f"input.tool === {json.dumps(t.lower())}" for t in tools ) body_lines.append( f" try {{ if ({checks}) {{ runEvent({command_lit}, {ev_lit}, input, output, {timeout_sec}); }} }} catch (e) {{ errors.push((e as Error).message); }}" ) else: body_lines.append( f" try {{ runEvent({command_lit}, {ev_lit}, input, output, {timeout_sec}); }} catch (e) {{ errors.push((e as Error).message); }}" ) elif is_injection: body_lines.append( f" try {{ const ctx = runEvent({command_lit}, {ev_lit}, input, output, {timeout_sec}); if (ctx) contexts.push(ctx); }} catch (e) {{ errors.push((e as Error).message); }}" ) else: body_lines.append( f" try {{ runEvent({command_lit}, {ev_lit}, input, output, {timeout_sec}); }} catch (e) {{ errors.push((e as Error).message); }}" ) body_lines.append(" if (errors.length > 0) { throw new Error(errors.join('; ')); }") if native.startswith("tool.execute."): ts_hook = native event_entries.append( f"function _{ev}(input: any, output: any) {{\n" + "\n".join(body_lines) + "\n" " }" ) plugin_returns.append( f" {json.dumps(ts_hook)}: async (input: any, output: any) => {{\n" f" _{ev}(input, output);\n" f" }}," ) elif native == "experimental.chat.system.transform": body_lines.append(' return contexts.join("\\n\\n");') event_entries.append( f"function _{ev}(input: any, output: any): string {{\n" + "\n".join(body_lines) + "\n" " }" ) # OpenCode fires experimental.chat.system.transform for non-session # operations (e.g. agent generation) with no sessionID. Guard so # canonical session_start handlers only run when a session is # present, preventing their output from being injected into # internal prompts. Cache the handler output per sessionID so # non-idempotent handlers (setup, telemetry, file-mutating # scripts) execute once per session instead of on every LLM # request; the cache is evicted on session.deleted. plugin_returns.append( f" {native_lit}: async (input: any, output: any) => {{\n" f" if (!input.sessionID) return;\n" f" let ctx = sessionStartCache.get(input.sessionID);\n" f" if (ctx === undefined) {{\n" f" ctx = _{ev}(input, output);\n" f" sessionStartCache.set(input.sessionID, ctx ?? \"\");\n" f" }}\n" f" if (ctx) output.system.push(ctx);\n" f" }}," ) elif native == "chat.message": body_lines.append(' return contexts.join("\\n\\n");') event_entries.append( f"function _{ev}(input: any, output: any): string {{\n" + "\n".join(body_lines) + "\n" " }" ) # Part id must start with "prt" (opencode's Identifier brand): an # invalid id fails the user-part schema validation and crashes the # whole session (C12). Derive it from the last existing part so the # brand survives an opencode prefix change, falling back to "prt_" # when output.parts is empty. plugin_returns.append( f" {native_lit}: async (input: any, output: any) => {{\n" f" const ctx = _{ev}(input, output);\n" f" if (!ctx) return;\n" f" const base = output.parts[output.parts.length - 1]?.id ?? \"prt_\" + Date.now().toString(36) + Math.random().toString(36).slice(2, 10);\n" f" output.parts.push({{ id: base + \".speckit\" + Math.random().toString(36).slice(2, 8), sessionID: input.sessionID, messageID: output.message.id, type: \"text\", text: ctx, synthetic: true }});\n" f" }}," ) else: event_entries.append( f"function _{ev}(input: any, output: any) {{\n" + "\n".join(body_lines) + "\n" " }" ) # Evict the sessionStartCache when the session is deleted so the # cache doesn't grow unbounded across sessions. eviction = "" if native == "session.deleted": eviction = "if (event.sessionID) sessionStartCache.delete(event.sessionID); " event_handlers.append( f" if (event.type === {native_lit}) {{ {eviction}_{ev}(event, event); }}" ) if event_handlers: plugin_returns.append( " event: async ({ event }) => {\n" + "\n".join(event_handlers) + "\n" " }," ) return _TS_PLUGIN_TEMPLATE.format( buffer=EVENT_TIMEOUT_BUFFER, event_entries="\n\n".join(event_entries), plugin_returns="\n".join(plugin_returns), ) def _merge_opencode_plugin_ref(config_path: Path, ref: str) -> bool: """Merge the speckit-events plugin ref into opencode.json. Aborts with a warning (#23) when the file cannot be parsed (e.g. JSONC or malformed JSON) instead of resetting user configuration to ``{}``. Returns False when skipped so callers avoid tracking the untouched file (S5). """ existing = _load_user_json(config_path) if existing is None: return False plugins = existing.get("plugin", []) if not isinstance(plugins, list): plugins = [] if ref not in plugins: plugins.append(ref) existing["plugin"] = plugins _safe_write_json(config_path, existing) return True def _remove_opencode_entries(config_path: Path) -> bool: """Remove the speckit-events plugin ref from opencode.json (#23). Returns True if the file was deleted (now empty of user content), False otherwise. Aborts without writing when the file cannot be parsed. """ _ensure_safe_destination(config_path) existing = _load_user_json(config_path) if existing is None: return False plugins = existing.get("plugin", []) if isinstance(plugins, list): ref = "./.opencode/plugin/speckit-events.ts" plugins = [p for p in plugins if p != ref] if plugins: existing["plugin"] = plugins else: existing.pop("plugin", None) if not existing: config_path.unlink(missing_ok=True) return True _safe_write_json(config_path, existing) return False def _merge_toml_fragment(dst: Path, fragment: str) -> bool: """Merge Specify-owned TOML entries into *dst*, regenerating the file. An unreadable or undecodable pre-existing file aborts the merge instead of discarding the user's bytes, mirroring ``_load_user_json`` (#22). Returns False when skipped so callers avoid tracking the untouched file (S5). """ _ensure_safe_destination(dst) existing = "" if dst.exists(): try: existing = dst.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError) as exc: logger.warning( "Could not read %s (it may be unreadable or not UTF-8); " "skipping event-config merge to preserve user content.", dst, ) logger.debug("Read error detail: %s", exc) return False existing = re.sub( r'\[\[hooks\.\w+\]\]\n(?:(?!\[\[hooks\.\w+\]\]).)*?speckit_marker = true\n*', "", existing, flags=re.DOTALL, ) dst.parent.mkdir(parents=True, exist_ok=True) dst.write_text(existing.rstrip() + "\n\n" + fragment + "\n", encoding="utf-8") return True def _remove_toml_entries(dst: Path) -> bool: """Remove Specify-marked TOML entries; delete the file if now empty (#14). Returns True if the file was deleted (no user content remained). """ if not dst.exists(): return False # R3: validate the destination before reading/writing so a symlink swap of # the config after install can't make teardown overwrite a file outside # the project (the merge/write path already validates; teardown must too). _ensure_safe_destination(dst) try: existing = dst.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError) as exc: # An unreadable or undecodable file is left untouched rather than # crashing teardown — it contains only user content as far as we can # tell, and the caller drops the manifest claim either way (S9). logger.warning( "Could not read %s (it may be unreadable or not UTF-8); " "skipping event-config cleanup to preserve user content.", dst, ) logger.debug("Read error detail: %s", exc) return False cleaned = re.sub( r'\[\[hooks\.\w+\]\]\n(?:(?!\[\[hooks\.\w+\]\]).)*?speckit_marker = true\n*', "", existing, flags=re.DOTALL, ) # If only whitespace/comments remain, the file had no user content — # delete it rather than leaving an empty stub that confuses uninstall. stripped = "\n".join( line for line in cleaned.splitlines() if line.strip() and not line.strip().startswith("#") ) if not stripped: dst.unlink(missing_ok=True) return True dst.write_text(cleaned, encoding="utf-8") return False def _merge_copilot_json(dst: Path, new_hooks: dict[str, list]) -> bool: """Merge Specify-owned hooks into Copilot's dedicated hooks JSON (#8). A pre-existing user-authored ``.github/hooks/speckit.json`` is merged (owned entries replaced via markers) rather than overwritten, and a parse failure aborts instead of resetting user content (#22). Returns False when skipped so callers avoid tracking the untouched file (S5). """ existing = _load_user_json(dst) if existing is None: return False if not isinstance(existing, dict): existing = {} existing.setdefault("version", 1) existing_hooks = existing.get("hooks", {}) if not isinstance(existing_hooks, dict): existing_hooks = {} # #11: strip ALL Specify-marked entries from every event first. cleaned_hooks: dict[str, list] = {} for event, entries in existing_hooks.items(): if not isinstance(entries, list): continue kept_entries = _drop_marked_entries(entries) if kept_entries: cleaned_hooks[event] = kept_entries for event, entries in new_hooks.items(): cleaned_hooks.setdefault(event, []).extend(entries) if cleaned_hooks: existing["hooks"] = cleaned_hooks else: existing.pop("hooks", None) _safe_write_json(dst, existing) return True def _remove_copilot_entries(dst: Path) -> bool: """Remove Specify-owned hooks from Copilot's hooks JSON (#8, #14). Deletes the file when no user-authored hooks remain; otherwise keeps the file with user content. Aborts (no write) on parse failure (#22). """ _ensure_safe_destination(dst) existing = _load_user_json(dst) if existing is None: return False if not isinstance(existing, dict): return False hooks = existing.get("hooks", {}) if not isinstance(hooks, dict): hooks = {} cleaned: dict[str, list] = {} for event, entries in hooks.items(): if not isinstance(entries, list): continue kept_entries = _drop_marked_entries(entries) if kept_entries: cleaned[event] = kept_entries if cleaned: existing["hooks"] = cleaned else: existing.pop("hooks", None) # Dedicated Spec-Kit file: delete when only the (Spec-Kit-invented) # ``version`` key would remain — no user content to preserve. user_keys = {k for k in existing if k != "version"} if not user_keys: dst.unlink(missing_ok=True) return True _safe_write_json(dst, existing) return False def _merge_json_fragment(dst: Path, new_hooks: dict, *, version: int | None = None) -> bool: """Merge Specify-authored hook entries into a native JSON config. Idempotent: removes ALL prior Specify-marked entries from every event in the existing config first (#11), so an override that drops an event (e.g. ``pre_tool_use`` → ``stop``) doesn't leave stale marked entries behind. Marker detection recurses into nested ``hooks`` arrays (#9) so a matcher-group containing Specify-owned inner hooks is recognized and replaced rather than duplicated on every upgrade. Aborts with a warning (no write) when the existing file cannot be parsed (#22) — e.g. JSONC with comments — instead of resetting user content to ``{}``. Returns False when the merge was skipped so callers avoid tracking the untouched file (S5: otherwise manifest.uninstall() later deletes the user's JSONC/malformed file). When *version* is given, the top-level ``version`` field is ensured (preserving a user's value if present) so formats that require it — e.g. Cursor's ``.cursor/hooks.json`` schema (``version: 1``) — stay valid on a freshly generated file (#7). """ existing = _load_user_json(dst) if existing is None: return False if not isinstance(existing, dict): existing = {} if version is not None: existing.setdefault("version", version) hooks_key = "hooks" existing_hooks = existing.get(hooks_key, {}) if not isinstance(existing_hooks, dict): existing_hooks = {} # #11: strip ALL Specify-marked entries from every event first. cleaned_hooks: dict[str, list] = {} for event, entries in existing_hooks.items(): if not isinstance(entries, list): continue kept_entries = _drop_marked_entries(entries) if kept_entries: cleaned_hooks[event] = kept_entries # Then add the newly resolved set. for event, entries in new_hooks.items(): cleaned_hooks.setdefault(event, []).extend(entries) if cleaned_hooks: existing[hooks_key] = cleaned_hooks else: existing.pop(hooks_key, None) _safe_write_json(dst, existing) return True def _merge_json_root(dst: Path, new_hooks: dict) -> bool: """Merge Specify-authored hooks into a root-nested JSON config (Devin U2). Devin's ``.devin/hooks.v1.json`` is a root event map (``{"PreToolUse": [...]}``) with no ``hooks`` wrapper, so the event keys are top-level. Same idempotent strip-all-marked-then-add semantics and JSONC-abort behavior as ``_merge_json_fragment``. """ existing = _load_user_json(dst) if existing is None: return False if not isinstance(existing, dict): existing = {} # #11: strip ALL Specify-marked entries from every root event first. cleaned: dict[str, list] = {} for event, entries in existing.items(): if not isinstance(entries, list): # Preserve non-list user fields at the root (Devin has none, but # be defensive against a mixed user file). cleaned[event] = entries # type: ignore[assignment] continue kept_entries = _drop_marked_entries(entries) if kept_entries: cleaned[event] = kept_entries # Then add the newly resolved set (list values only). for event, entries in new_hooks.items(): cleaned.setdefault(event, []).extend(entries) if cleaned: existing = cleaned else: existing = {} if not existing: dst.unlink(missing_ok=True) return True _safe_write_json(dst, existing) return True def _remove_json_root_entries(dst: Path) -> bool: """Remove Specify-authored entries from a root-nested JSON config (Devin U2). Deletes the file when no user content remains (C5/#14 mirror). """ _ensure_safe_destination(dst) existing = _load_user_json(dst) if existing is None: return False if not isinstance(existing, dict): return False cleaned: dict[str, Any] = {} for event, entries in existing.items(): if not isinstance(entries, list): cleaned[event] = entries continue kept_entries = _drop_marked_entries(entries) if kept_entries: cleaned[event] = kept_entries if not cleaned: dst.unlink(missing_ok=True) return True _safe_write_json(dst, cleaned) return False def _drop_marked_entries(entries: list) -> list: """Return *entries* with Specify-marked hooks removed, preserving user hooks. Handles both flat entries (marker on the entry itself) and nested entries (marker on inner ``hooks`` elements). A nested matcher-group whose inner hooks are all Specify-owned is dropped; one with surviving user inner hooks is kept with only the user hooks retained (#9). """ kept: list = [] for entry in entries: if not isinstance(entry, dict): kept.append(entry) continue inner = entry.get("hooks") if isinstance(inner, list): kept_inner = [h for h in inner if not _has_marker(h)] if kept_inner: entry["hooks"] = kept_inner kept.append(entry) # else: outer group was entirely Specify-owned → drop elif _has_marker(entry): pass # flat Specify-owned entry → drop else: kept.append(entry) return kept def _load_user_json(path: Path) -> dict | None: """Load a user-owned JSON file, aborting (None) on read/parse failure (#22/#23). Returns the parsed dict, or ``None`` when the file is missing or cannot be read or parsed (e.g. JSONC with comments, a temporarily malformed JSON document, or an unreadable path). Callers must skip the merge rather than resetting user content to ``{}``. """ if not path.exists(): return {} try: data = json.loads(path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError, ValueError) as exc: logger.warning( "Could not read or parse %s (it may be unreadable, contain JSONC " "comments, or be malformed); " "skipping event-config merge to preserve user content.", path, ) logger.debug("Read/parse error detail: %s", exc) return None if not isinstance(data, dict): logger.warning("%s is not a JSON object; skipping event-config merge.", path) return None return data def _safe_write_json(dst: Path, data: dict) -> None: """Write *data* as JSON to *dst* after validating the destination (#12).""" _ensure_safe_destination(dst) dst.parent.mkdir(parents=True, exist_ok=True) dst.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") def _ensure_safe_destination(dst: Path) -> None: """Validate a write target is a regular path inside the project (#12). Walks each path component and rejects symlinks (which could escape the project — e.g. a symlinked ``.claude`` or ``.specify`` directory pointing outside the repo would redirect writes to external files). Then validates lexical containment so ``..`` traversal is also rejected. """ from .agents import CommandRegistrar # Walk each component so a symlinked ancestor (e.g. ``.claude`` → outside) # cannot be silently followed. Mirrors IntegrationManifest.record_existing. walked = dst.anchor and Path(dst.anchor) or Path("/") for part in dst.relative_to(dst.anchor).parts if dst.anchor else dst.parts: walked = walked / part if walked.is_symlink(): raise ValueError( f"Refusing to write event config through a symlink: {walked}" ) # Containment check against the nearest existing ancestor directory. base = dst.parent while not base.exists() and base != base.parent: base = base.parent CommandRegistrar._ensure_inside(dst, base) def _remove_json_entries(dst: Path) -> bool: """Remove Specify-authored entries; delete the file if now empty (#14). Returns True if the file was deleted (Spec Kit created it and no user content remains), False otherwise. """ _ensure_safe_destination(dst) existing = _load_user_json(dst) if existing is None: return False hooks = existing.get("hooks", {}) if not isinstance(hooks, dict): return False cleaned: dict[str, list] = {} for event, entries in hooks.items(): if not isinstance(entries, list): continue kept_entries = _drop_marked_entries(entries) if kept_entries: cleaned[event] = kept_entries if cleaned: existing["hooks"] = cleaned else: existing.pop("hooks", None) # #14/C5: if the config is now empty of user content, delete the file # rather than leaving a stub that confuses manifest.uninstall(). A # Spec-Kit-created Cursor file retains {"version": 1} after all owned # hooks are removed (we added the version field); treat the version-only # case as empty too, mirroring _remove_copilot_entries, so clean teardown # doesn't leave a generated stub behind. user_keys = {k for k in existing if k != "version"} if not user_keys: dst.unlink(missing_ok=True) return True _safe_write_json(dst, existing) return False def _has_marker(entry: Any) -> bool: """Return True if *entry* (or any nested inner hook) is Specify-marked (#9). Flat entries carry the marker directly; nested matcher-groups carry it on their inner ``hooks`` elements, so detection recurses one level to recognize groups that are (wholly or partly) Specify-owned. """ if not isinstance(entry, dict): return False if entry.get(_SPECKIT_MARKER, False) is True: return True inner = entry.get("hooks") if isinstance(inner, list): return any(_has_marker(h) for h in inner) return False