From da6c20d9f0644651af1976197656d6b0e9202a87 Mon Sep 17 00:00:00 2001 From: Lior Kanfi Date: Sat, 25 Jul 2026 17:17:17 +0300 Subject: [PATCH] refactor: rework integration events per maintainer review - 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) --- src/specify_cli/__init__.py | 5 + src/specify_cli/commands/event.py | 36 + src/specify_cli/commands/init.py | 8 + src/specify_cli/events.py | 769 ++++++++++++++ src/specify_cli/extensions/__init__.py | 9 +- src/specify_cli/integrations/_hooks.py | 975 ------------------ .../integrations/_install_commands.py | 9 + .../integrations/_migrate_commands.py | 16 + src/specify_cli/integrations/base.py | 77 +- .../integrations/claude/__init__.py | 11 + .../integrations/codex/__init__.py | 19 +- .../integrations/copilot/__init__.py | 30 +- .../integrations/cursor_agent/__init__.py | 19 +- .../integrations/devin/__init__.py | 11 + .../integrations/gemini/__init__.py | 10 + .../integrations/opencode/__init__.py | 9 + src/specify_cli/integrations/qwen/__init__.py | 11 + .../integrations/tabnine/__init__.py | 9 + tests/integrations/test_events.py | 374 +++++++ tests/integrations/test_hooks.py | 678 ------------ 20 files changed, 1399 insertions(+), 1686 deletions(-) create mode 100644 src/specify_cli/commands/event.py create mode 100644 src/specify_cli/events.py delete mode 100644 src/specify_cli/integrations/_hooks.py create mode 100644 tests/integrations/test_events.py delete mode 100644 tests/integrations/test_hooks.py diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index 1d5a34073..c74db4087 100644 --- a/src/specify_cli/__init__.py +++ b/src/specify_cli/__init__.py @@ -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, diff --git a/src/specify_cli/commands/event.py b/src/specify_cli/commands/event.py new file mode 100644 index 000000000..bca6877dd --- /dev/null +++ b/src/specify_cli/commands/event.py @@ -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") diff --git a/src/specify_cli/commands/init.py b/src/specify_cli/commands/init.py index ccdd7b0a1..76af5022c 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -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() diff --git a/src/specify_cli/events.py b/src/specify_cli/events.py new file mode 100644 index 000000000..a4f126aa4 --- /dev/null +++ b/src/specify_cli/events.py @@ -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 diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index 00a6d92b4..5e3408ce2 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -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. diff --git a/src/specify_cli/integrations/_hooks.py b/src/specify_cli/integrations/_hooks.py deleted file mode 100644 index 7f24395b3..000000000 --- a/src/specify_cli/integrations/_hooks.py +++ /dev/null @@ -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")) diff --git a/src/specify_cli/integrations/_install_commands.py b/src/specify_cli/integrations/_install_commands.py index 4d7bb44f9..aba6f5117 100644 --- a/src/specify_cli/integrations/_install_commands.py +++ b/src/specify_cli/integrations/_install_commands.py @@ -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]) diff --git a/src/specify_cli/integrations/_migrate_commands.py b/src/specify_cli/integrations/_migrate_commands.py index 7116a744c..79895cf14 100644 --- a/src/specify_cli/integrations/_migrate_commands.py +++ b/src/specify_cli/integrations/_migrate_commands.py @@ -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, diff --git a/src/specify_cli/integrations/base.py b/src/specify_cli/integrations/base.py index 477c69834..7b0abde03 100644 --- a/src/specify_cli/integrations/base.py +++ b/src/specify_cli/integrations/base.py @@ -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 diff --git a/src/specify_cli/integrations/claude/__init__.py b/src/specify_cli/integrations/claude/__init__.py index 923a77607..39732794a 100644 --- a/src/specify_cli/integrations/claude/__init__.py +++ b/src/specify_cli/integrations/claude/__init__.py @@ -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. diff --git a/src/specify_cli/integrations/codex/__init__.py b/src/specify_cli/integrations/codex/__init__.py index 7d1ff86e2..2ffa59ca4 100644 --- a/src/specify_cli/integrations/codex/__init__.py +++ b/src/specify_cli/integrations/codex/__init__.py @@ -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 diff --git a/src/specify_cli/integrations/copilot/__init__.py b/src/specify_cli/integrations/copilot/__init__.py index ce41c00d6..38479c2bc 100644 --- a/src/specify_cli/integrations/copilot/__init__.py +++ b/src/specify_cli/integrations/copilot/__init__.py @@ -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, diff --git a/src/specify_cli/integrations/cursor_agent/__init__.py b/src/specify_cli/integrations/cursor_agent/__init__.py index 07f2a6318..58bd89b21 100644 --- a/src/specify_cli/integrations/cursor_agent/__init__.py +++ b/src/specify_cli/integrations/cursor_agent/__init__.py @@ -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 diff --git a/src/specify_cli/integrations/devin/__init__.py b/src/specify_cli/integrations/devin/__init__.py index 0d60bc954..6b2b925df 100644 --- a/src/specify_cli/integrations/devin/__init__.py +++ b/src/specify_cli/integrations/devin/__init__.py @@ -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, diff --git a/src/specify_cli/integrations/gemini/__init__.py b/src/specify_cli/integrations/gemini/__init__.py index 9a459862a..e422a7426 100644 --- a/src/specify_cli/integrations/gemini/__init__.py +++ b/src/specify_cli/integrations/gemini/__init__.py @@ -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" diff --git a/src/specify_cli/integrations/opencode/__init__.py b/src/specify_cli/integrations/opencode/__init__.py index 0f734b7f4..660fd0b5f 100644 --- a/src/specify_cli/integrations/opencode/__init__.py +++ b/src/specify_cli/integrations/opencode/__init__.py @@ -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, diff --git a/src/specify_cli/integrations/qwen/__init__.py b/src/specify_cli/integrations/qwen/__init__.py index 1e8c15bf9..2cd879f06 100644 --- a/src/specify_cli/integrations/qwen/__init__.py +++ b/src/specify_cli/integrations/qwen/__init__.py @@ -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" diff --git a/src/specify_cli/integrations/tabnine/__init__.py b/src/specify_cli/integrations/tabnine/__init__.py index 9edf1e160..265175531 100644 --- a/src/specify_cli/integrations/tabnine/__init__.py +++ b/src/specify_cli/integrations/tabnine/__init__.py @@ -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" diff --git a/tests/integrations/test_events.py b/tests/integrations/test_events.py new file mode 100644 index 000000000..5509f56ba --- /dev/null +++ b/tests/integrations/test_events.py @@ -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 diff --git a/tests/integrations/test_hooks.py b/tests/integrations/test_hooks.py deleted file mode 100644 index cb1dd1c6e..000000000 --- a/tests/integrations/test_hooks.py +++ /dev/null @@ -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