feat: first-class agent-native runtime hooks for integrations

This commit is contained in:
Lior Kanfi
2026-07-24 01:13:28 +03:00
parent 4d3a4281bc
commit c5f1e96e81
3 changed files with 1679 additions and 0 deletions

View File

@@ -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"))

View File

@@ -29,6 +29,7 @@ import yaml
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 @@ class IntegrationBase(ABC):
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 @@ class MarkdownIntegration(IntegrationBase):
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 @@ class TomlIntegration(IntegrationBase):
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 @@ class YamlIntegration(IntegrationBase):
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 @@ class SkillsIntegration(IntegrationBase):
created.append(dst)
# Install agent runtime hooks
hook_files = install_integration_hooks(
self, project_root, manifest, parsed_options
)
created.extend(hook_files)
return created

View File

@@ -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