github/github-spec-kit
1
0
Fork 0
mirror of https://github.com/github/spec-kit.git synced 2026-08-03 06:26:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: rework integration events per maintainer review

Browse Source 
- Rename hooks terminology to 'events' (events:, --events flag, events.py).
- Use snake_case names for canonical events consistent with spec-kit vocabulary.
- Fold event config adapters into integration classes via class attributes (CANONICAL_TO_NATIVE, events_config_file, events_format).
- Lift event command-script resolution to core 'specify event run' command.
- Split events sourcing from integration config writing.
- Support first-class Copilot CLI events JSON generation under '.github/hooks/speckit.json'.
- Rewrite and expand full test suite under 'tests/integrations/test_events.py'.

Assisted-by: opencode (model: litellm/gemini-3.5-flash, autonomous)
This commit is contained in:
Lior Kanfi
2026-07-25 17:17:17 +03:00
parent c5f1e96e81
commit da6c20d9f0
20 changed files with 1399 additions and 1686 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/specify_cli/__init__.py
View File

@@ -508,6 +508,11 @@ _register_extension_cmds(app)
from .integrations._commands import register as _register_integration_cmds # noqa: E402
_register_integration_cmds(app)
# ===== Event Commands =====
from .commands.event import register as _register_event_cmds
_register_event_cmds(app)
# Re-export selected helpers to preserve the public import surface.
from .integrations._helpers import ( # noqa: E402
_clear_init_options_for_integration as _clear_init_options_for_integration,

36
src/specify_cli/commands/event.py Normal file
View File

@@ -0,0 +1,36 @@
"""specify event * command handlers."""
from __future__ import annotations
from pathlib import Path
import sys
import typer
from .._console import console
event_app = typer.Typer(
name="event",
help="Manage and execute event-driven commands",
add_completion=False,
)
@event_app.command("run")
def event_run(
command_name: str = typer.Argument(..., help="Name of the command to execute"),
event_name: str = typer.Argument(..., help="Canonical event name (e.g., session_start)"),
):
"""Resolve and run an event-driven command script with stdin payload."""
from ..events import resolve_and_run_event_command
# Read payload from stdin if available
payload = sys.stdin.read() if not sys.stdin.isatty() else "{}"
# Run the event command
project_root = Path.cwd() # The agent runs events from project root
exit_code = resolve_and_run_event_command(command_name, event_name, payload, project_root)
raise typer.Exit(code=exit_code)
def register(app: typer.Typer) -> None:
app.add_typer(event_app, name="event")

8
src/specify_cli/commands/init.py
View File

@@ -442,12 +442,20 @@ def register(app: typer.Typer) -> None:
if extra:
integration_parsed_options.update(extra)
from ..events import resolve_events
events_map = resolve_events(
resolved_integration.key,
resolved_integration.config,
project_path,
integration_parsed_options or None,
)
resolved_integration.setup(
project_path,
manifest,
parsed_options=integration_parsed_options or None,
script_type=selected_script,
raw_options=integration_options,
events=events_map,
)
manifest.save()

769
src/specify_cli/events.py Normal file
View File

@@ -0,0 +1,769 @@
"""Agent runtime events for integrations.
Provides:
- ``resolve_events`` — layered event resolution (CLI flag → YAML override → extension-declared → built-in).
- ``collect_extension_events`` — scan installed extension.yml files for ``events:``.
- ``install_integration_events`` / ``remove_integration_events`` — entry points called from ``IntegrationBase.setup()`` / ``teardown()``.
"""
from __future__ import annotations
import json
import logging
import re
import sys
import subprocess
import platform
from pathlib import Path
from typing import TYPE_CHECKING, Any
import yaml
if TYPE_CHECKING:
from .integrations.base import IntegrationBase
from .integrations.manifest import IntegrationManifest
logger = logging.getLogger(__name__)
# -- Constants -------------------------------------------------------------
EVENTS_DISPATCHER_DIR = Path(".specify")
EVENTS_DISPATCHER_FILENAME = "events.py"
EVENTS_DISPATCHER_REL = str(EVENTS_DISPATCHER_DIR / EVENTS_DISPATCHER_FILENAME)
YAML_OVERRIDE_FILENAME = Path(".specify") / "integration-events.yml"
_SPECKIT_MARKER = "__speckit_event__"
# Canonical event names (snake_case)
CANONICAL_EVENTS = frozenset({
"session_start",
"pre_tool_use",
"post_tool_use",
"session_end",
"user_prompt_submit",
"stop",
})
# -- Events Dispatcher template ---------------------------------------------
_EVENTS_DISPATCHER_TEMPLATE = '''#!/usr/bin/env python3
"""Specify CLI Event Dispatcher — dispatches agent runtime events.
Generated by: specify integration install/upgrade
Do not edit manually.
"""
import sys, subprocess, os
from pathlib import Path
def _find_specify():
project_root = Path(__file__).parent.parent.resolve()
# 1. Check local virtualenv
for candidate in ("venv", ".venv"):
py_bin = project_root / candidate / "bin" / "specify"
if py_bin.exists():
return [str(py_bin)]
# Windows virtualenv
win_bin = project_root / candidate / "Scripts" / "specify.exe"
if win_bin.exists():
return [str(win_bin)]
# 2. Check python -m specify_cli
for candidate in ("venv", ".venv"):
py_exe = project_root / candidate / "bin" / "python"
if py_exe.exists():
return [str(py_exe), "-m", "specify_cli"]
win_py = project_root / candidate / "Scripts" / "python.exe"
if win_py.exists():
return [str(win_py), "-m", "specify_cli"]
# 3. Fallback to system path specify
return ["specify"]
def main():
if len(sys.argv) < 3:
sys.exit(0)
command_name = sys.argv[1]
event_name = sys.argv[2]
payload = sys.stdin.read() if not sys.stdin.isatty() else "{}"
specify_args = _find_specify() + ["event", "run", command_name, event_name]
try:
result = subprocess.run(
specify_args,
input=payload,
capture_output=True,
text=True,
timeout=120,
)
if result.stdout:
sys.stdout.write(result.stdout)
if result.returncode != 0:
if result.stderr:
sys.stderr.write(result.stderr)
sys.exit(result.returncode)
sys.exit(0)
except subprocess.TimeoutExpired:
print(f"Event {command_name} timed out", file=sys.stderr)
sys.exit(2)
except Exception as e:
print(f"Event {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 DISPATCHER = path.join(process.cwd(), '.specify', 'events.py');
function runEvent(command: string, event: string, input: any): void {{
try {{
execSync(`python3 ${{DISPATCHER}} ${{command}} ${{event}}`, {{
input: JSON.stringify(input),
stdio: ['pipe', 'inherit', 'inherit'],
timeout: 60000,
}});
}} catch (e) {{
process.exit(2);
}}
}}
{event_entries}
export default (async ({{ client, project, directory, $ }}) => {{
return {{
{plugin_returns}
}};
}});
'''
# -- Command runner logic (core) --------------------------------------------
def _find_command_template(command_name: str, project_root: Path) -> tuple[Path | None, str | None]:
# 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
# Fallback to package bundled templates
try:
import inspect
pkg_dir = Path(inspect.getfile(validate_events)).resolve().parent
core_pack = pkg_dir / "core_pack" / "templates" / "commands"
if core_pack.is_dir():
stem = command_name.replace("speckit.", "").replace("spec.", "")
candidate = core_pack / f"{stem}.md"
if candidate.exists():
return candidate, None
except Exception:
pass
return None, None
def _extract_script_path(template_path: Path, project_root: Path, ext_id: str | None) -> str | None:
content = template_path.read_text(encoding="utf-8")
m = re.match(r'^---\n(.*?)\n---', content, re.DOTALL)
if not m:
return None
fm = m.group(1)
try:
fm_data = yaml.safe_load(fm) or {}
except Exception:
return None
if not isinstance(fm_data, dict):
return None
scripts = fm_data.get("scripts", {})
if not isinstance(scripts, dict):
return None
default = "ps" if platform.system().lower().startswith("win") else "sh"
script_rel = scripts.get(default) or scripts.get("sh") or scripts.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 resolve_and_run_event_command(command_name: str, event_name: str, payload: str, project_root: Path) -> int:
"""Core entry point to resolve and execute an event-driven command."""
template_path, ext_id = _find_command_template(command_name, project_root)
if not template_path:
logger.warning("Event command '%s' not found", command_name)
return 0
script_path = _extract_script_path(template_path, project_root, ext_id)
if not script_path:
logger.warning("No script found for event command '%s'", command_name)
return 0
try:
result = subprocess.run(
[script_path],
input=payload,
capture_output=True,
text=True,
timeout=120,
)
if result.stdout:
sys.stdout.write(result.stdout)
if result.returncode != 0:
if result.stderr:
sys.stderr.write(result.stderr)
return result.returncode
return 0
except subprocess.TimeoutExpired:
sys.stderr.write(f"Event command {command_name} timed out\n")
return 2
except Exception as e:
sys.stderr.write(f"Event command {command_name} error: {e}\n")
return 2
# -- Sourcing events map (CLI/Orchestration domain) -------------------------
def resolve_events(
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 event set for an integration."""
# Layer 1: CLI flag gate
if parsed_options:
events_flag = str(parsed_options.get("events", "true")).lower()
if events_flag in ("false", "0", "no", "off"):
return {}
# Layer 4: built-in defaults from integration config
events: dict[str, dict[str, Any]] = {}
if integration_config and isinstance(integration_config.get("events"), dict):
events = dict(integration_config["events"])
# Layer 3: extension-declared events
ext_events = collect_extension_events(project_root)
events.update(ext_events)
# 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):
events = key_data.get("events", {}) or {}
else:
events = {}
return events
def collect_extension_events(project_root: Path) -> dict[str, dict[str, Any]]:
"""Scan all installed extensions for ``events:`` declarations."""
events: dict[str, dict[str, Any]] = {}
exts_dir = project_root / ".specify" / "extensions"
if not exts_dir.is_dir():
return events
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("events", {}) or {}
if not isinstance(runtime, dict):
continue
for event, config in runtime.items():
if isinstance(config, dict):
events[event] = config
return events
# -- Writing/Merging Config (Integration domain) ---------------------------
def install_integration_events(
integration: IntegrationBase,
project_root: Path,
manifest: IntegrationManifest,
events: dict[str, dict[str, Any]],
) -> list[Path]:
"""Generate dispatcher, merge native config, return created files."""
canonical_to_native = getattr(integration, "CANONICAL_TO_NATIVE", {})
if not canonical_to_native:
return []
# Filter to only supported events
filtered: dict[str, dict[str, Any]] = {}
for ev, cfg in events.items():
if ev in canonical_to_native:
filtered[ev] = cfg
else:
print(
f"\u26a0\ufe0f {integration.key} does not support '{ev}' events; skipping",
file=sys.stderr,
)
if not filtered:
return []
created: list[Path] = []
# 1. Generate events.py dispatcher script
dispatcher_dir = project_root / EVENTS_DISPATCHER_DIR
dispatcher_dir.mkdir(parents=True, exist_ok=True)
dispatcher_path = dispatcher_dir / EVENTS_DISPATCHER_FILENAME
dispatcher_path.write_text(_EVENTS_DISPATCHER_TEMPLATE, encoding="utf-8")
dispatcher_path.chmod(0o755)
manifest.record_file(
str(dispatcher_path.relative_to(project_root)),
dispatcher_path.read_bytes(),
)
created.append(dispatcher_path)
# 2. Format-specific merge/write
fmt = getattr(integration, "events_format", "json-nested")
config_file = getattr(integration, "events_config_file", None)
if not config_file:
return created
config_path = project_root / config_file
if fmt == "ts-plugin":
# Opencode TS plugin custom merge
plugin_rel = ".opencode/plugin/speckit-events.ts"
plugin_path = project_root / plugin_rel
plugin_path.parent.mkdir(parents=True, exist_ok=True)
plugin_path.write_text(_build_opencode_plugin(filtered, canonical_to_native), encoding="utf-8")
manifest.record_file(
plugin_rel,
plugin_path.read_bytes(),
)
created.append(plugin_path)
# Merge plugin path into opencode.json
_merge_opencode_plugin_ref(config_path, f"./{plugin_rel}")
rel = str(config_path.relative_to(project_root))
if rel not in manifest.files:
manifest.record_existing(rel)
created.append(config_path)
elif fmt == "copilot-json":
# Copilot hooks JSON write (creates dedicated .github/hooks/speckit.json)
copilot_hooks = {}
for ev, cfg in filtered.items():
native = canonical_to_native[ev]
command = cfg.get("command", "")
copilot_hooks[native] = [
{
"type": "command",
"bash": f"python3 {EVENTS_DISPATCHER_REL} {command} {ev}",
"powershell": f"python3 {EVENTS_DISPATCHER_REL} {command} {ev}",
"timeoutSec": cfg.get("timeout", 60),
}
]
fragment = {"version": 1, "hooks": copilot_hooks}
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(json.dumps(fragment, indent=2) + "\n", encoding="utf-8")
rel = str(config_path.relative_to(project_root))
if rel not in manifest.files:
manifest.record_existing(rel)
created.append(config_path)
elif fmt == "toml":
# Codex config.toml custom merge
lines: list[str] = []
for ev, cfg in filtered.items():
native = canonical_to_native[ev]
timeout = cfg.get("timeout", 60)
command = cfg.get("command", "")
lines.append(f'[[hooks.{native}]]')
lines.append(f'matcher = "{cfg.get("matcher", "*")}"')
lines.append('')
lines.append(f'[[hooks.{native}.hooks]]')
lines.append('type = "command"')
lines.append(f'command = \'python3 {EVENTS_DISPATCHER_REL} {command} {ev}\'')
lines.append(f'timeout = {timeout}')
lines.append('speckit_marker = true')
lines.append('')
_merge_toml_fragment(config_path, "\n".join(lines))
rel = str(config_path.relative_to(project_root))
if rel not in manifest.files:
manifest.record_existing(rel)
created.append(config_path)
elif fmt == "json-flat":
# Cursor hooks.json custom merge
cursor_hooks = {}
for ev, cfg in filtered.items():
native = canonical_to_native[ev]
command = cfg.get("command", "")
cursor_hooks[native] = [
{
"command": f"python3 {EVENTS_DISPATCHER_REL} {command} {ev}",
"type": "command",
"timeout": cfg.get("timeout", 60),
"matcher": cfg.get("matcher", "*"),
_SPECKIT_MARKER: True,
}
]
_merge_json_fragment(config_path, cursor_hooks)
rel = str(config_path.relative_to(project_root))
if rel not in manifest.files:
manifest.record_existing(rel)
created.append(config_path)
elif fmt == "json-nested":
# Claude/Qwen/Gemini/Devin/Tabnine nested config JSON merge
nested_hooks = {}
bridge_path_prefix = "${CLAUDE_PROJECT_DIR}/" if integration.key == "claude" else ""
for ev, cfg in filtered.items():
native = canonical_to_native[ev]
command = cfg.get("command", "")
nested_hooks[native] = [
{
"matcher": cfg.get("matcher", "*"),
"hooks": [
{
"type": "command",
"command": "python3",
"args": [
bridge_path_prefix + EVENTS_DISPATCHER_REL,
command,
ev,
],
"timeout": cfg.get("timeout", 60),
_SPECKIT_MARKER: True,
}
],
}
]
_merge_json_fragment(config_path, nested_hooks)
rel = str(config_path.relative_to(project_root))
if rel not in manifest.files:
manifest.record_existing(rel)
created.append(config_path)
return created
def remove_integration_events(integration: IntegrationBase, project_root: Path, manifest: IntegrationManifest) -> None:
"""Remove Specify-authored event entries from native config."""
fmt = getattr(integration, "events_format", None)
config_file = getattr(integration, "events_config_file", None)
if config_file:
config_path = project_root / config_file
if config_path.exists():
if fmt == "copilot-json":
# Clean delete of dedicated file
config_path.unlink(missing_ok=True)
manifest.remove(config_file)
elif fmt == "toml":
_remove_toml_entries(config_path)
elif fmt in ("json-nested", "json-flat"):
_remove_json_entries(config_path)
elif fmt == "ts-plugin":
_remove_opencode_entries(config_path)
# Clean up dispatcher script
dispatcher_rel = EVENTS_DISPATCHER_REL
if dispatcher_rel in manifest.files:
dispatcher_path = project_root / dispatcher_rel
if dispatcher_path.exists():
dispatcher_path.unlink(missing_ok=True)
manifest.remove(dispatcher_rel)
# Clean up opencode TS plugin
if integration.key == "opencode":
plugin_rel = ".opencode/plugin/speckit-events.ts"
if plugin_rel in manifest.files:
plugin_path = project_root / plugin_rel
if plugin_path.exists():
plugin_path.unlink(missing_ok=True)
manifest.remove(plugin_rel)
def events_stale_exclusions(integration_key: str) -> set[str]:
"""Return project-relative paths to protect from stale cleanup."""
from .integrations import get_integration
integration = get_integration(integration_key)
if not integration:
return set()
exclusions = set()
config_file = getattr(integration, "events_config_file", None)
if config_file:
exclusions.add(config_file)
if integration_key == "opencode":
exclusions.add(".opencode/plugin/speckit-events.ts")
return exclusions
# -- Manifest validation ---------------------------------------------------
def validate_events(data: dict[str, Any]) -> None:
"""Validate ``events`` field in extension manifest data."""
from .extensions import ValidationError
events = data.get("events")
if "events" in data and not isinstance(events, dict):
raise ValidationError("Invalid events: expected a mapping")
if events:
for event_name, event_config in events.items():
if not isinstance(event_config, dict):
raise ValidationError(
f"Invalid event '{event_name}': expected a mapping"
)
if not event_config.get("command"):
raise ValidationError(
f"Event '{event_name}' missing required 'command' field"
)
if event_name not in CANONICAL_EVENTS:
raise ValidationError(
f"Unknown event '{event_name}': "
f"must be one of {sorted(CANONICAL_EVENTS)}"
)
def has_events(data: dict[str, Any]) -> bool:
"""Return True if ``events`` is present and non-empty."""
return bool(data.get("events"))
# -- Helper merging functions ----------------------------------------------
def _build_opencode_plugin(filtered_events: dict[str, dict[str, Any]], canonical_to_native: dict[str, str]) -> str:
event_entries: list[str] = []
plugin_returns: list[str] = []
event_handlers: list[str] = []
for ev, cfg in filtered_events.items():
native = canonical_to_native[ev]
command = cfg.get("command", "")
matcher = cfg.get("matcher", "*")
if native.startswith("tool.execute."):
ts_hook = native
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 = "{"
event_entries.append(
f"function _{ev}(input: any) {match_cond}\n"
f" runEvent('{command}', '{ev}', input);\n"
f" }}"
)
plugin_returns.append(
f" \"{ts_hook}\": async (input: any, output: any) => {{\n"
f" _{ev}(input);\n"
f" }},"
)
else:
event_entries.append(
f"function _{ev}(input: any) {{\n"
f" runEvent('{command}', '{ev}', input);\n"
f" }}"
)
event_handlers.append(
f" if (event.type === '{native}') {{ _{ev}(event); }}"
)
if event_handlers:
plugin_returns.append(
f" event: async ({{ event }}) => {{\n"
+ "\n".join(event_handlers) + "\n"
f" }},"
)
return _TS_PLUGIN_TEMPLATE.format(
event_entries="\n\n".join(event_entries),
plugin_returns="\n".join(plugin_returns),
)
def _merge_opencode_plugin_ref(config_path: Path, ref: str) -> None:
existing: dict = {}
if config_path.exists():
try:
existing = json.loads(config_path.read_text(encoding="utf-8"))
except Exception:
existing = {}
if not isinstance(existing, dict):
existing = {}
plugins = existing.get("plugin", [])
if not isinstance(plugins, list):
plugins = []
if ref not in plugins:
plugins.append(ref)
existing["plugin"] = plugins
config_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
def _remove_opencode_entries(config_path: Path) -> None:
if not config_path.exists():
return
try:
existing = json.loads(config_path.read_text(encoding="utf-8"))
except Exception:
return
if not isinstance(existing, dict):
return
plugins = existing.get("plugin", [])
if isinstance(plugins, list):
ref = "./.opencode/plugin/speckit-events.ts"
plugins = [p for p in plugins if p != ref]
if plugins:
existing["plugin"] = plugins
else:
existing.pop("plugin", None)
config_path.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
def _merge_toml_fragment(dst: Path, fragment: str) -> None:
existing = ""
if dst.exists():
existing = dst.read_text(encoding="utf-8")
existing = re.sub(
r'\[\[hooks\.\w+\]\]\n(?:(?!\[\[hooks\.\w+\]\]).)*?speckit_marker = true\n*',
"",
existing,
flags=re.DOTALL,
)
dst.parent.mkdir(parents=True, exist_ok=True)
dst.write_text(existing.rstrip() + "\n\n" + fragment + "\n", encoding="utf-8")
def _remove_toml_entries(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")
def _merge_json_fragment(dst: Path, new_hooks: dict) -> None:
existing: dict = {}
if dst.exists():
try:
existing = json.loads(dst.read_text(encoding="utf-8"))
except Exception:
pass
if not isinstance(existing, dict):
existing = {}
hooks_key = "hooks"
existing_hooks = existing.get(hooks_key, {})
if not isinstance(existing_hooks, dict):
existing_hooks = {}
for event, entries in new_hooks.items():
existing_list = existing_hooks.get(event, [])
if not isinstance(existing_list, list):
existing_list = []
# Filter out existing entries carrying our marker
existing_list = [e for e in existing_list if not _has_marker(e)]
existing_list.extend(entries)
existing_hooks[event] = existing_list
existing[hooks_key] = existing_hooks
dst.parent.mkdir(parents=True, exist_ok=True)
dst.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
def _remove_json_entries(dst: Path) -> None:
try:
existing = json.loads(dst.read_text(encoding="utf-8"))
except Exception:
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
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)
elif _has_marker(entry):
pass
else:
kept_entries.append(entry)
if kept_entries:
cleaned[event] = kept_entries
if cleaned:
existing["hooks"] = cleaned
else:
existing.pop("hooks", None)
dst.write_text(json.dumps(existing, indent=2) + "\n", encoding="utf-8")
def _has_marker(entry: Any) -> bool:
if isinstance(entry, dict):
return entry.get(_SPECKIT_MARKER, False) is True
return False

9
src/specify_cli/extensions/__init__.py
View File

@@ -300,17 +300,22 @@ class ExtensionManifest:
provides = self.data["provides"]
commands = provides.get("commands", [])
hooks = self.data.get("hooks")
events = self.data.get("events")
if "commands" in provides and not isinstance(commands, list):
raise ValidationError("Invalid provides.commands: expected a list")
if "hooks" in self.data and not isinstance(hooks, dict):
raise ValidationError("Invalid hooks: expected a mapping")
if "events" in self.data:
from ..events import validate_events
validate_events(self.data)
has_commands = bool(commands)
has_hooks = bool(hooks)
has_events = bool(events)
if not has_commands and not has_hooks:
raise ValidationError("Extension must provide at least one command or hook")
if not has_commands and not has_hooks and not has_events:
raise ValidationError("Extension must provide at least one command, hook, or event")
# Validate hook values (if present).
# Each event is a single mapping or a list of mappings.

975
src/specify_cli/integrations/_hooks.py
View File

@@ -1,975 +0,0 @@
"""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"))

9
src/specify_cli/integrations/_install_commands.py
View File

@@ -139,12 +139,21 @@ def integration_install(
integration.key, project_root, version=_get_speckit_version()
)
from ..events import resolve_events
events_map = resolve_events(
integration.key,
integration.config,
project_root,
parsed_options,
)
try:
integration.setup(
project_root, manifest,
parsed_options=parsed_options,
script_type=selected_script,
raw_options=raw_options,
events=events_map,
)
manifest.save()
new_installed = _dedupe_integration_keys([*installed_keys, integration.key])

16
src/specify_cli/integrations/_migrate_commands.py
View File

@@ -342,12 +342,20 @@ def integration_switch(
target_integration.key, project_root, version=_get_speckit_version()
)
from ..events import resolve_events
events_map = resolve_events(
target_integration.key,
target_integration.config,
project_root,
parsed_options,
)
try:
target_integration.setup(
project_root, manifest,
parsed_options=parsed_options,
script_type=selected_script,
raw_options=raw_options,
events=events_map,
)
manifest.save()
_set_default_integration(
@@ -566,6 +574,13 @@ def integration_upgrade(
console.print(f"Upgrading integration: [cyan]{key}[/cyan]")
new_manifest = IntegrationManifest(key, project_root, version=_get_speckit_version())
from ..events import resolve_events
events_map = resolve_events(
key,
integration.config,
project_root,
parsed_options,
)
try:
integration.setup(
project_root,
@@ -573,6 +588,7 @@ def integration_upgrade(
parsed_options=parsed_options,
script_type=selected_script,
raw_options=raw_options,
events=events_map,
)
settings = _with_integration_setting(
current,

77
src/specify_cli/integrations/base.py
View File

@@ -29,7 +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
from ..events import install_integration_events, remove_integration_events
if TYPE_CHECKING:
from .manifest import IntegrationManifest
@@ -159,7 +159,17 @@ class IntegrationBase(ABC):
@classmethod
def options(cls) -> list[IntegrationOption]:
"""Return options this integration accepts. Default: none."""
return []
opts = []
if bool(getattr(cls, "CANONICAL_TO_NATIVE", None) and getattr(cls, "events_config_file", None)):
opts.append(
IntegrationOption(
"--events",
is_flag=False,
default="true",
help="Enable/disable runtime events (true|false, default: true)",
)
)
return opts
def effective_invoke_separator(
self,
@@ -480,7 +490,11 @@ class IntegrationBase(ABC):
tracking) would otherwise be deleted even though they are still
managed. Subclasses list such paths here to protect them.
"""
return set()
exclusions = set()
if self.supports_events():
from ..events import events_stale_exclusions
exclusions.update(events_stale_exclusions(self.key))
return exclusions
def commands_dest(self, project_root: Path) -> Path:
"""Return the absolute path to the commands output directory.
@@ -903,9 +917,32 @@ class IntegrationBase(ABC):
Returns ``(removed, skipped)`` file lists.
"""
remove_integration_hooks(self, project_root, manifest)
self.remove_events(project_root, manifest)
return manifest.uninstall(project_root, force=force)
def emit_events(
self,
project_root: Path,
manifest: IntegrationManifest,
events: dict[str, dict[str, Any]] | None = None,
parsed_options: dict[str, Any] | None = None,
**opts: Any,
) -> list[Path]:
"""Emit native event configuration for this integration."""
return install_integration_events(self, project_root, manifest, events or {})
def remove_events(
self,
project_root: Path,
manifest: IntegrationManifest,
) -> None:
"""Remove Specify-authored event entries from native config."""
remove_integration_events(self, project_root, manifest)
def supports_events(self) -> bool:
"""Return True if this integration supports agent-native events."""
return bool(getattr(self, "CANONICAL_TO_NATIVE", None) and getattr(self, "events_config_file", None))
# -- Convenience helpers for subclasses -------------------------------
def install(
@@ -1010,11 +1047,11 @@ class MarkdownIntegration(IntegrationBase):
created.append(dst_file)
# Install agent runtime hooks
hook_files = install_integration_hooks(
self, project_root, manifest, parsed_options
# Install agent runtime events
event_files = self.emit_events(
project_root, manifest, events=opts.get("events"), parsed_options=parsed_options
)
created.extend(hook_files)
created.extend(event_files)
return created
@@ -1223,11 +1260,11 @@ class TomlIntegration(IntegrationBase):
created.append(dst_file)
# Install agent runtime hooks
hook_files = install_integration_hooks(
self, project_root, manifest, parsed_options
# Install agent runtime events
event_files = self.emit_events(
project_root, manifest, events=opts.get("events"), parsed_options=parsed_options
)
created.extend(hook_files)
created.extend(event_files)
return created
@@ -1465,11 +1502,11 @@ class YamlIntegration(IntegrationBase):
created.append(dst_file)
# Install agent runtime hooks
hook_files = install_integration_hooks(
self, project_root, manifest, parsed_options
# Install agent runtime events
event_files = self.emit_events(
project_root, manifest, events=opts.get("events"), parsed_options=parsed_options
)
created.extend(hook_files)
created.extend(event_files)
return created
@@ -1706,10 +1743,10 @@ class SkillsIntegration(IntegrationBase):
created.append(dst)
# Install agent runtime hooks
hook_files = install_integration_hooks(
self, project_root, manifest, parsed_options
# Install agent runtime events
event_files = self.emit_events(
project_root, manifest, events=opts.get("events"), parsed_options=parsed_options
)
created.extend(hook_files)
created.extend(event_files)
return created

11
src/specify_cli/integrations/claude/__init__.py
View File

@@ -54,6 +54,17 @@ class ClaudeIntegration(SkillsIntegration):
}
multi_install_safe = True
CANONICAL_TO_NATIVE = {
"session_start": "SessionStart",
"pre_tool_use": "PreToolUse",
"post_tool_use": "PostToolUse",
"session_end": "SessionEnd",
"user_prompt_submit": "UserPromptSubmit",
"stop": "Stop",
}
events_config_file = ".claude/settings.json"
events_format = "json-nested"
@staticmethod
def inject_argument_hint(content: str, hint: str) -> str:
"""Insert ``argument-hint`` after the first ``description:`` in YAML frontmatter.

19
src/specify_cli/integrations/codex/__init__.py
View File

@@ -29,6 +29,17 @@ class CodexIntegration(SkillsIntegration):
dev_no_symlink = True
multi_install_safe = True
CANONICAL_TO_NATIVE = {
"session_start": "SessionStart",
"pre_tool_use": "PreToolUse",
"post_tool_use": "PostToolUse",
"session_end": "SessionEnd",
"user_prompt_submit": "UserPromptSubmit",
"stop": "Stop",
}
events_config_file = ".codex/config.toml"
events_format = "toml"
def build_exec_args(
self,
prompt: str,
@@ -49,11 +60,13 @@ class CodexIntegration(SkillsIntegration):
@classmethod
def options(cls) -> list[IntegrationOption]:
return [
opts = super().options()
opts.append(
IntegrationOption(
"--skills",
is_flag=True,
default=True,
help="Install as agent skills (default for Codex)",
),
]
)
)
return opts

30
src/specify_cli/integrations/copilot/__init__.py
View File

@@ -118,6 +118,16 @@ class CopilotIntegration(IntegrationBase):
"extension": ".agent.md",
}
CANONICAL_TO_NATIVE = {
"session_start": "sessionStart",
"pre_tool_use": "preToolUse",
"post_tool_use": "postToolUse",
"session_end": "sessionEnd",
"user_prompt_submit": "userPromptSubmitted",
}
events_config_file = ".github/hooks/speckit.json"
events_format = "copilot-json"
# Mutable flag set by setup() — indicates the active scaffolding mode.
_skills_mode: bool = False
@@ -328,7 +338,9 @@ class CopilotIntegration(IntegrationBase):
be flagged stale and deleted, destroying user settings (and the file
the integration still manages).
"""
return {".vscode/settings.json"}
exclusions = super().stale_cleanup_exclusions()
exclusions.add(".vscode/settings.json")
return exclusions
def post_process_skill_content(self, content: str) -> str:
"""Inject shared hook guidance into Copilot skill content.
@@ -355,10 +367,18 @@ class CopilotIntegration(IntegrationBase):
parsed_options = parsed_options or {}
self._skills_mode = bool(parsed_options.get("skills"))
if self._skills_mode:
return self._setup_skills(project_root, manifest, parsed_options, **opts)
if "skills" not in parsed_options:
_warn_legacy_markdown_default()
return self._setup_default(project_root, manifest, parsed_options, **opts)
created = self._setup_skills(project_root, manifest, parsed_options, **opts)
else:
if "skills" not in parsed_options:
_warn_legacy_markdown_default()
created = self._setup_default(project_root, manifest, parsed_options, **opts)
# Install agent runtime events
event_files = self.emit_events(
project_root, manifest, events=opts.get("events"), parsed_options=parsed_options
)
created.extend(event_files)
return created
def _setup_default(
self,

19
src/specify_cli/integrations/cursor_agent/__init__.py
View File

@@ -38,6 +38,17 @@ class CursorAgentIntegration(SkillsIntegration):
multi_install_safe = True
CANONICAL_TO_NATIVE = {
"session_start": "sessionStart",
"pre_tool_use": "preToolUse",
"post_tool_use": "postToolUse",
"session_end": "sessionEnd",
"user_prompt_submit": "beforeSubmitPrompt",
"stop": "stop",
}
events_config_file = ".cursor/hooks.json"
events_format = "json-flat"
def build_exec_args(
self,
prompt: str,
@@ -92,11 +103,13 @@ class CursorAgentIntegration(SkillsIntegration):
@classmethod
def options(cls) -> list[IntegrationOption]:
return [
opts = super().options()
opts.append(
IntegrationOption(
"--skills",
is_flag=True,
default=True,
help="Install as agent skills (recommended for Cursor)",
),
]
)
)
return opts

11
src/specify_cli/integrations/devin/__init__.py
View File

@@ -31,6 +31,17 @@ class DevinIntegration(SkillsIntegration):
"extension": "/SKILL.md",
}
CANONICAL_TO_NATIVE = {
"session_start": "SessionStart",
"pre_tool_use": "PreToolUse",
"post_tool_use": "PostToolUse",
"session_end": "SessionEnd",
"user_prompt_submit": "UserPromptSubmit",
"stop": "Stop",
}
events_config_file = ".devin/hooks.v1.json"
events_format = "json-nested"
def build_exec_args(
self,
prompt: str,

10
src/specify_cli/integrations/gemini/__init__.py
View File

@@ -19,3 +19,13 @@ class GeminiIntegration(TomlIntegration):
"extension": ".toml",
}
multi_install_safe = True
CANONICAL_TO_NATIVE = {
"session_start": "SessionStart",
"pre_tool_use": "BeforeTool",
"post_tool_use": "AfterTool",
"session_end": "SessionEnd",
"stop": "AfterAgent",
}
events_config_file = ".gemini/settings.json"
events_format = "json-nested"

9
src/specify_cli/integrations/opencode/__init__.py
View File

@@ -20,6 +20,15 @@ class OpencodeIntegration(MarkdownIntegration):
"extension": ".md",
}
CANONICAL_TO_NATIVE = {
"pre_tool_use": "tool.execute.before",
"post_tool_use": "tool.execute.after",
"session_start": "session.created",
"session_end": "session.deleted",
}
events_config_file = "opencode.json"
events_format = "ts-plugin"
def build_exec_args(
self,
prompt: str,

11
src/specify_cli/integrations/qwen/__init__.py
View File

@@ -19,3 +19,14 @@ class QwenIntegration(MarkdownIntegration):
"extension": ".md",
}
multi_install_safe = True
CANONICAL_TO_NATIVE = {
"session_start": "SessionStart",
"pre_tool_use": "PreToolUse",
"post_tool_use": "PostToolUse",
"session_end": "SessionEnd",
"user_prompt_submit": "UserPromptSubmit",
"stop": "Stop",
}
events_config_file = ".qwen/settings.json"
events_format = "json-nested"

9
src/specify_cli/integrations/tabnine/__init__.py
View File

@@ -19,3 +19,12 @@ class TabnineIntegration(TomlIntegration):
"extension": ".toml",
}
multi_install_safe = True
CANONICAL_TO_NATIVE = {
"session_start": "SessionStart",
"pre_tool_use": "BeforeTool",
"post_tool_use": "AfterTool",
"session_end": "SessionEnd",
}
events_config_file = ".tabnine/agent/settings.json"
events_format = "json-nested"

374
tests/integrations/test_events.py Normal file
View File

@@ -0,0 +1,374 @@
"""Tests for events module: integration runtime events."""
from __future__ import annotations
import json
import platform
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
import yaml
from specify_cli.events import (
CANONICAL_EVENTS,
collect_extension_events,
events_stale_exclusions,
install_integration_events,
remove_integration_events,
resolve_events,
validate_events,
resolve_and_run_event_command,
)
from specify_cli.integrations.base import IntegrationBase
from specify_cli.integrations.manifest import IntegrationManifest
from specify_cli.integrations.claude import ClaudeIntegration
from specify_cli.integrations.cursor_agent import CursorAgentIntegration
from specify_cli.integrations.codex import CodexIntegration
from specify_cli.integrations.opencode import OpencodeIntegration
from specify_cli.integrations.qwen import QwenIntegration
from specify_cli.integrations.gemini import GeminiIntegration
from specify_cli.integrations.devin import DevinIntegration
from specify_cli.integrations.tabnine import TabnineIntegration
from specify_cli.integrations.copilot import CopilotIntegration
# -- resolve_events --------------------------------------------------------
class TestResolveEvents:
"""Test the 4-layer event resolution chain."""
def test_layer1_disabled_returns_empty(self, tmp_path):
"""--events false returns empty dict."""
result = resolve_events(
"claude",
{"events": {"post_tool_use": {"command": "speckit.tdd.validate"}}},
tmp_path,
{"events": "false"},
)
assert result == {}
def test_layer4_built_in_defaults(self, tmp_path):
"""Returns baseline defaults when no overrides exist."""
result = resolve_events(
"claude",
{"events": {"post_tool_use": {"command": "speckit.tdd.validate"}}},
tmp_path,
None,
)
assert result == {"post_tool_use": {"command": "speckit.tdd.validate"}}
def test_layer3_extension_events_appended(self, tmp_path):
"""Extension-declared events are resolved and appended."""
ext_dir = tmp_path / ".specify" / "extensions" / "my-ext"
ext_dir.mkdir(parents=True)
ext_yml = ext_dir / "extension.yml"
ext_yml.write_text(
"events:\n session_start:\n command: speckit.my-ext.boot\n",
encoding="utf-8",
)
result = resolve_events(
"claude",
{"events": {"post_tool_use": {"command": "speckit.tdd.validate"}}},
tmp_path,
None,
)
assert "post_tool_use" in result
assert "session_start" in result
assert result["session_start"] == {"command": "speckit.my-ext.boot"}
def test_layer2_yaml_override_replaces(self, tmp_path):
"""integration-events.yml override replaces baseline entirely."""
override_file = tmp_path / ".specify" / "integration-events.yml"
override_file.parent.mkdir(parents=True, exist_ok=True)
override_file.write_text(
"integrations:\n"
" claude:\n"
" events:\n"
" stop:\n"
" command: speckit.override.stop\n",
encoding="utf-8",
)
result = resolve_events(
"claude",
{"events": {"post_tool_use": {"command": "speckit.tdd.validate"}}},
tmp_path,
None,
)
assert result == {"stop": {"command": "speckit.override.stop"}}
def test_layer2_empty_events_disables(self, tmp_path):
"""Empty events override disables events."""
override_file = tmp_path / ".specify" / "integration-events.yml"
override_file.parent.mkdir(parents=True, exist_ok=True)
override_file.write_text(
"integrations:\n"
" claude:\n"
" events: {}\n",
encoding="utf-8",
)
result = resolve_events(
"claude",
{"events": {"post_tool_use": {"command": "speckit.tdd.validate"}}},
tmp_path,
None,
)
assert result == {}
def test_no_config_no_events(self, tmp_path):
"""Safe fallback with empty config/options."""
result = resolve_events("claude", None, tmp_path, None)
assert result == {}
# -- collect_extension_events -----------------------------------------------
class TestCollectExtensionEvents:
"""Test scanning extension.yml files for events: declarations."""
def test_no_extensions_dir(self, tmp_path):
assert collect_extension_events(tmp_path) == {}
def test_no_events_in_extension(self, tmp_path):
ext_dir = tmp_path / ".specify" / "extensions" / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "extension.yml").write_text("extension:\n id: my-ext\n", encoding="utf-8")
assert collect_extension_events(tmp_path) == {}
def test_events_collected(self, tmp_path):
ext_dir = tmp_path / ".specify" / "extensions" / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "extension.yml").write_text(
"events:\n pre_tool_use:\n command: speckit.my-ext.check\n",
encoding="utf-8",
)
result = collect_extension_events(tmp_path)
assert result == {"pre_tool_use": {"command": "speckit.my-ext.check"}}
def test_invalid_yaml_skipped(self, tmp_path):
ext_dir = tmp_path / ".specify" / "extensions" / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "extension.yml").write_text("invalid: - - -", encoding="utf-8")
assert collect_extension_events(tmp_path) == {}
# -- Class-driven mappings --------------------------------------------------
class TestCanonicalEventMapping:
"""Verify registry-driven mapping is correct on integration classes."""
def test_claude_identity(self):
integration = ClaudeIntegration()
assert integration.supports_events() is True
assert integration.CANONICAL_TO_NATIVE["pre_tool_use"] == "PreToolUse"
assert integration.CANONICAL_TO_NATIVE["session_start"] == "SessionStart"
def test_cursor_camelcase(self):
integration = CursorAgentIntegration()
assert integration.supports_events() is True
assert integration.CANONICAL_TO_NATIVE["pre_tool_use"] == "preToolUse"
assert integration.CANONICAL_TO_NATIVE["user_prompt_submit"] == "beforeSubmitPrompt"
def test_opencode_limited(self):
integration = OpencodeIntegration()
assert integration.supports_events() is True
assert integration.CANONICAL_TO_NATIVE["pre_tool_use"] == "tool.execute.before"
assert "stop" not in integration.CANONICAL_TO_NATIVE
def test_copilot_mapping(self):
integration = CopilotIntegration()
assert integration.supports_events() is True
assert integration.CANONICAL_TO_NATIVE["session_start"] == "sessionStart"
assert integration.CANONICAL_TO_NATIVE["user_prompt_submit"] == "userPromptSubmitted"
# -- validate_events --------------------------------------------------------
class TestValidateEvents:
"""Test manifest validation."""
def test_unknown_event_rejected(self):
from specify_cli.extensions import ValidationError
data = {"events": {"unknown_event": {"command": "speckit.tdd.validate"}}}
with pytest.raises(ValidationError) as exc:
validate_events(data)
assert "Unknown event" in str(exc.value)
def test_known_event_accepted(self):
data = {"events": {"pre_tool_use": {"command": "speckit.tdd.validate"}}}
validate_events(data) # no raise
def test_all_canonical_events_accepted(self):
data = {
"events": {
name: {"command": "speckit.test"}
for name in CANONICAL_EVENTS
}
}
validate_events(data) # no raise
# -- Claude settings JSON merging -------------------------------------------
class TestClaudeJsonMerging:
"""Test Claude settings JSON merging and cleanup."""
def test_merge_into_empty_file(self, tmp_path):
integration = ClaudeIntegration()
manifest = MagicMock(spec=IntegrationManifest)
manifest.files = {}
manifest.record_file = MagicMock()
manifest.record_existing = MagicMock()
events = {
"pre_tool_use": {"command": "speckit.tdd.validate", "matcher": "Edit|Write"},
}
install_integration_events(integration, tmp_path, manifest, events)
config_path = tmp_path / ".claude/settings.json"
assert config_path.is_file()
data = json.loads(config_path.read_text())
assert "hooks" in data
assert "PreToolUse" in data["hooks"]
assert data["hooks"]["PreToolUse"][0]["matcher"] == "Edit|Write"
def test_remove_preserves_user_hooks(self, tmp_path):
integration = ClaudeIntegration()
manifest = MagicMock(spec=IntegrationManifest)
manifest.files = {}
manifest.record_file = MagicMock()
manifest.record_existing = MagicMock()
# Pre-seed user setting
config_path = tmp_path / ".claude/settings.json"
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(
json.dumps(
{
"hooks": {
"PreToolUse": [
{
"matcher": "Bash",
"hooks": [
{
"type": "command",
"command": "user-check",
}
],
}
]
}
}
)
)
events = {
"pre_tool_use": {"command": "speckit.tdd.validate"},
}
install_integration_events(integration, tmp_path, manifest, events)
remove_integration_events(integration, tmp_path, manifest)
data = json.loads(config_path.read_text())
assert "hooks" in data
assert "PreToolUse" in data["hooks"]
assert len(data["hooks"]["PreToolUse"]) == 1
assert data["hooks"]["PreToolUse"][0]["matcher"] == "Bash"
# -- Copilot events JSON writing --------------------------------------------
class TestCopilotJsonWriting:
"""Test Copilot dedicated .github/hooks/speckit.json generation."""
def test_copilot_json_generation(self, tmp_path):
integration = CopilotIntegration()
manifest = MagicMock(spec=IntegrationManifest)
manifest.files = {}
manifest.record_file = MagicMock()
manifest.record_existing = MagicMock()
events = {
"session_start": {"command": "speckit.agent-context.update", "timeout": 60},
}
install_integration_events(integration, tmp_path, manifest, events)
config_path = tmp_path / ".github/hooks/speckit.json"
assert config_path.is_file()
data = json.loads(config_path.read_text())
assert data["version"] == 1
assert "hooks" in data
assert "sessionStart" in data["hooks"]
entry = data["hooks"]["sessionStart"][0]
assert entry["type"] == "command"
assert "speckit.agent-context.update" in entry["bash"]
assert "session_start" in entry["bash"]
assert entry["timeoutSec"] == 60
# -- Opencode TS Plugin merging ---------------------------------------------
class TestOpencodePluginMerging:
"""Test Opencode typescript plugin generation."""
def test_opencode_ts_plugin_generation(self, tmp_path):
integration = OpencodeIntegration()
manifest = MagicMock(spec=IntegrationManifest)
manifest.files = {}
manifest.record_file = MagicMock()
manifest.record_existing = MagicMock()
events = {
"pre_tool_use": {"command": "speckit.tdd.validate", "matcher": "Edit"},
"session_start": {"command": "speckit.agent-context.update"},
}
install_integration_events(integration, tmp_path, manifest, events)
plugin_path = tmp_path / ".opencode/plugin/speckit-events.ts"
assert plugin_path.is_file()
content = plugin_path.read_text()
assert "runEvent" in content
assert "tool.execute.before" in content
assert "session.created" in content
assert "speckit.tdd.validate" in content
assert "speckit.agent-context.update" in content
# -- Command runner test (core execution) -----------------------------------
class TestCommandRunner:
"""Test the core command/script resolution and runner."""
def test_run_command_not_found(self, tmp_path):
code = resolve_and_run_event_command("nonexistent.command", "session_start", "{}", tmp_path)
assert code == 0 # no-ops gracefully
def test_run_command_resolves_and_executes(self, tmp_path):
# Create a mock core command md file
cmd_dir = tmp_path / ".specify" / "templates" / "commands"
cmd_dir.mkdir(parents=True)
cmd_file = cmd_dir / "test.md"
cmd_file.write_text(
"---\n"
"description: \"Test\"\n"
"scripts:\n"
" sh: scripts/test.sh\n"
"---\n"
"Body\n",
encoding="utf-8",
)
script_dir = tmp_path / ".specify" / "scripts"
script_dir.mkdir(parents=True)
script_file = script_dir / "test.sh"
script_file.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
script_file.chmod(0o755)
# Skip on Windows because sh is POSIX
if platform.system().lower().startswith("win"):
return
code = resolve_and_run_event_command("speckit.test", "session_start", "{}", tmp_path)
assert code == 0

678
tests/integrations/test_hooks.py
View File

@@ -1,678 +0,0 @@
"""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