diff --git a/src/specify_cli/integrations/_hooks.py b/src/specify_cli/integrations/_hooks.py new file mode 100644 index 0000000000..7f24395b34 --- /dev/null +++ b/src/specify_cli/integrations/_hooks.py @@ -0,0 +1,975 @@ +"""Agent runtime hooks for integrations. + +Provides: +- ``resolve_hooks`` — layered hook resolution (CLI flag → YAML override → extension-declared → built-in). +- ``collect_extension_runtime_hooks`` — scan installed extension.yml files for ``runtime_hooks:``. +- ``HookAdapter`` / adapters — generate and merge native hook config per agent CLI. +- ``install_integration_hooks`` / ``remove_integration_hooks`` — entry points called from ``IntegrationBase.setup()`` / ``teardown()``. + +""" + +from __future__ import annotations + +import json +import logging +import re +from abc import ABC, abstractmethod +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import yaml + +if TYPE_CHECKING: + from .base import IntegrationBase + from .manifest import IntegrationManifest + +logger = logging.getLogger(__name__) + +# -- Constants ------------------------------------------------------------- + +HOOK_BRIDGE_DIR = Path(".specify") / "hooks" +HOOK_BRIDGE_FILENAME = "bridge.py" +HOOK_BRIDGE_REL = str(HOOK_BRIDGE_DIR / HOOK_BRIDGE_FILENAME) + +YAML_OVERRIDE_FILENAME = Path(".specify") / "integration-hooks.yml" + +# Sentinel marker embedded in generated native config entries so we can +# identify and remove only our entries on uninstall/upgrade. +_SPECKIT_MARKER = "__speckit_hook__" + +# Canonical runtime hook event names that extensions use in ``runtime_hooks:``. +# Adapters translate these to each agent's native event names via +# ``CANONICAL_TO_NATIVE``. Extensions never need to know agent-specific names. +CANONICAL_RUNTIME_EVENTS = frozenset({ + "PreToolUse", + "PostToolUse", + "Stop", + "SessionStart", + "SessionEnd", + "UserPromptSubmit", +}) + +# -- Bridge script template ------------------------------------------------ + +_BRIDGE_TEMPLATE = '''#!/usr/bin/env python3 +"""Specify CLI Hook Bridge — dispatches agent runtime hooks to slash commands. + +Generated by: specify integration install/upgrade +Do not edit manually. Regenerate with: specify integration upgrade +""" +import json, sys, subprocess, re, os, platform +from pathlib import Path + + +def _project_root(): + for env_var in ("CLAUDE_PROJECT_DIR", "CURSOR_PROJECT_ROOT", "PROJECT_ROOT"): + val = os.environ.get(env_var) + if val and Path(val).is_dir(): + return Path(val) + return Path.cwd() + + +def _find_command_template(command_name, project_root): + # 1. Check extension .registry + registry = project_root / ".specify" / "extensions" / ".registry" + if registry.exists(): + try: + data = json.loads(registry.read_text(encoding="utf-8")) + except Exception: + data = {} + for ext_id, meta in data.items(): + if not isinstance(meta, dict): + continue + for cmd in meta.get("commands", []): + if cmd.get("name") == command_name: + ext_dir = project_root / ".specify" / "extensions" / ext_id + return ext_dir / cmd["file"], ext_id + # 2. Scan extension directories + exts_dir = project_root / ".specify" / "extensions" + if exts_dir.is_dir(): + for ext_dir in sorted(exts_dir.iterdir()): + 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 + 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 + return None, None + + +def _extract_script_path(template_path, project_root, ext_id): + 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) + scripts_m = re.search(r'scripts:\\s*\\n((?:\\s+\\w+:.*\\n)*)', fm) + if not scripts_m: + return None + variants = {} + for line in scripts_m.group(1).strip().splitlines(): + kv = re.match(r'\\s+(\\w+):\\s*(.+)', line) + if kv: + variants[kv.group(1)] = kv.group(2).strip().strip('"\\'') + default = "ps" if platform.system().lower().startswith("win") else "sh" + script_rel = variants.get(default) or variants.get("sh") or variants.get("py") + if not script_rel: + return None + if ext_id: + script_abs = project_root / ".specify" / "extensions" / ext_id / script_rel + else: + script_abs = project_root / ".specify" / script_rel + return str(script_abs) if script_abs.exists() else None + + +def main(): + if len(sys.argv) < 3: + sys.exit(0) + command_name = sys.argv[1] + event_name = sys.argv[2] + project_root = _project_root() + payload = sys.stdin.read() if not sys.stdin.isatty() else "{}" + template, ext_id = _find_command_template(command_name, project_root) + if not template: + sys.exit(0) + script_path = _extract_script_path(template, project_root, ext_id) + if not script_path: + sys.exit(0) + try: + result = subprocess.run( + [script_path], input=payload, capture_output=True, text=True, timeout=120, + ) + if result.stdout: + print(result.stdout, end="") + if result.returncode != 0: + if result.stderr: + print(result.stderr, file=sys.stderr, end="") + sys.exit(2) + sys.exit(0) + except subprocess.TimeoutExpired: + print(f"Hook {command_name} timed out", file=sys.stderr) + sys.exit(2) + except Exception as e: + print(f"Hook {command_name} error: {e}", file=sys.stderr) + sys.exit(2) + + +if __name__ == "__main__": + main() +''' + +# -- TS plugin template (opencode) ---------------------------------------- + +_TS_PLUGIN_TEMPLATE = '''import {{ execSync }} from 'child_process'; +import * as path from 'path'; + +const BRIDGE = path.join(process.cwd(), '.specify', 'hooks', 'bridge.py'); + +function runHook(command: string, event: string, input: any): void {{ + try {{ + execSync(`python3 ${{BRIDGE}} ${{command}} ${{event}}`, {{ + input: JSON.stringify(input), + stdio: ['pipe', 'inherit', 'inherit'], + timeout: 60000, + }}); + }} catch (e) {{ + process.exit(2); + }} +}} + +{hook_entries} + +export default (async ({{ client, project, directory, $ }}) => {{ + return {{ +{plugin_returns} + }}; +}}); +''' + + +# -- Hook resolution ------------------------------------------------------- + +def resolve_hooks( + integration_key: str, + integration_config: dict[str, Any] | None, + project_root: Path, + parsed_options: dict[str, Any] | None, +) -> dict[str, dict[str, Any]]: + """Resolve the final hook set for an integration. + + Layer 1: ``--hooks false`` → return ``{}`` + Layer 2: ``.specify/integration-hooks.yml`` → replace entirely if key present + Layer 3: extension-declared ``runtime_hooks:`` → appended + Layer 4: ``config["hooks"]`` → built-in baseline + """ + # Layer 1: CLI flag gate + if parsed_options: + hooks_flag = str(parsed_options.get("hooks", "true")).lower() + if hooks_flag in ("false", "0", "no", "off"): + return {} + + # Layer 4: built-in defaults from integration config + hooks: dict[str, dict[str, Any]] = {} + if integration_config and isinstance(integration_config.get("hooks"), dict): + hooks = dict(integration_config["hooks"]) + + # Layer 3: extension-declared runtime hooks + ext_hooks = collect_extension_runtime_hooks(project_root) + hooks.update(ext_hooks) + + # Layer 2: 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 yaml.YAMLError: + logger.warning("Could not parse %s; ignoring override", override_file) + override = {} + integrations = override.get("integrations", {}) if isinstance(override, dict) else {} + if integration_key in integrations: + key_data = integrations[integration_key] + if isinstance(key_data, dict): + hooks = key_data.get("hooks", {}) or {} + else: + hooks = {} + + return hooks + + +def collect_extension_runtime_hooks(project_root: Path) -> dict[str, dict[str, Any]]: + """Scan all installed extensions for ``runtime_hooks:`` declarations.""" + hooks: dict[str, dict[str, Any]] = {} + exts_dir = project_root / ".specify" / "extensions" + if not exts_dir.is_dir(): + return hooks + + for ext_dir in sorted(exts_dir.iterdir()): + if not ext_dir.is_dir(): + continue + ext_yml = ext_dir / "extension.yml" + if not ext_yml.exists(): + continue + try: + data = yaml.safe_load(ext_yml.read_text(encoding="utf-8")) or {} + except yaml.YAMLError: + continue + if not isinstance(data, dict): + continue + runtime = data.get("runtime_hooks", {}) or {} + if not isinstance(runtime, dict): + continue + for event, config in runtime.items(): + if isinstance(config, dict): + hooks[event] = config + return hooks + + +# -- Adapter infrastructure ------------------------------------------------ + +class HookAdapter(ABC): + """Abstract base for per-agent native hook config generation/merge. + + Subclasses set ``CANONICAL_TO_NATIVE`` to map canonical event names + (e.g. ``PreToolUse``) to the agent's native event names. Events not + in the mapping are silently skipped with a warning at install time. + """ + + # Maps canonical event name → agent-native event name. + # Empty dict means no events are supported (adapter is a no-op). + CANONICAL_TO_NATIVE: dict[str, str] = {} + + @property + @abstractmethod + def config_file_rel(self) -> str: + """Project-relative path to the native settings file.""" + + @abstractmethod + def generate_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> tuple[str, Any]: + """Return (format, fragment) where format is 'json' or 'toml'.""" + + @abstractmethod + def merge_fragment(self, dst: Path, fragment: Any, *, format: str) -> None: + """Merge fragment into the native config file at *dst*.""" + + @abstractmethod + def remove_entries(self, dst: Path) -> None: + """Remove all Specify-authored hook entries from *dst*.""" + + def _native_event(self, canonical: str) -> str | None: + """Return the native event name for a canonical name, or None.""" + return self.CANONICAL_TO_NATIVE.get(canonical) + + def _filter_hooks(self, hooks: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]: + """Filter hooks to only those supported by this adapter. + + Prints a warning for each unsupported event and returns a dict + containing only the supported ones. + """ + import sys + filtered: dict[str, dict[str, Any]] = {} + for event, config in hooks.items(): + if self._native_event(event) is not None: + filtered[event] = config + else: + adapter_name = type(self).__name__ + print( + f"\u26a0\ufe0f {adapter_name} does not support '{event}' events; skipping", + file=sys.stderr, + ) + return filtered + + def install( + self, + project_root: Path, + manifest: "IntegrationManifest", + hooks: dict[str, dict[str, Any]], + ) -> list[Path]: + """Generate bridge, merge native config, return created files.""" + # Filter to only events this adapter supports + hooks = self._filter_hooks(hooks) + if not hooks: + return [] + + created: list[Path] = [] + + # Generate bridge script + bridge_dir = project_root / HOOK_BRIDGE_DIR + bridge_dir.mkdir(parents=True, exist_ok=True) + bridge_path = bridge_dir / HOOK_BRIDGE_FILENAME + bridge_path.write_text(_BRIDGE_TEMPLATE, encoding="utf-8") + bridge_path.chmod(0o755) + manifest.record_file( + str(bridge_path.relative_to(project_root)), + bridge_path.read_bytes(), + ) + created.append(bridge_path) + + # Generate / merge native config + config_path = project_root / self.config_file_rel + fmt, fragment = self.generate_fragment(hooks, project_root) + self.merge_fragment(config_path, fragment, format=fmt) + + # Record or mark as existing + if config_path.exists(): + 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(self, project_root: Path, manifest: "IntegrationManifest") -> None: + """Remove Specify-authored entries from native config.""" + config_path = project_root / self.config_file_rel + if config_path.exists(): + self.remove_entries(config_path) + + +class JSONHookAdapter(HookAdapter): + """Base for agents using JSON settings files. + + Subclasses set ``config_file_rel``, ``CANONICAL_TO_NATIVE``, and + optionally ``bridge_path_prefix`` (e.g. ``${CLAUDE_PROJECT_DIR}/``). + The default ``_build_fragment()`` produces Claude-style nested + matcher-groups. Override for different JSON structures (e.g. Cursor's + flat format). + """ + + config_file_rel: str = "" + bridge_path_prefix: str = "" + + def generate_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> tuple[str, Any]: + fragment = self._build_fragment(hooks, project_root) + return "json", fragment + + @abstractmethod + def _build_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> dict: + ... + + def _build_nested_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> dict: + """Build Claude-style nested matcher-groups fragment. + + Shared by Claude, Qwen, Devin, Gemini, Tabnine — all use the same + nested ``hooks.{Event}[].{matcher, hooks[]}`` structure. + """ + result: dict[str, list] = {"hooks": {}} + for event, config in hooks.items(): + native = self._native_event(event) + if native is None: + continue + matcher = config.get("matcher", "*") + timeout = config.get("timeout", 60) + command = config.get("command", "") + entry = { + "type": "command", + "command": "python3", + "args": [ + self.bridge_path_prefix + HOOK_BRIDGE_REL, + command, + event, + ], + "timeout": timeout, + _SPECKIT_MARKER: True, + } + result["hooks"][native] = [{ + "matcher": matcher, + "hooks": [entry], + }] + return result + + def merge_fragment(self, dst: Path, fragment: Any, *, format: str) -> None: + if format != "json": + return + existing: dict = {} + if dst.exists(): + try: + existing = json.loads(dst.read_text(encoding="utf-8")) + if not isinstance(existing, dict): + existing = {} + except (json.JSONDecodeError, OSError): + logger.warning("Could not parse %s; skipping merge", dst) + return + + # Merge: only touch the "hooks" key + existing_hooks = existing.get("hooks", {}) + our_hooks = fragment.get("hooks", {}) + + for event, entries in our_hooks.items(): + existing_list = existing_hooks.get(event, []) + if not isinstance(existing_list, list): + existing_list = [] + # Remove our old entries for this event (by marker), then append new + existing_list = [e for e in existing_list if not _has_marker(e)] + existing_list.extend(entries) + existing_hooks[event] = existing_list + + existing["hooks"] = existing_hooks + dst.parent.mkdir(parents=True, exist_ok=True) + dst.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8") + + def remove_entries(self, dst: Path) -> None: + try: + existing = json.loads(dst.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return + if not isinstance(existing, dict): + return + hooks = existing.get("hooks", {}) + if not isinstance(hooks, dict): + return + cleaned: dict[str, list] = {} + for event, entries in hooks.items(): + if not isinstance(entries, list): + continue + kept_entries = [] + for entry in entries: + if not isinstance(entry, dict): + kept_entries.append(entry) + continue + # Claude nested format: entry has "hooks" list + 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_entries.append(entry) + # else: drop the entire matcher-group (all our hooks) + elif _has_marker(entry): + pass # drop this flat entry (Cursor format) + else: + kept_entries.append(entry) + if kept_entries: + cleaned[event] = kept_entries + if cleaned: + existing["hooks"] = cleaned + else: + existing.pop("hooks", None) + dst.parent.mkdir(parents=True, exist_ok=True) + dst.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8") + + +class ClaudeHookAdapter(JSONHookAdapter): + """Claude Code: nested matcher-groups in .claude/settings.json.""" + + config_file_rel = ".claude/settings.json" + bridge_path_prefix = "${CLAUDE_PROJECT_DIR}/" + CANONICAL_TO_NATIVE = { + "PreToolUse": "PreToolUse", + "PostToolUse": "PostToolUse", + "Stop": "Stop", + "SessionStart": "SessionStart", + "SessionEnd": "SessionEnd", + "UserPromptSubmit": "UserPromptSubmit", + } + + def _build_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> dict: + return self._build_nested_fragment(hooks, project_root) + + +class CursorHookAdapter(JSONHookAdapter): + """Cursor: flat handler arrays in .cursor/hooks.json.""" + + config_file_rel = ".cursor/hooks.json" + CANONICAL_TO_NATIVE = { + "PreToolUse": "preToolUse", + "PostToolUse": "postToolUse", + "Stop": "stop", + "SessionStart": "sessionStart", + "SessionEnd": "sessionEnd", + "UserPromptSubmit": "beforeSubmitPrompt", + } + + def _build_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> dict: + result: dict[str, list] = {"hooks": {}} + for event, config in hooks.items(): + native = self._native_event(event) + if native is None: + continue + matcher = config.get("matcher", "*") + timeout = config.get("timeout", 60) + command = config.get("command", "") + entry = { + "command": f"python3 {HOOK_BRIDGE_REL} {command} {event}", + "type": "command", + "timeout": timeout, + "matcher": matcher, + _SPECKIT_MARKER: True, + } + result["hooks"][native] = [entry] + return result + + +class CodexHookAdapter(HookAdapter): + """Codex CLI: TOML [[hooks.*]] in config.toml.""" + + config_file_rel = ".codex/config.toml" + CANONICAL_TO_NATIVE = { + "PreToolUse": "PreToolUse", + "PostToolUse": "PostToolUse", + "Stop": "Stop", + "SessionStart": "SessionStart", + "SessionEnd": "SessionEnd", + "UserPromptSubmit": "UserPromptSubmit", + } + + def generate_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> tuple[str, Any]: + lines: list[str] = [] + for event, config in hooks.items(): + native = self._native_event(event) + if native is None: + continue + matcher = config.get("matcher", "*") + timeout = config.get("timeout", 60) + command = config.get("command", "") + bridge_abs = f"$(git rev-parse --show-toplevel)/{HOOK_BRIDGE_REL}" + lines.append(f'[[hooks.{native}]]') + lines.append(f'matcher = "{matcher}"') + lines.append('') + lines.append(f'[[hooks.{native}.hooks]]') + lines.append('type = "command"') + lines.append(f'command = \'python3 {bridge_abs} {command} {event}\'') + lines.append(f'timeout = {timeout}') + lines.append('speckit_marker = true') + lines.append('') + return "toml", "\n".join(lines) + + def merge_fragment(self, dst: Path, fragment: Any, *, format: str) -> None: + if format != "toml": + return + existing = "" + if dst.exists(): + existing = dst.read_text(encoding="utf-8") + # Remove old speckit-marked blocks + existing = re.sub( + r'\[\[hooks\.\w+\]\]\n(?:(?!\[\[hooks\.\w+\]\]).)*?speckit_marker = true\n*', + "", + existing, + flags=re.DOTALL, + ) + # Append new + dst.parent.mkdir(parents=True, exist_ok=True) + dst.write_text(existing.rstrip() + "\n\n" + fragment + "\n", encoding="utf-8") + + def remove_entries(self, dst: Path) -> None: + if not dst.exists(): + return + existing = dst.read_text(encoding="utf-8") + cleaned = re.sub( + r'\[\[hooks\.\w+\]\]\n(?:(?!\[\[hooks\.\w+\]\]).)*?speckit_marker = true\n*', + "", + existing, + flags=re.DOTALL, + ) + dst.write_text(cleaned, encoding="utf-8") + + +class OpencodeHookAdapter(HookAdapter): + """opencode: TypeScript plugin in .opencode/plugin/ + opencode.json merge.""" + + config_file_rel = "opencode.json" + _PLUGIN_REL = ".opencode/plugin/speckit-hooks.ts" + CANONICAL_TO_NATIVE = { + "PreToolUse": "tool.execute.before", + "PostToolUse": "tool.execute.after", + } + + def install( + self, + project_root: Path, + manifest: "IntegrationManifest", + hooks: dict[str, dict[str, Any]], + ) -> list[Path]: + # Filter to only events this adapter supports + hooks = self._filter_hooks(hooks) + if not hooks: + return [] + + created: list[Path] = [] + + # Generate bridge script + bridge_dir = project_root / HOOK_BRIDGE_DIR + bridge_dir.mkdir(parents=True, exist_ok=True) + bridge_path = bridge_dir / HOOK_BRIDGE_FILENAME + bridge_path.write_text(_BRIDGE_TEMPLATE, encoding="utf-8") + bridge_path.chmod(0o755) + manifest.record_file( + str(bridge_path.relative_to(project_root)), + bridge_path.read_bytes(), + ) + created.append(bridge_path) + + # Generate TS plugin + plugin_path = project_root / self._PLUGIN_REL + plugin_path.parent.mkdir(parents=True, exist_ok=True) + plugin_path.write_text(self._build_plugin(hooks), encoding="utf-8") + manifest.record_file( + str(plugin_path.relative_to(project_root)), + plugin_path.read_bytes(), + ) + created.append(plugin_path) + + # Merge plugin path into opencode.json + config_path = project_root / self.config_file_rel + self._merge_plugin_ref(config_path) + 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 generate_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> tuple[str, Any]: + return "skip", None + + def merge_fragment(self, dst: Path, fragment: Any, *, format: str) -> None: + pass + + def remove_entries(self, dst: Path) -> None: + if not dst.exists(): + return + try: + existing = json.loads(dst.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + return + if not isinstance(existing, dict): + return + plugins = existing.get("plugin", []) + if isinstance(plugins, list): + plugins = [p for p in plugins if p != f"./{self._PLUGIN_REL}"] + if plugins: + existing["plugin"] = plugins + else: + existing.pop("plugin", None) + dst.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8") + + def _build_plugin(self, hooks: dict[str, dict[str, Any]]) -> str: + hook_entries: list[str] = [] + plugin_returns: list[str] = [] + for event, config in hooks.items(): + native = self._native_event(event) + if native is None: + continue + command = config.get("command", "") + matcher = config.get("matcher", "*") + ts_hook = native # already "tool.execute.before" etc. + match_cond = "" + if matcher and matcher != "*": + tools = [t.strip().strip('"') for t in matcher.split("|")] + checks = " || ".join(f"input.tool === '{t.lower()}'" for t in tools) + match_cond = f"if ({checks}) {{" + else: + match_cond = "{" + hook_entries.append( + f"function _{event}(input: any) {match_cond}\n" + f" runHook('{command}', '{event}', input);\n" + f" }}" + ) + plugin_returns.append( + f" \"{ts_hook}\": async (input: any, output: any) => {{\n" + f" _{event}(input);\n" + f" }}," + ) + return _TS_PLUGIN_TEMPLATE.format( + hook_entries="\n\n".join(hook_entries), + plugin_returns="\n".join(plugin_returns), + ) + + def _merge_plugin_ref(self, config_path: Path) -> None: + existing: dict = {} + if config_path.exists(): + try: + existing = json.loads(config_path.read_text(encoding="utf-8")) + if not isinstance(existing, dict): + existing = {} + except (json.JSONDecodeError, OSError): + return + plugins = existing.get("plugin", []) + if not isinstance(plugins, list): + plugins = [] + ref = f"./{self._PLUGIN_REL}" + if ref not in plugins: + plugins.append(ref) + existing["plugin"] = plugins + config_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8") + + +# -- Phase 1: JSON config-file adapters (nested matcher-groups) ------------ + +class QwenHookAdapter(JSONHookAdapter): + """Qwen Code: nested matcher-groups in .qwen/settings.json. + + Qwen uses the same event names and JSON structure as Claude Code. + """ + + config_file_rel = ".qwen/settings.json" + CANONICAL_TO_NATIVE = { + "PreToolUse": "PreToolUse", + "PostToolUse": "PostToolUse", + "Stop": "Stop", + "SessionStart": "SessionStart", + "SessionEnd": "SessionEnd", + "UserPromptSubmit": "UserPromptSubmit", + } + + def _build_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> dict: + return self._build_nested_fragment(hooks, project_root) + + +class GeminiHookAdapter(JSONHookAdapter): + """Gemini CLI: nested matcher-groups in .gemini/settings.json. + + Gemini uses different event names (BeforeTool, AfterTool, AfterAgent). + """ + + config_file_rel = ".gemini/settings.json" + CANONICAL_TO_NATIVE = { + "PreToolUse": "BeforeTool", + "PostToolUse": "AfterTool", + "Stop": "AfterAgent", + "SessionStart": "SessionStart", + "SessionEnd": "SessionEnd", + } + + def _build_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> dict: + return self._build_nested_fragment(hooks, project_root) + + +class DevinHookAdapter(JSONHookAdapter): + """Devin for Terminal: hooks in .devin/hooks.v1.json. + + Devin uses the same event names and JSON structure as Claude Code. + Also reads .claude/settings.json for compatibility, but we write to + the native format for independence. + """ + + config_file_rel = ".devin/hooks.v1.json" + CANONICAL_TO_NATIVE = { + "PreToolUse": "PreToolUse", + "PostToolUse": "PostToolUse", + "Stop": "Stop", + "SessionStart": "SessionStart", + "SessionEnd": "SessionEnd", + "UserPromptSubmit": "UserPromptSubmit", + } + + def _build_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> dict: + return self._build_nested_fragment(hooks, project_root) + + +class TabnineHookAdapter(JSONHookAdapter): + """Tabnine CLI: nested matcher-groups in .tabnine/agent/settings.json. + + Tabnine uses different event names (BeforeTool, AfterTool), like Gemini. + """ + + config_file_rel = ".tabnine/agent/settings.json" + CANONICAL_TO_NATIVE = { + "PreToolUse": "BeforeTool", + "PostToolUse": "AfterTool", + "SessionStart": "SessionStart", + "SessionEnd": "SessionEnd", + } + + def _build_fragment(self, hooks: dict[str, dict[str, Any]], project_root: Path) -> dict: + return self._build_nested_fragment(hooks, project_root) + + +# -- Adapter registry ----------------------------------------------------- + +_ADAPTERS: dict[str, type[HookAdapter]] = { + "claude": ClaudeHookAdapter, + "cursor-agent": CursorHookAdapter, + "codex": CodexHookAdapter, + "opencode": OpencodeHookAdapter, + "qwen": QwenHookAdapter, + "gemini": GeminiHookAdapter, + "devin": DevinHookAdapter, + "tabnine": TabnineHookAdapter, +} + + +def resolve_adapter(integration_key: str) -> HookAdapter | None: + """Return a hook adapter for the integration key, or None if unsupported.""" + cls = _ADAPTERS.get(integration_key) + return cls() if cls else None + + +# -- Public entry points (called from IntegrationBase) -------------------- + +def install_integration_hooks( + integration: "IntegrationBase", + project_root: Path, + manifest: "IntegrationManifest", + parsed_options: dict[str, Any] | None = None, +) -> list[Path]: + """Install runtime hooks for an integration. + + Called from ``IntegrationBase.setup()`` after commands are written. + Returns list of created/managed files. No-ops gracefully when hooks + are disabled or the agent has no adapter. + """ + adapter = resolve_adapter(integration.key) + if adapter is None: + return [] + + hooks = resolve_hooks( + integration.key, + integration.config, + project_root, + parsed_options, + ) + if not hooks: + # Hooks disabled or empty — clean up any previous hooks + remove_integration_hooks(integration, project_root, manifest) + return [] + + # Remove old hooks before installing new ones so stale events + # from a previous install/upgrade are cleaned up. + remove_integration_hooks(integration, project_root, manifest) + + return adapter.install(project_root, manifest, hooks) + + +def remove_integration_hooks( + integration: "IntegrationBase", + project_root: Path, + manifest: "IntegrationManifest", +) -> None: + """Remove runtime hooks for an integration. + + Called from ``IntegrationBase.teardown()`` before manifest uninstall. + Removes only Specify-authored entries from native config files. + """ + adapter = resolve_adapter(integration.key) + if adapter is None: + return + + adapter.remove(project_root, manifest) + + # Remove bridge script if tracked + bridge_rel = HOOK_BRIDGE_REL + if bridge_rel in manifest.files: + bridge_path = project_root / bridge_rel + if bridge_path.exists(): + bridge_path.unlink(missing_ok=True) + manifest.remove(bridge_rel) + + # Remove opencode TS plugin if tracked + if integration.key == "opencode": + plugin_rel = str(OpencodeHookAdapter._PLUGIN_REL) + if plugin_rel in manifest.files: + plugin_path = project_root / plugin_rel + if plugin_path.exists(): + plugin_path.unlink(missing_ok=True) + manifest.remove(plugin_rel) + + +def hooks_stale_exclusions(integration_key: str) -> set[str]: + """Return project-relative paths to protect from stale cleanup.""" + adapter = resolve_adapter(integration_key) + if adapter is None: + return set() + exclusions = {adapter.config_file_rel} + if integration_key == "opencode": + exclusions.add(OpencodeHookAdapter._PLUGIN_REL) + return exclusions + + +# -- Helpers --------------------------------------------------------------- + +def _has_marker(entry: Any) -> bool: + """Check if a JSON hook entry has the Specify marker.""" + if isinstance(entry, dict): + return entry.get(_SPECKIT_MARKER, False) is True + return False + + +def _to_camel_case(snake_or_pascal: str) -> str: + """Convert PascalCase or snake_case to camelCase.""" + if not snake_or_pascal: + return snake_or_pascal + first = snake_or_pascal[0].lower() + rest = snake_or_pascal[1:] + # Handle PascalCase: PreToolUse -> preToolUse + return first + rest + + +# -- Extension manifest validation callout --------------------------------- + +def validate_runtime_hooks(data: dict[str, Any]) -> None: + """Validate ``runtime_hooks`` field in extension manifest data. + + Called from ``ExtensionManifest._validate()`` via a callout. + Raises ``ValidationError`` if ``runtime_hooks`` is present but malformed. + """ + from ..extensions import ValidationError + + runtime_hooks = data.get("runtime_hooks") + if "runtime_hooks" in data and not isinstance(runtime_hooks, dict): + raise ValidationError("Invalid runtime_hooks: expected a mapping") + if runtime_hooks: + for hook_name, hook_config in runtime_hooks.items(): + if not isinstance(hook_config, dict): + raise ValidationError( + f"Invalid runtime_hook '{hook_name}': expected a mapping" + ) + if not hook_config.get("command"): + raise ValidationError( + f"Runtime hook '{hook_name}' missing required 'command' field" + ) + if hook_name not in CANONICAL_RUNTIME_EVENTS: + raise ValidationError( + f"Unknown runtime_hook '{hook_name}': " + f"must be one of {sorted(CANONICAL_RUNTIME_EVENTS)}" + ) + + +def has_runtime_hooks(data: dict[str, Any]) -> bool: + """Return True if ``runtime_hooks`` is present and non-empty.""" + return bool(data.get("runtime_hooks")) diff --git a/src/specify_cli/integrations/base.py b/src/specify_cli/integrations/base.py index ccc8587709..477c69834e 100644 --- a/src/specify_cli/integrations/base.py +++ b/src/specify_cli/integrations/base.py @@ -29,6 +29,7 @@ from .._toml_string import escape_toml_basic as _escape_toml_basic from .._toml_string import has_illegal_toml_control as _has_illegal_toml_control +from ._hooks import install_integration_hooks, remove_integration_hooks if TYPE_CHECKING: from .manifest import IntegrationManifest @@ -902,6 +903,7 @@ def teardown( Returns ``(removed, skipped)`` file lists. """ + remove_integration_hooks(self, project_root, manifest) return manifest.uninstall(project_root, force=force) # -- Convenience helpers for subclasses ------------------------------- @@ -1008,6 +1010,12 @@ def setup( created.append(dst_file) + # Install agent runtime hooks + hook_files = install_integration_hooks( + self, project_root, manifest, parsed_options + ) + created.extend(hook_files) + return created @@ -1215,6 +1223,12 @@ def setup( created.append(dst_file) + # Install agent runtime hooks + hook_files = install_integration_hooks( + self, project_root, manifest, parsed_options + ) + created.extend(hook_files) + return created @@ -1451,6 +1465,12 @@ def setup( created.append(dst_file) + # Install agent runtime hooks + hook_files = install_integration_hooks( + self, project_root, manifest, parsed_options + ) + created.extend(hook_files) + return created @@ -1686,4 +1706,10 @@ def setup( created.append(dst) + # Install agent runtime hooks + hook_files = install_integration_hooks( + self, project_root, manifest, parsed_options + ) + created.extend(hook_files) + return created diff --git a/tests/integrations/test_hooks.py b/tests/integrations/test_hooks.py new file mode 100644 index 0000000000..cb1dd1c6e2 --- /dev/null +++ b/tests/integrations/test_hooks.py @@ -0,0 +1,678 @@ +"""Tests for _hooks module: integration runtime hooks.""" + +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import yaml + +from specify_cli.integrations._hooks import ( + ClaudeHookAdapter, + CodexHookAdapter, + CursorHookAdapter, + DevinHookAdapter, + GeminiHookAdapter, + HookAdapter, + JSONHookAdapter, + OpencodeHookAdapter, + QwenHookAdapter, + TabnineHookAdapter, + _has_marker, + _to_camel_case, + CANONICAL_RUNTIME_EVENTS, + collect_extension_runtime_hooks, + hooks_stale_exclusions, + install_integration_hooks, + remove_integration_hooks, + resolve_adapter, + resolve_hooks, + validate_runtime_hooks, +) +from specify_cli.integrations.manifest import IntegrationManifest + + +# -- resolve_hooks -------------------------------------------------------- + +class TestResolveHooks: + """Test the 4-layer hook resolution chain.""" + + def test_layer1_disabled_returns_empty(self, tmp_path): + """--hooks false returns empty dict.""" + result = resolve_hooks( + "claude", + {"hooks": {"PostToolUse": {"command": "speckit.tdd.validate"}}}, + tmp_path, + {"hooks": "false"}, + ) + assert result == {} + + def test_layer4_built_in_defaults(self, tmp_path): + """Built-in config hooks are returned when no override.""" + config = {"hooks": {"PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write"}}} + result = resolve_hooks("claude", config, tmp_path, None) + assert "PostToolUse" in result + assert result["PostToolUse"]["command"] == "speckit.tdd.validate" + + def test_layer3_extension_hooks_appended(self, tmp_path): + """Extension-declared runtime_hooks are appended to built-in.""" + ext_dir = tmp_path / ".specify" / "extensions" / "myext" + ext_dir.mkdir(parents=True) + (ext_dir / "extension.yml").write_text(yaml.dump({ + "schema_version": "1.0", + "extension": {"id": "myext", "name": "My Ext", "version": "1.0.0", + "description": "test"}, + "requires": {"speckit_version": ">=0.0.80"}, + "provides": {"commands": []}, + "runtime_hooks": { + "Stop": {"command": "speckit.myext.check", "matcher": "*"}, + }, + })) + config = {"hooks": {"PostToolUse": {"command": "speckit.tdd.validate"}}} + result = resolve_hooks("claude", config, tmp_path, None) + assert "PostToolUse" in result + assert "Stop" in result + assert result["Stop"]["command"] == "speckit.myext.check" + + def test_layer2_yaml_override_replaces(self, tmp_path): + """YAML override replaces built-in and extension hooks entirely.""" + override_file = tmp_path / ".specify" / "integration-hooks.yml" + override_file.parent.mkdir(parents=True) + override_file.write_text(yaml.dump({ + "version": 1, + "integrations": { + "claude": { + "hooks": { + "PreToolUse": {"command": "speckit.protected_paths", "matcher": "Edit|Write"}, + }, + }, + }, + })) + config = {"hooks": {"PostToolUse": {"command": "speckit.tdd.validate"}}} + result = resolve_hooks("claude", config, tmp_path, None) + assert "PreToolUse" in result + assert "PostToolUse" not in result # replaced, not merged + + def test_layer2_empty_hooks_disables(self, tmp_path): + """Empty hooks: {} in YAML override disables all hooks.""" + override_file = tmp_path / ".specify" / "integration-hooks.yml" + override_file.parent.mkdir(parents=True) + override_file.write_text(yaml.dump({ + "version": 1, + "integrations": {"claude": {"hooks": {}}}, + })) + config = {"hooks": {"PostToolUse": {"command": "speckit.tdd.validate"}}} + result = resolve_hooks("claude", config, tmp_path, None) + assert result == {} + + def test_no_config_no_hooks(self, tmp_path): + """No config, no extensions, no override → empty.""" + result = resolve_hooks("gemini", None, tmp_path, None) + assert result == {} + + +# -- collect_extension_runtime_hooks -------------------------------------- + +class TestCollectExtensionRuntimeHooks: + """Test scanning installed extensions for runtime_hooks.""" + + def test_no_extensions_dir(self, tmp_path): + assert collect_extension_runtime_hooks(tmp_path) == {} + + def test_no_runtime_hooks_in_extension(self, tmp_path): + ext_dir = tmp_path / ".specify" / "extensions" / "myext" + ext_dir.mkdir(parents=True) + (ext_dir / "extension.yml").write_text(yaml.dump({ + "schema_version": "1.0", + "extension": {"id": "myext", "name": "My Ext", "version": "1.0.0", + "description": "test"}, + "requires": {"speckit_version": ">=0.0.80"}, + "provides": {"commands": [{"name": "speckit.myext.cmd", "file": "cmd.md"}]}, + "hooks": {"after_plan": {"command": "speckit.myext.cmd"}}, + })) + result = collect_extension_runtime_hooks(tmp_path) + assert result == {} + + def test_runtime_hooks_collected(self, tmp_path): + ext_dir = tmp_path / ".specify" / "extensions" / "tdd" + ext_dir.mkdir(parents=True) + (ext_dir / "extension.yml").write_text(yaml.dump({ + "schema_version": "1.0", + "extension": {"id": "tdd", "name": "TDD", "version": "1.0.0", + "description": "test"}, + "requires": {"speckit_version": ">=0.0.80"}, + "provides": {"commands": []}, + "runtime_hooks": { + "PostToolUse": {"command": "adlc.tdd.validate", "matcher": "Edit|Write"}, + "Stop": {"command": "adlc.tdd.validate", "matcher": "*"}, + }, + })) + result = collect_extension_runtime_hooks(tmp_path) + assert "PostToolUse" in result + assert "Stop" in result + assert result["PostToolUse"]["command"] == "adlc.tdd.validate" + + def test_invalid_yaml_skipped(self, tmp_path): + ext_dir = tmp_path / ".specify" / "extensions" / "broken" + ext_dir.mkdir(parents=True) + (ext_dir / "extension.yml").write_text("{{invalid yaml") + result = collect_extension_runtime_hooks(tmp_path) + assert result == {} + + +# -- Adapter resolution --------------------------------------------------- + +class TestResolveAdapter: + """Test adapter registry.""" + + def test_claude_returns_adapter(self): + adapter = resolve_adapter("claude") + assert isinstance(adapter, ClaudeHookAdapter) + + def test_cursor_returns_adapter(self): + adapter = resolve_adapter("cursor-agent") + assert isinstance(adapter, CursorHookAdapter) + + def test_codex_returns_adapter(self): + adapter = resolve_adapter("codex") + assert isinstance(adapter, CodexHookAdapter) + + def test_opencode_returns_adapter(self): + adapter = resolve_adapter("opencode") + assert isinstance(adapter, OpencodeHookAdapter) + + def test_gemini_returns_adapter(self): + adapter = resolve_adapter("gemini") + assert isinstance(adapter, GeminiHookAdapter) + + def test_qwen_returns_adapter(self): + adapter = resolve_adapter("qwen") + assert isinstance(adapter, QwenHookAdapter) + + def test_devin_returns_adapter(self): + adapter = resolve_adapter("devin") + assert isinstance(adapter, DevinHookAdapter) + + def test_tabnine_returns_adapter(self): + adapter = resolve_adapter("tabnine") + assert isinstance(adapter, TabnineHookAdapter) + + def test_copilot_returns_none(self): + assert resolve_adapter("copilot") is None + + def test_unknown_returns_none(self): + assert resolve_adapter("nonexistent") is None + + +# -- Canonical event mapping ----------------------------------------------- + +class TestCanonicalEventMapping: + """Test canonical→native event name translation per adapter.""" + + def test_claude_passthrough(self): + adapter = ClaudeHookAdapter() + assert adapter._native_event("PreToolUse") == "PreToolUse" + assert adapter._native_event("PostToolUse") == "PostToolUse" + assert adapter._native_event("Stop") == "Stop" + + def test_cursor_camelcase(self): + adapter = CursorHookAdapter() + assert adapter._native_event("PreToolUse") == "preToolUse" + assert adapter._native_event("PostToolUse") == "postToolUse" + assert adapter._native_event("UserPromptSubmit") == "beforeSubmitPrompt" + + def test_opencode_limited(self): + adapter = OpencodeHookAdapter() + assert adapter._native_event("PreToolUse") == "tool.execute.before" + assert adapter._native_event("PostToolUse") == "tool.execute.after" + assert adapter._native_event("Stop") is None # unsupported + + def test_gemini_mapping(self): + adapter = GeminiHookAdapter() + assert adapter._native_event("PreToolUse") == "BeforeTool" + assert adapter._native_event("PostToolUse") == "AfterTool" + assert adapter._native_event("Stop") == "AfterAgent" + + def test_qwen_identity(self): + adapter = QwenHookAdapter() + assert adapter._native_event("PreToolUse") == "PreToolUse" + + def test_devin_identity(self): + adapter = DevinHookAdapter() + assert adapter._native_event("PreToolUse") == "PreToolUse" + + def test_tabnine_mapping(self): + adapter = TabnineHookAdapter() + assert adapter._native_event("PreToolUse") == "BeforeTool" + assert adapter._native_event("PostToolUse") == "AfterTool" + assert adapter._native_event("Stop") is None # not supported + + +class TestUnknownEventRejection: + """Test that validate_runtime_hooks rejects unknown event names.""" + + def test_unknown_event_rejected(self): + from specify_cli.extensions import ValidationError + data = { + "runtime_hooks": { + "FooBar": {"command": "speckit.test"}, + }, + } + with pytest.raises(ValidationError, match="Unknown runtime_hook 'FooBar'"): + validate_runtime_hooks(data) + + def test_known_event_accepted(self): + data = { + "runtime_hooks": { + "PostToolUse": {"command": "speckit.test"}, + }, + } + validate_runtime_hooks(data) # should not raise + + def test_all_canonical_events_accepted(self): + for event in CANONICAL_RUNTIME_EVENTS: + data = {"runtime_hooks": {event: {"command": "speckit.test"}}} + validate_runtime_hooks(data) # should not raise + + +class TestUnsupportedEventWarning: + """Test that adapters warn and skip unsupported events.""" + + def test_opencode_skips_stop(self, tmp_path, capsys): + from specify_cli.integrations.manifest import IntegrationManifest + adapter = OpencodeHookAdapter() + manifest = IntegrationManifest("opencode", tmp_path) + hooks = { + "PostToolUse": {"command": "speckit.test", "matcher": "Edit|Write"}, + "Stop": {"command": "speckit.test", "matcher": "*"}, + } + adapter.install(tmp_path, manifest, hooks) + captured = capsys.readouterr() + assert "Stop" in captured.err + assert "skipping" in captured.err + # Plugin should only have PostToolUse, not Stop + plugin = tmp_path / ".opencode" / "plugin" / "speckit-hooks.ts" + content = plugin.read_text() + assert "tool.execute.after" in content + assert "tool.execute.before" not in content + + def test_tabnine_skips_stop(self, tmp_path, capsys): + adapter = TabnineHookAdapter() + from specify_cli.integrations.manifest import IntegrationManifest + manifest = IntegrationManifest("tabnine", tmp_path) + hooks = { + "PostToolUse": {"command": "speckit.test", "matcher": "Edit|Write"}, + "Stop": {"command": "speckit.test", "matcher": "*"}, + } + adapter.install(tmp_path, manifest, hooks) + captured = capsys.readouterr() + assert "Stop" in captured.err + settings = tmp_path / ".tabnine" / "agent" / "settings.json" + data = json.loads(settings.read_text()) + assert "AfterTool" in data.get("hooks", {}) + assert "Stop" not in data.get("hooks", {}) + + +# -- New adapter tests ----------------------------------------------------- + +class TestGeminiHookAdapter: + + def test_generate_fragment_uses_native_names(self): + adapter = GeminiHookAdapter() + hooks = { + "PreToolUse": {"command": "speckit.test", "matcher": "Edit|Write", "timeout": 10}, + } + fmt, fragment = adapter.generate_fragment(hooks, Path("/tmp")) + assert fmt == "json" + assert "BeforeTool" in fragment["hooks"] + assert "PreToolUse" not in fragment["hooks"] + + def test_install_creates_settings(self, tmp_path): + from specify_cli.integrations.manifest import IntegrationManifest + adapter = GeminiHookAdapter() + manifest = IntegrationManifest("gemini", tmp_path) + hooks = {"PostToolUse": {"command": "speckit.test", "matcher": "Edit|Write", "timeout": 60}} + created = adapter.install(tmp_path, manifest, hooks) + settings = tmp_path / ".gemini" / "settings.json" + assert settings.exists() + data = json.loads(settings.read_text()) + assert "AfterTool" in data["hooks"] + + +class TestQwenHookAdapter: + + def test_generate_fragment_identity(self): + adapter = QwenHookAdapter() + hooks = { + "PreToolUse": {"command": "speckit.test", "matcher": "Edit|Write", "timeout": 10}, + } + fmt, fragment = adapter.generate_fragment(hooks, Path("/tmp")) + assert "PreToolUse" in fragment["hooks"] + + +class TestDevinHookAdapter: + + def test_install_creates_hooks_file(self, tmp_path): + from specify_cli.integrations.manifest import IntegrationManifest + adapter = DevinHookAdapter() + manifest = IntegrationManifest("devin", tmp_path) + hooks = {"PostToolUse": {"command": "speckit.test", "matcher": "Edit|Write", "timeout": 60}} + created = adapter.install(tmp_path, manifest, hooks) + hooks_file = tmp_path / ".devin" / "hooks.v1.json" + assert hooks_file.exists() + data = json.loads(hooks_file.read_text()) + assert "PostToolUse" in data["hooks"] + + +class TestTabnineHookAdapter: + + def test_generate_fragment_native_names(self): + adapter = TabnineHookAdapter() + hooks = { + "PreToolUse": {"command": "speckit.test", "matcher": "Edit|Write", "timeout": 10}, + } + fmt, fragment = adapter.generate_fragment(hooks, Path("/tmp")) + assert "BeforeTool" in fragment["hooks"] + + +# -- ClaudeHookAdapter ---------------------------------------------------- + +class TestClaudeHookAdapter: + """Test Claude Code native config generation and merge.""" + + def test_generate_fragment_structure(self): + adapter = ClaudeHookAdapter() + hooks = { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + } + fmt, fragment = adapter.generate_fragment(hooks, Path("/tmp")) + assert fmt == "json" + assert "hooks" in fragment + assert "PostToolUse" in fragment["hooks"] + entry_group = fragment["hooks"]["PostToolUse"][0] + assert entry_group["matcher"] == "Edit|Write" + handler = entry_group["hooks"][0] + assert handler["type"] == "command" + assert handler["command"] == "python3" + assert "speckit.tdd.validate" in handler["args"] + assert "PostToolUse" in handler["args"] + assert handler["timeout"] == 60 + + def test_merge_into_empty_file(self, tmp_path): + adapter = ClaudeHookAdapter() + config_path = tmp_path / ".claude" / "settings.json" + config_path.parent.mkdir(parents=True) + hooks = { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + } + fmt, fragment = adapter.generate_fragment(hooks, tmp_path) + adapter.merge_fragment(config_path, fragment, format=fmt) + data = json.loads(config_path.read_text()) + assert "hooks" in data + assert "PostToolUse" in data["hooks"] + assert len(data["hooks"]["PostToolUse"]) == 1 + + def test_merge_preserves_user_keys(self, tmp_path): + adapter = ClaudeHookAdapter() + config_path = tmp_path / ".claude" / "settings.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({ + "model": "claude-sonnet-4-20250514", + "permissions": {"allow": ["Bash(git *)"]}, + "hooks": { + "PostToolUse": [{ + "matcher": "Bash", + "hooks": [{"type": "command", "command": "/usr/bin/my-hook.sh"}], + }], + }, + })) + hooks = { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + } + fmt, fragment = adapter.generate_fragment(hooks, tmp_path) + adapter.merge_fragment(config_path, fragment, format=fmt) + data = json.loads(config_path.read_text()) + assert data["model"] == "claude-sonnet-4-20250514" + assert data["permissions"]["allow"] == ["Bash(git *)"] + assert len(data["hooks"]["PostToolUse"]) == 2 # user hook + our hook + + def test_remove_entries_preserves_user_hooks(self, tmp_path): + adapter = ClaudeHookAdapter() + config_path = tmp_path / ".claude" / "settings.json" + config_path.parent.mkdir(parents=True) + config_path.write_text(json.dumps({ + "hooks": { + "PostToolUse": [ + {"matcher": "Bash", "hooks": [{"type": "command", "command": "/usr/bin/my-hook.sh"}]}, + {"matcher": "Edit|Write", "hooks": [{"type": "command", "command": "python3", + "args": [".specify/hooks/bridge.py", "speckit.tdd.validate", "PostToolUse"], + "__speckit_hook__": True}]}, + ], + }, + })) + adapter.remove_entries(config_path) + data = json.loads(config_path.read_text()) + assert len(data["hooks"]["PostToolUse"]) == 1 + assert data["hooks"]["PostToolUse"][0]["matcher"] == "Bash" + + +# -- CursorHookAdapter ---------------------------------------------------- + +class TestCursorHookAdapter: + """Test Cursor native config generation.""" + + def test_generate_fragment_camelcase(self): + adapter = CursorHookAdapter() + hooks = { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + } + fmt, fragment = adapter.generate_fragment(hooks, Path("/tmp")) + assert "postToolUse" in fragment["hooks"] + entry = fragment["hooks"]["postToolUse"][0] + assert "python3" in entry["command"] + assert entry["matcher"] == "Edit|Write" + + +# -- CodexHookAdapter ----------------------------------------------------- + +class TestCodexHookAdapter: + """Test Codex TOML config generation.""" + + def test_generate_fragment_toml(self): + adapter = CodexHookAdapter() + hooks = { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + } + fmt, fragment = adapter.generate_fragment(hooks, Path("/tmp")) + assert fmt == "toml" + assert "[[hooks.PostToolUse]]" in fragment + assert 'matcher = "Edit|Write"' in fragment + assert "speckit.tdd.validate" in fragment + assert "speckit_marker = true" in fragment + + +# -- OpencodeHookAdapter -------------------------------------------------- + +class TestOpencodeHookAdapter: + """Test opencode TS plugin generation.""" + + def test_install_generates_plugin_and_merges_config(self, tmp_path): + adapter = OpencodeHookAdapter() + manifest = MagicMock(spec=IntegrationManifest) + manifest.files = {} + manifest.record_file = MagicMock() + manifest.record_existing = MagicMock() + hooks = { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + } + created = adapter.install(tmp_path, manifest, hooks) + plugin_path = tmp_path / ".opencode/plugin/speckit-hooks.ts" + assert plugin_path.exists() + content = plugin_path.read_text() + assert "tool.execute.after" in content + assert "speckit.tdd.validate" in content + config_path = tmp_path / "opencode.json" + assert config_path.exists() + config = json.loads(config_path.read_text()) + assert "./.opencode/plugin/speckit-hooks.ts" in config["plugin"] + + def test_remove_removes_plugin_ref(self, tmp_path): + adapter = OpencodeHookAdapter() + config_path = tmp_path / "opencode.json" + config_path.write_text(json.dumps({ + "plugin": ["./.opencode/plugin/speckit-hooks.ts", "./other.ts"], + })) + adapter.remove_entries(config_path) + config = json.loads(config_path.read_text()) + assert "./.opencode/plugin/speckit-hooks.ts" not in config["plugin"] + assert "./other.ts" in config["plugin"] + + +# -- Helpers -------------------------------------------------------------- + +class TestHelpers: + def test_has_marker_true(self): + assert _has_marker({"__speckit_hook__": True, "command": "foo"}) + + def test_has_marker_false(self): + assert not _has_marker({"command": "foo"}) + + def test_to_camel_case(self): + assert _to_camel_case("PostToolUse") == "postToolUse" + assert _to_camel_case("PreToolUse") == "preToolUse" + assert _to_camel_case("Stop") == "stop" + assert _to_camel_case("") == "" + + +# -- hooks_stale_exclusions ----------------------------------------------- + +class TestStaleExclusions: + def test_claude_returns_settings_path(self): + exclusions = hooks_stale_exclusions("claude") + assert ".claude/settings.json" in exclusions + + def test_opencode_returns_config_and_plugin(self): + exclusions = hooks_stale_exclusions("opencode") + assert "opencode.json" in exclusions + assert ".opencode/plugin/speckit-hooks.ts" in exclusions + + def test_gemini_returns_settings_path(self): + exclusions = hooks_stale_exclusions("gemini") + assert ".gemini/settings.json" in exclusions + + def test_copilot_returns_empty(self): + assert hooks_stale_exclusions("copilot") == set() + + +# -- Integration: install + remove round-trip ----------------------------- + +class TestInstallRemoveRoundTrip: + """End-to-end install → remove for Claude adapter.""" + + def test_claude_install_then_remove(self, tmp_path): + from specify_cli.integrations._hooks import HOOK_BRIDGE_REL + + manifest = IntegrationManifest("claude", tmp_path) + adapter = ClaudeHookAdapter() + hooks = { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + } + created = adapter.install(tmp_path, manifest, hooks) + assert len(created) == 2 # bridge + settings.json + bridge = tmp_path / HOOK_BRIDGE_REL + assert bridge.exists() + settings = tmp_path / ".claude/settings.json" + assert settings.exists() + data = json.loads(settings.read_text()) + assert "PostToolUse" in data["hooks"] + + # Remove + adapter.remove(tmp_path, manifest) + data = json.loads(settings.read_text()) + # Our entries removed, but file still exists + assert "hooks" not in data or "PostToolUse" not in data.get("hooks", {}) + + def test_group_b_no_crash(self, tmp_path): + """Installing hooks for a Group B agent (no adapter) is a no-op.""" + mock_integration = MagicMock() + mock_integration.key = "copilot" + mock_integration.config = {"hooks": {"PostToolUse": {"command": "speckit.tdd.validate"}}} + manifest = IntegrationManifest("gemini", tmp_path) + result = install_integration_hooks(mock_integration, tmp_path, manifest, None) + assert result == [] + + def test_upgrade_removes_stale_events(self, tmp_path): + """Upgrading from hook set A to hook set B removes stale events.""" + mock_integration = MagicMock() + mock_integration.key = "claude" + mock_integration.config = None + (tmp_path / ".specify").mkdir(parents=True) + + # v1: install PostToolUse only + mock_integration.config = {"hooks": { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + }} + manifest_v1 = IntegrationManifest("claude", tmp_path) + install_integration_hooks(mock_integration, tmp_path, manifest_v1, None) + settings = tmp_path / ".claude" / "settings.json" + data = json.loads(settings.read_text()) + assert "PostToolUse" in data.get("hooks", {}) + + # v2: upgrade to PreToolUse + Stop (no PostToolUse) + mock_integration.config = {"hooks": { + "PreToolUse": {"command": "speckit.protected_paths", "matcher": "Edit|Write", "timeout": 10}, + "Stop": {"command": "speckit.tdd.validate", "matcher": "*", "timeout": 30}, + }} + manifest_v2 = IntegrationManifest("claude", tmp_path) + install_integration_hooks(mock_integration, tmp_path, manifest_v2, None) + data2 = json.loads(settings.read_text()) + hooks_v2 = data2.get("hooks", {}) + assert "PostToolUse" not in hooks_v2, "Stale PostToolUse should be removed on upgrade" + assert "PreToolUse" in hooks_v2 + assert "Stop" in hooks_v2 + + def test_upgrade_preserves_user_hooks(self, tmp_path): + """User-defined hooks survive upgrade.""" + mock_integration = MagicMock() + mock_integration.key = "claude" + mock_integration.config = None + (tmp_path / ".specify").mkdir(parents=True) + + # Pre-existing user hook + settings_dir = tmp_path / ".claude" + settings_dir.mkdir(parents=True) + (settings_dir / "settings.json").write_text(json.dumps({ + "hooks": { + "PostToolUse": [{ + "matcher": "Bash", + "hooks": [{"type": "command", "command": "/usr/bin/my-linter.sh"}], + }], + }, + })) + + # Install our hook + mock_integration.config = {"hooks": { + "PostToolUse": {"command": "speckit.tdd.validate", "matcher": "Edit|Write", "timeout": 60}, + }} + manifest = IntegrationManifest("claude", tmp_path) + install_integration_hooks(mock_integration, tmp_path, manifest, None) + data = json.loads((settings_dir / "settings.json").read_text()) + post_hooks = data["hooks"]["PostToolUse"] + assert len(post_hooks) == 2 # user hook + our hook + + # Upgrade: change our hook to PreToolUse only + mock_integration.config = {"hooks": { + "PreToolUse": {"command": "speckit.protected_paths", "matcher": "Edit|Write", "timeout": 10}, + }} + manifest_v2 = IntegrationManifest("claude", tmp_path) + install_integration_hooks(mock_integration, tmp_path, manifest_v2, None) + data2 = json.loads((settings_dir / "settings.json").read_text()) + hooks2 = data2.get("hooks", {}) + # User's PostToolUse hook should survive + assert "PostToolUse" in hooks2 + user_entry = hooks2["PostToolUse"][0] + assert user_entry["matcher"] == "Bash" + # Our new PreToolUse should be present + assert "PreToolUse" in hooks2