github-spec-kit/src/specify_cli/events.py

"""Agent runtime events for integrations.

Provides:
- ``resolve_events`` — layered event resolution (CLI flag → YAML override → extension-declared → built-in).
- ``collect_extension_events`` — scan installed extension.yml files for ``events:``.
- ``install_integration_events`` / ``remove_integration_events`` — entry points called from ``IntegrationBase.setup()`` / ``teardown()``.
"""

from __future__ import annotations

import json
import logging
import os
import re
import shlex
import shutil
import sys
import subprocess
import platform
from pathlib import Path
from typing import TYPE_CHECKING, Any

import yaml

if TYPE_CHECKING:
    from .integrations.base import IntegrationBase
    from .integrations.manifest import IntegrationManifest

logger = logging.getLogger(__name__)

# -- Constants -------------------------------------------------------------

EVENTS_DISPATCHER_DIR = Path(".specify")
EVENTS_DISPATCHER_FILENAME = "events.py"
# POSIX-form (forward-slash) relative path so it matches manifest keys, which
# are always stored in POSIX form (record_file/record_existing normalize via
# .as_posix()). On Windows, str(Path(".specify")/"events.py") yields
# ".specify\\events.py", which never matched a manifest key, so the shared-
# dispatcher manifest-claim drop was skipped and uninstall(force=True) deleted
# the dispatcher another integration still depended on.
EVENTS_DISPATCHER_REL = (EVENTS_DISPATCHER_DIR / EVENTS_DISPATCHER_FILENAME).as_posix()

YAML_OVERRIDE_FILENAME = Path(".specify") / "integration-events.yml"

_SPECKIT_MARKER = "__speckit_event__"

# Buffer (seconds) added to the native hook timeout so the agent's outer cap
# fires after the dispatcher's inner subprocess timeout, letting the inner
# kill its child cleanly instead of being killed mid-flight (which orphans
# the grandchild script process). The dispatcher receives the raw seconds
# (no buffer); the native config field gets seconds + buffer (R2).
EVENT_TIMEOUT_BUFFER = 5

# Canonical event names (snake_case)
CANONICAL_EVENTS = frozenset({
    "session_start",
    "pre_tool_use",
    "post_tool_use",
    "session_end",
    "user_prompt_submit",
    "stop",
})

# -- Events Dispatcher template ---------------------------------------------

_EVENTS_DISPATCHER_TEMPLATE = '''#!/usr/bin/env python3
"""Specify CLI Event Dispatcher — dispatches agent runtime events.

Generated by: specify integration install/upgrade
Do not edit manually.

Self-contained: it prefers `specify_cli` when the package is importable
(durable pip/pipx/uv-tool install) and falls back to an inline stdlib-only
resolver when Spec Kit is not installed at runtime — e.g. a one-time `uvx`
init whose environment is discarded after `specify init` finishes (R1). In
both cases it resolves the event's command template and runs its script
directly, without requiring a persistent `specify` executable on PATH.
"""
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
from pathlib import Path


def _find_command_template(command_name, project_root):
    """Locate the command's .md template. Returns (path, ext_id|None)."""
    exts_dir = project_root / ".specify" / "extensions"
    disabled_ids = set()
    registry_file = exts_dir / ".registry"
    if registry_file.is_file():
        try:
            reg_data = json.loads(registry_file.read_text(encoding="utf-8"))
            for ext_id, meta in reg_data.get("extensions", {}).items():
                if isinstance(meta, dict) and meta.get("enabled") is False:
                    disabled_ids.add(ext_id)
        except Exception:
            pass

    stem = command_name.replace("speckit.", "").replace("spec.", "")

    # 1. Manifest-driven resolution from extension.yml in enabled extensions (Suppressed #1)
    if exts_dir.is_dir():
        for ext_dir in sorted(exts_dir.iterdir()):
            if not ext_dir.is_dir() or ext_dir.name in disabled_ids:
                continue
            ext_yml = ext_dir / "extension.yml"
            if ext_yml.is_file():
                try:
                    yml_text = ext_yml.read_text(encoding="utf-8")
                    cur_name = None
                    cur_file = None
                    in_provides = False
                    in_commands = False
                    for line in yml_text.splitlines():
                        stripped = line.strip()
                        if stripped == "provides:":
                            in_provides = True
                            continue
                        if in_provides and stripped == "commands:":
                            in_commands = True
                            continue
                        if in_commands and stripped and not line[0].isspace():
                            in_provides = False
                            in_commands = False
                            continue
                        if in_commands:
                            if "name:" in line:
                                cur_name = line.split("name:", 1)[1].strip().strip('"').strip("'")
                            if "file:" in line:
                                cur_file = line.split("file:", 1)[1].strip().strip('"').strip("'")
                            if cur_name and cur_file:
                                if cur_name == command_name:
                                    candidate = ext_dir / cur_file
                                    if candidate.exists():
                                        return candidate, ext_dir.name
                                cur_name = None
                                cur_file = None
                except Exception:
                    pass

    # 2. On-disk extension commands by file stem (non-disabled extensions)
    if exts_dir.is_dir():
        for ext_dir in sorted(exts_dir.iterdir()):
            if not ext_dir.is_dir() or ext_dir.name in disabled_ids:
                continue
            cmds_dir = ext_dir / "commands"
            if cmds_dir.is_dir():
                for f in cmds_dir.glob("*.md"):
                    if f.stem == command_name or f.stem == stem:
                        return f, ext_dir.name

    # 3. Core templates in the project
    core = project_root / ".specify" / "templates" / "commands"
    if core.is_dir():
        candidate = core / (stem + ".md")
        if candidate.exists():
            return candidate, None
    return None, None


def _script_variant(project_root):
    """Return the project's persisted script type ('sh'|'ps'|'py')."""
    default = "ps" if os.name == "nt" else "sh"
    init_opts = project_root / ".specify" / "init-options.json"
    try:
        data = json.loads(init_opts.read_text(encoding="utf-8"))
        script = data.get("script")
        if script in ("sh", "ps", "py"):
            return script
    except Exception:
        pass
    return default


def _extract_scripts(template_path):
    """Parse the scripts: block from a command template's frontmatter."""
    try:
        content = template_path.read_text(encoding="utf-8")
    except Exception:
        return {}
    m = re.match(r"^---\\n(.*?)\\n---", content, re.DOTALL)
    if not m:
        return {}
    scripts = {}
    in_scripts = False
    for line in m.group(1).splitlines():
        if line.rstrip() == "scripts:":
            in_scripts = True
            continue
        if in_scripts and line and not line[0].isspace():
            break
        if in_scripts and ":" in line:
            k, _, v = line.partition(":")
            scripts[k.strip()] = v.strip()
    return scripts


def _resolve_argv(template_path, project_root, ext_id):
    """Resolve the command's script to a runnable argv (stdlib only)."""
    scripts = _extract_scripts(template_path)
    if not scripts:
        return None
    requested = _script_variant(project_root)
    order = (requested,) if requested in scripts else ()
    fallbacks = (requested, "ps" if requested != "ps" else "sh", "py", "sh")
    seen = set()
    for cand in order + fallbacks:
        if cand in seen:
            continue
        seen.add(cand)
        if cand in scripts:
            variant = cand
            break
    else:
        return None
    script_cmd = scripts.get(variant, "").strip()
    if not script_cmd:
        return None

    base = (project_root / ".specify" / "extensions" / ext_id) if ext_id else (project_root / ".specify")
    try:
        tokens = shlex.split(script_cmd, posix=(os.name != "nt"))
    except ValueError:
        return None
    if not tokens:
        return None
    script_abs = base / tokens[0]
    if not script_abs.exists():
        return None
    rest = tokens[1:]

    if variant == "py":
        # .py files aren't directly executable; run under the dispatcher's own
        # Python (sys.executable), which is always available here.
        return [sys.executable or "python3", str(script_abs), *rest]
    if variant == "ps":
        launcher = shutil.which("pwsh") or shutil.which("powershell")
        if not launcher:
            return None
        return [launcher, "-File", str(script_abs), *rest]
    # sh: direct on POSIX; a bash/sh launcher on Windows.
    if os.name == "nt":
        launcher = shutil.which("bash") or shutil.which("sh")
        if launcher:
            return [launcher, str(script_abs), *rest]
        return None
    return [str(script_abs), *rest]


def _run_inline(command_name, payload, project_root, timeout):
    """Resolve and run the event command with stdlib only (no specify_cli)."""
    template_path, ext_id = _find_command_template(command_name, project_root)
    if not template_path:
        return 0  # command not found: fail open (no-op) for lifecycle events
    argv = _resolve_argv(template_path, project_root, ext_id)
    if not argv:
        return 0
    try:
        result = subprocess.run(
            argv,
            input=payload,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=str(project_root),
        )
        if result.stdout:
            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:
        print(f"Event {command_name} timed out", file=sys.stderr)
        return 2
    except Exception as e:
        print(f"Event {command_name} error: {e}", file=sys.stderr)
        return 2


def main():
    if len(sys.argv) < 3:
        sys.exit(0)
    command_name = sys.argv[1]
    # event_name is accepted for argv-compat with the native hook command but
    # is not needed for resolution (the command template drives everything).
    _event_name = sys.argv[2]
    # Optional 4th arg: per-handler timeout in seconds (S4).
    timeout = 120
    if len(sys.argv) >= 4:
        try:
            timeout = int(sys.argv[3])
        except (TypeError, ValueError):
            timeout = 120
    payload = sys.stdin.read() if not sys.stdin.isatty() else "{}"
    project_root = Path(__file__).parent.parent.resolve()

    # Preferred path: specify_cli is importable (durable install) — delegate to
    # the full resolver, which also handles extension manifests whose file stem
    # differs from the command name and the project's custom script selection.
    try:
        from specify_cli.events import resolve_and_run_event_command
        sys.exit(
            resolve_and_run_event_command(
                command_name, _event_name, payload, project_root, timeout=timeout
            )
        )
    except ImportError:
        pass

    # Fallback: self-contained stdlib resolver (one-time/temporary installs).
    sys.exit(_run_inline(command_name, payload, project_root, timeout))


if __name__ == "__main__":
    main()
'''

# -- TS plugin template (opencode) ----------------------------------------

_TS_PLUGIN_TEMPLATE = '''import {{ execFileSync }} from 'child_process';
import * as path from 'path';

// The dispatcher + interpreter are resolved per-project at plugin load from
// the `directory` OpenCode passes to the plugin factory (C8), not
// process.cwd() — OpenCode may be launched from a parent directory or host
// another workspace, in which case process.cwd() points at the wrong project.
let DISPATCHER = '';
let INTERPRETER = '';

function canImportSpecifyCli(py: string): boolean {{
  // R2: a project-local venv commonly lacks Spec Kit (installed globally or
  // via uv tool). Probe the interpreter can import specify_cli before
  // selecting it, so an unrelated venv doesn't shadow the PATH fallback.
  try {{
    execFileSync(py, ['-c', 'import specify_cli'], {{
      stdio: ['ignore', 'ignore', 'ignore'],
      timeout: 10000,
    }});
    return true;
  }} catch (e) {{
    return false;
  }}
}}

function resolveDispatcher(directory: string): void {{
  DISPATCHER = path.join(directory, '.specify', 'events.py');
  // Prefer a project-local venv interpreter that can import specify_cli (R2),
  // then fall back to a platform-appropriate PATH interpreter (S2: python on
  // Windows, where python3 is commonly absent; python3 on POSIX).
  const venvPy = path.join(directory, '.venv', 'bin', 'python');
  const venvWin = path.join(directory, '.venv', 'Scripts', 'python.exe');
  INTERPRETER = (
    (require('fs').existsSync(venvPy) && canImportSpecifyCli(venvPy) && venvPy) ||
    (require('fs').existsSync(venvWin) && canImportSpecifyCli(venvWin) && venvWin) ||
    (process.platform === 'win32' ? 'python' : 'python3')
  ) as string;
}}

function runEvent(command: string, event: string, input: any, output: any, timeoutSec: number): void {{
  if (!DISPATCHER) return;
  try {{
    // execFileSync with an argv array invokes the interpreter directly — no
    // shell — so command/event strings with metacharacters can't break out
    // of the dispatcher argument (C9). The dispatcher arg is seconds; the
    // execFileSync timeout is ms with a buffer so the outer cap fires after
    // the dispatcher's inner subprocess (S3).
    execFileSync(INTERPRETER, [DISPATCHER, command, event, String(timeoutSec)], {{
      input: JSON.stringify({{ input, output }}),
      stdio: ['pipe', 'inherit', 'inherit'],
      timeout: (timeoutSec + {buffer}) * 1000,
    }});
  }} catch (e) {{
    // Propagate to OpenCode's hook machinery so only this hook is rejected,
    // not the entire host process. process.exit() would kill the agent.
    throw new Error(`specify event ${{command}} (${{event}}) failed: ${{(e as Error).message}}`);
  }}
}}

{event_entries}

export default (async ({{ client, project, directory, $ }}) => {{
  resolveDispatcher(directory);
  return {{
{plugin_returns}
  }};
}});
'''


# -- Command runner logic (core) --------------------------------------------

def _find_command_template(command_name: str, project_root: Path) -> tuple[Path | None, str | None]:
    # 1. Resolve via installed extension manifests (authoritative). The
    #    registry stores per-agent ``registered_commands`` name-lists, not a
    #    ``{name, file}`` map, so the command→file mapping lives only in each
    #    extension's ``extension.yml`` ``provides.commands`` (S8). Match the
    #    command name to its declared ``file`` so commands whose file stem
    #    differs from the command name (e.g. ``speckit.selftest.extension`` →
    #    ``commands/selftest.md``) resolve correctly.
    exts_dir = project_root / ".specify" / "extensions"
    # S1: build the set of explicitly-disabled extension IDs so dispatch skips
    # disabled extensions (a stale hook would otherwise keep executing a
    # disabled extension's command). Applied to both the manifest loop and the
    # on-disk fallback below.
    disabled_ids = _disabled_extension_ids(project_root)
    try:
        from .extensions import ExtensionManager
        manager = ExtensionManager(project_root)
        for ext_id in sorted(manager.registry.keys()):
            if ext_id in disabled_ids:
                continue
            manifest = manager.get_extension(ext_id)
            if manifest is None:
                continue
            for cmd in manifest.commands:
                if not isinstance(cmd, dict):
                    continue
                if cmd.get("name") == command_name and cmd.get("file"):
                    candidate = exts_dir / ext_id / cmd["file"]
                    if candidate.exists():
                        return candidate, ext_id
    except Exception:
        # Fall through to the on-disk scan if the registry/manifests can't be
        # read; event dispatch should degrade gracefully, not crash.
        pass

    # 2. Scan extension directories by file stem (covers extensions present on
    #    disk but not resolvable via the manifest above). S1: skip disabled
    #    extensions here too so the disk fallback can't re-enable them.
    if exts_dir.is_dir():
        for ext_dir in sorted(exts_dir.iterdir()):
            if ext_dir.name in disabled_ids:
                continue
            cmds_dir = ext_dir / "commands"
            if cmds_dir.is_dir():
                for f in cmds_dir.glob("*.md"):
                    if f.stem == command_name:
                        return f, ext_dir.name

    # 3. Check core templates in the project
    core = project_root / ".specify" / "templates" / "commands"
    if core.is_dir():
        stem = command_name.replace("speckit.", "").replace("spec.", "")
        candidate = core / f"{stem}.md"
        if candidate.exists():
            return candidate, None

    # 4. Fallback to package-bundled templates via the canonical asset
    #    resolvers (wheel: core_pack/commands; source: repo-root
    #    templates/commands). The previous bespoke inspect.getfile() math
    #    pointed at core_pack/templates/commands, which never exists in a
    #    wheel build (force-include maps templates/commands -> core_pack/commands).
    from ._assets import _locate_core_pack, _repo_root
    core_pack = _locate_core_pack()
    candidate_dirs = [
        core_pack / "commands" if core_pack is not None else None,
        _repo_root() / "templates" / "commands",
    ]
    stem = command_name.replace("speckit.", "").replace("spec.", "")
    for candidate_dir in candidate_dirs:
        if candidate_dir is None or not candidate_dir.is_dir():
            continue
        candidate = candidate_dir / f"{stem}.md"
        if candidate.exists():
            return candidate, None

    return None, None


def _resolve_event_command_argv(
    template_path: Path, project_root: Path, ext_id: str | None
) -> list[str] | None:
    """Resolve a command template's ``scripts:`` entry to a runnable argv.

    ``scripts:`` values are command strings (e.g. ``scripts/bash/setup-plan.sh --json``),
    not bare paths, so joining the whole value into a ``Path`` made ``exists()``
    false and real commands silently no-op'd. This resolves the stored variant
    (honoring the project's sh/ps/py selection), splits the command string
    safely into argv, and prepends the appropriate interpreter (Python for
    ``.py``, the platform shell otherwise). Returns ``None`` if no runnable
    script is declared.
    """
    from .integrations.base import IntegrationBase

    content = template_path.read_text(encoding="utf-8")
    m = re.match(r'^---\n(.*?)\n---', content, re.DOTALL)
    if not m:
        return None
    fm = m.group(1)
    try:
        fm_data = yaml.safe_load(fm) or {}
    except Exception:
        return None
    if not isinstance(fm_data, dict):
        return None
    scripts = fm_data.get("scripts", {})
    if not isinstance(scripts, dict):
        return None
    # Determine the requested variant from the project's persisted selection,
    # falling back to the platform default — same logic MarkdownIntegration
    # uses for command scaffolding.
    requested = _load_project_script_type(project_root)
    try:
        variant = IntegrationBase.select_script_variant(requested, scripts)
    except ValueError:
        return None
    script_cmd = scripts.get(variant)
    if not isinstance(script_cmd, str) or not script_cmd.strip():
        return None

    # Base under which the script's leading path component is anchored —
    # .specify/ (core) or .specify/extensions/<id>/ (extension). All variants
    # share this anchoring so a `scripts/...` token resolves correctly (S2:
    # the py branch previously invoked build_python_invocation() on the raw
    # command string, leaving `scripts/...` anchored at the project root).
    if ext_id:
        base = project_root / ".specify" / "extensions" / ext_id
    else:
        base = project_root / ".specify"

    tokens = shlex.split(script_cmd, posix=(os.name != "nt"))
    if not tokens:
        return None
    script_abs = base / tokens[0]
    if not script_abs.exists():
        return None
    rest_args = tokens[1:]

    if variant == "py":
        # .py files aren't directly executable on Windows; prefix the resolved
        # interpreter. argv is passed to subprocess.run(shell=False), so no
        # shell quoting is needed.
        interpreter = IntegrationBase.resolve_python_interpreter(project_root)
        return [interpreter, str(script_abs), *rest_args]

    if variant == "ps":
        # PowerShell scripts cannot be executed directly by
        # subprocess.run(shell=False); invoke via `pwsh -File` (PowerShell 7+),
        # falling back to `powershell -File` (Windows PowerShell) when pwsh is
        # absent (S6). The default Windows script type would otherwise fail.
        launcher = shutil.which("pwsh") or shutil.which("powershell") or "pwsh"
        return [launcher, "-File", str(script_abs), *rest_args]

    # sh: the script is chmod'd executable during install on POSIX. On Windows
    # subprocess.run(shell=False) can't execute a .sh directly, so prefix a
    # bash/sh launcher when one is available (mirroring the ps branch's
    # pwsh -File handling, S5).
    if os.name == "nt":
        launcher = shutil.which("bash") or shutil.which("sh")
        if launcher:
            return [launcher, str(script_abs), *rest_args]
    return [str(script_abs), *rest_args]


def _load_project_script_type(project_root: Path) -> str:
    """Return the project's persisted script type ('sh'|'ps'|'py').

    Falls back to the platform default when init-options are absent or
    unreadable so event dispatch still works in a partially-initialized
    project.
    """
    default = "ps" if platform.system().lower().startswith("win") else "sh"
    try:
        from ._init_options import load_init_options
        opts = load_init_options(project_root)
        if isinstance(opts, dict):
            script = opts.get("script")
            if isinstance(script, str) and script in ("sh", "ps", "py"):
                return script
    except Exception:
        pass
    return default


def resolve_and_run_event_command(
    command_name: str,
    event_name: str,
    payload: str,
    project_root: Path,
    *,
    timeout: int = 120,
) -> int:
    """Core entry point to resolve and execute an event-driven command.

    *timeout* is the per-handler timeout in seconds, passed through from the
    native hook config via the dispatcher (S4) so a handler configured above
    the previous fixed 120s cap can run for its full duration.
    """
    template_path, ext_id = _find_command_template(command_name, project_root)
    if not template_path:
        logger.warning("Event command '%s' not found", command_name)
        return 0
    argv = _resolve_event_command_argv(template_path, project_root, ext_id)
    if not argv:
        logger.warning("No script found for event command '%s'", command_name)
        return 0
    try:
        result = subprocess.run(
            argv,
            input=payload,
            capture_output=True,
            text=True,
            timeout=timeout,
            cwd=str(project_root),
        )
        if result.stdout:
            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) -------------------------

# Resolved events map: each canonical event name maps to an *ordered list* of
# handler configs. Built-in defaults and per-extension declarations both
# contribute, so two extensions declaring ``session_start`` both run (finding
# #2) instead of the last one silently winning.
ResolvedEvents = dict[str, list[dict[str, Any]]]


def _normalize_handlers(value: Any) -> list[dict[str, Any]]:
    """Coerce a single handler config or a list of them into a validated list.

    Accepts both the legacy single-mapping shape (``{command: ...}``) and the
    explicit list shape (``[{command: ...}, ...]``). Drops any entry that is
    not a mapping or lacks a ``command`` with a warning, so a malformed user
    override never reaches installation and crashes on ``cfg.get(...)`` (#21).
    """
    if isinstance(value, dict):
        value = [value]
    if not isinstance(value, list):
        return []
    handlers: list[dict[str, Any]] = []
    for entry in value:
        if not isinstance(entry, dict):
            logger.warning("Skipping malformed event handler (expected a mapping): %r", entry)
            continue
        handlers.append(entry)
    return handlers


def _validate_resolved_event(event_name: str, handlers: list[dict[str, Any]]) -> None:
    """Validate a resolved event's handlers, raising a user-facing error.

    Raised for structural problems the user must fix (unknown event name,
    handler missing a ``command``, or ``command`` not a non-empty string per
    #17). Malformed-but-skipable entries are already dropped by
    ``_normalize_handlers``.
    """
    from .extensions import ValidationError

    if event_name not in CANONICAL_EVENTS:
        raise ValidationError(
            f"Unknown event '{event_name}': must be one of {sorted(CANONICAL_EVENTS)}"
        )
    for handler in handlers:
        command = handler.get("command")
        if not isinstance(command, str) or not command.strip():
            raise ValidationError(
                f"Event '{event_name}' handler missing required non-empty 'command' string"
            )
        # C10: matcher must be a string (or absent). A non-string matcher such
        # as `matcher: []` passes extension validation but later crashes
        # by_matcher.setdefault(matcher, ...) with TypeError: unhashable type.
        matcher = handler.get("matcher")
        if matcher is not None and not isinstance(matcher, str):
            raise ValidationError(
                f"Event '{event_name}' handler has invalid 'matcher': "
                "must be a string"
            )
        timeout = handler.get("timeout")
        if timeout is not None:
            if not isinstance(timeout, int) or isinstance(timeout, bool) or timeout <= 0:
                raise ValidationError(
                    f"Event '{event_name}' handler has invalid 'timeout': must be a positive integer"
                )


def resolve_events(
    integration_key: str,
    integration_config: dict[str, Any] | None,
    project_root: Path,
    parsed_options: dict[str, Any] | None,
) -> ResolvedEvents:
    """Resolve the final event set for an integration.

    Returns a mapping of canonical event name → ordered list of handler
    configs. Layers (lowest → highest precedence):

    1. CLI gate ``--events false`` → empty map (caller still removes prior
       native hooks; see ``install_integration_events``).
    2. Built-in defaults from ``integration_config["events"]`` (single-config
       per event, wrapped as one-element lists).
    3. Extension-declared ``events:`` — appended per extension so multiple
       extensions can declare the same event (#2).
    4. User YAML override (``.specify/integration-events.yml``) — replaces the
       accumulated set entirely when the integration key is present. Validated
       (#21) before returning; a malformed override is warned about and
       ignored rather than crashing downstream.
    """
    # Layer 1: CLI flag gate
    if parsed_options:
        events_flag = str(parsed_options.get("events", "true")).lower()
        if events_flag in ("false", "0", "no", "off"):
            return {}

    events: ResolvedEvents = {}

    # Layer 2: built-in defaults from integration config
    if integration_config and isinstance(integration_config.get("events"), dict):
        for ev, cfg in integration_config["events"].items():
            handlers = _normalize_handlers(cfg)
            if handlers:
                events.setdefault(ev, []).extend(handlers)

    # Layer 3: extension-declared events (accumulated, not overwriting)
    for ev, handlers in collect_extension_events(project_root).items():
        events.setdefault(ev, []).extend(handlers)

    # Layer 4: user YAML override (replaces entirely if key present)
    override_file = project_root / YAML_OVERRIDE_FILENAME
    if override_file.exists():
        try:
            override = yaml.safe_load(override_file.read_text(encoding="utf-8")) or {}
        except yaml.YAMLError:
            logger.warning("Could not parse %s; ignoring override", override_file)
            override = {}
        integrations = override.get("integrations", {}) if isinstance(override, dict) else {}
        if isinstance(integrations, dict) and integration_key in integrations:
            key_data = integrations[integration_key]
            if not isinstance(key_data, dict):
                # C6: a non-mapping integration entry (e.g. `claude: bad`) must
                # not be treated as a valid explicit disable. Warn and abandon
                # the override, keeping the accumulated built-in + extension
                # layers. Only an explicitly present, mapping-valued `events`
                # field replaces the prior layers.
                logger.warning(
                    "Override %s: entry for '%s' is not a mapping; ignoring override",
                    override_file, integration_key,
                )
            else:
                key_events = key_data.get("events", {})
                if not isinstance(key_events, dict):
                    logger.warning(
                        "Override %s: 'events' for '%s' is not a mapping; ignoring override",
                        override_file, integration_key,
                    )
                else:
                    # Validate every entry before adopting the override. A single
                    # invalid entry abandons the whole override and keeps the
                    # accumulated built-in + extension layers (#10): previously a
                    # typo reset resolved_override to {} and then assigned that
                    # empty map to events, silently disabling all hooks despite
                    # the "ignored" warning. Only a fully-valid override (including
                    # an explicit `events: {}`) replaces the prior layers.
                    resolved_override: ResolvedEvents = {}
                    override_valid = True
                    for ev, raw in key_events.items():
                        handlers = _normalize_handlers(raw)
                        if not handlers:
                            # C4: a malformed handler (e.g. `stop: []` or
                            # `stop: bad-value`) normalizes to no handlers.
                            # Abandon the whole override (keep prior layers)
                            # rather than skipping the entry — otherwise an
                            # override whose only entry is malformed silently
                            # disabled every built-in and extension hook. An
                            # explicit `events: {}` (no entries) remains a
                            # valid disable.
                            logger.warning(
                                "Override %s: event '%s' has no valid handler; ignoring entire override",
                                override_file, ev,
                            )
                            override_valid = False
                            break
                        try:
                            _validate_resolved_event(ev, handlers)
                        except Exception as exc:
                            logger.warning(
                                "Override %s: invalid event '%s': %s; ignoring entire override",
                                override_file, ev, exc,
                            )
                            override_valid = False
                            break
                        resolved_override[ev] = handlers
                    if override_valid:
                        events = resolved_override
                    # else: keep the accumulated built-in + extension layers.

    return events


def _disabled_extension_ids(project_root: Path) -> set[str]:
    """Return the set of explicitly-disabled extension IDs.

    Extensions not tracked in the registry are treated as enabled (backward
    compat). Used by ``collect_extension_events`` and ``_find_command_template``
    so a disabled extension's events and commands are never emitted or
    executed (S1) — otherwise a stale native hook would keep running a
    disabled extension after its config file was preserved (e.g. a JSONC
    parse failure that skipped native cleanup).
    """
    from .extensions import ExtensionRegistry

    exts_dir = project_root / ".specify" / "extensions"
    disabled_ids: set[str] = set()
    if not exts_dir.is_dir():
        return disabled_ids
    try:
        registry = ExtensionRegistry(exts_dir)
        for ext_id, meta in registry.list_by_priority(include_disabled=True):
            if not isinstance(meta, dict) or not meta.get("enabled", True):
                disabled_ids.add(ext_id)
    except Exception:
        pass
    return disabled_ids


def collect_extension_events(project_root: Path) -> ResolvedEvents:
    """Scan all installed extensions for ``events:`` declarations.

    Returns a mapping of event name → list of handler configs. Multiple
    extensions declaring the same event each contribute a handler (in
    extension-directory sort order), so callers can emit all of them (#2).

    Honors the extension registry's ``enabled`` flag (#1): an explicitly
    disabled extension's events are skipped so disabling an extension actually
    deactivates its runtime hooks. Extensions absent from the registry (e.g.
    a partially-staged install) are still included to preserve the on-disk
    scan behavior.

    Events are read from a validated ``ExtensionManifest`` (R1) rather than
    the raw ``extension.yml`` YAML, so the command-reference canonicalization
    applied during install validation (C11, e.g. ``my-ext.boot`` →
    ``speckit.my-ext.boot``) is reflected — otherwise refresh would emit the
    obsolete name and ``_find_command_template`` could not match it, leaving
    the hook silently inert.
    """
    from .extensions import ExtensionManager

    events: ResolvedEvents = {}
    exts_dir = project_root / ".specify" / "extensions"
    if not exts_dir.is_dir():
        return events

    manager = ExtensionManager(project_root)

    # Build the set of explicitly-disabled extension IDs. Extensions not
    # tracked in the registry are treated as enabled (backward compat).
    disabled_ids = _disabled_extension_ids(project_root)

    # Union of extension IDs to consider: registry-tracked IDs (validated
    # manifests, canonicalized refs) plus on-disk dirs not yet in the registry
    # (partially-staged installs). The latter fall back to the raw YAML since
    # no validated manifest is available, preserving the on-disk scan behavior.
    registry_ids = set()
    try:
        registry_ids = set(manager.registry.keys())
    except Exception:
        pass
    on_disk_ids = {
        d.name for d in exts_dir.iterdir() if d.is_dir() and (d / "extension.yml").exists()
    }
    for ext_id in sorted(registry_ids | on_disk_ids):
        if ext_id in disabled_ids:
            continue
        # Prefer the validated manifest (canonicalized command refs, R1);
        # fall back to the raw YAML for an on-disk extension not yet
        # registered (a malformed extension shouldn't abort collection).
        runtime: dict[str, Any] = {}
        if ext_id in registry_ids:
            try:
                manifest = manager.get_extension(ext_id)
            except Exception:
                manifest = None
            if manifest is not None:
                runtime = manifest.data.get("events", {}) or {}
        if not runtime:
            ext_yml = exts_dir / ext_id / "extension.yml"
            if not ext_yml.exists():
                continue
            try:
                data = yaml.safe_load(ext_yml.read_text(encoding="utf-8")) or {}
            except yaml.YAMLError:
                continue
            if not isinstance(data, dict):
                continue
            runtime = data.get("events", {}) or {}
        if not isinstance(runtime, dict):
            continue
        for event, config in runtime.items():
            handlers = _normalize_handlers(config)
            if handlers:
                events.setdefault(event, []).extend(handlers)
    return events


# -- Writing/Merging Config (Integration domain) ---------------------------

def _resolve_interpreter(project_root: Path) -> str:
    """Resolve a portable Python interpreter for native hook commands (#16).

    Delegates to ``IntegrationBase.resolve_python_interpreter`` so generated
    commands honor the project venv and never hard-code ``python3`` (which is
    commonly absent on Windows even when ``py.exe``/``python.exe`` exist).
    """
    from .integrations.base import IntegrationBase
    return IntegrationBase.resolve_python_interpreter(project_root)


def _resolve_interpreter_for_target(target_os: str) -> str:
    """Resolve a Python interpreter for a target OS, independent of the host (#S4).

    Copilot's native config carries both a ``bash`` (POSIX) and a
    ``powershell`` (Windows) variant in the same checked-in file. Resolving
    both with the *host* interpreter writes a Linux venv path into the
    PowerShell hook (or vice-versa), so the config fails on the other OS.
    Each variant instead gets a portable interpreter for its target shell;
    the dispatcher script's own ``_find_specify()`` does per-OS venv
    resolution at runtime.
    """
    if target_os == "windows":
        # Windows: ``python`` is the most portable on PATH; the py launcher
        # (``py -3``) is the recommended fallback when ``python`` is absent.
        return "python"
    # POSIX (bash): ``python3`` is universally available.
    return "python3"


def _native_timeout(integration: IntegrationBase, timeout_seconds: Any) -> int:
    """Return the timeout in the unit the integration's native config expects.

    Claude/Cursor/Codex/Copilot measure timeouts in seconds; Gemini measures
    in milliseconds (#7). An integration declares its unit via
    ``events_timeout_unit`` (``"s"`` default, ``"ms"`` for Gemini).
    """
    try:
        seconds = int(timeout_seconds)
    except (TypeError, ValueError):
        seconds = 60
    if getattr(integration, "events_timeout_unit", "s") == "ms":
        return seconds * 1000
    return seconds


def _shell_quote(value: str, target_os: str) -> str:
    """Quote *value* as one argument for the target shell (R2).

    ``host`` and ``posix`` targets use ``shlex.quote`` (POSIX shells). Safe
    tokens — ``python3``, ``speckit.ext.cmd`` — pass through bare, so a
    single-``command``-string hook (Claude/Gemini/etc.) stays invocable on
    every platform. ``windows`` targets use a PowerShell single-quoted literal
    with embedded quotes doubled, for Copilot's dedicated ``powershell`` field.

    Prevents a component containing spaces (e.g. a venv interpreter path under
    a directory with spaces) or shell metacharacters (a malformed
    extension/override ``command``) from breaking the hook or being
    interpreted by the native shell instead of passed as one dispatcher
    argument.
    """
    if target_os == "windows":
        return "'" + value.replace("'", "''") + "'"
    # "host" and "posix" both use POSIX quoting. On Windows the single-
    # command-string formats (Claude/Gemini/Qwen/Devin/Tabnine) are run via
    # Git Bash or the agent's POSIX-ish shell, so POSIX quoting is correct and
    # avoids emitting 'python' (which PowerShell wouldn't invoke without &).
    return shlex.quote(value)


def _dispatcher_command(
    integration: IntegrationBase,
    project_root: Path,
    command_name: str,
    event_name: str,
    *,
    target_os: str = "host",
    timeout_seconds: Any = None,
) -> str:
    """Build the single shell command string that invokes the dispatcher (#6).

    Claude/Gemini/Qwen/Devin/Tabnine accept one ``command`` string (not a
    ``command``+``args`` split), so each adapter renders a complete invocation:
    ``<interpreter> <dispatcher> <command> <event> [<timeout>]``. The
    interpreter is resolved portably (#16); Claude's dispatcher path is
    prefixed with ``${CLAUDE_PROJECT_DIR}/`` (Claude expands it before shell
    execution).

    ``target_os`` selects an OS-appropriate interpreter for adapters that emit
    both POSIX and Windows variants into one checked-in file (Copilot): ``host``
    uses the host-resolved interpreter (venv-aware), while ``posix``/``windows``
    emit portable interpreters so the config works on either OS (#S4).

    Each component is shell-quoted for the target shell (R2) so an interpreter
    path with spaces or a command/event containing shell metacharacters is
    passed as a single argument rather than reinterpreted by the native shell.
    The Claude dispatcher is double-quoted (``"${CLAUDE_PROJECT_DIR}/..."``) so
    the variable still expands but a project path with spaces doesn't
    word-split (C2). For the explicit ``windows`` target (Copilot's
    powershell field) the quoted interpreter is prefixed with PowerShell's
    call operator ``&`` so the quoted command is actually invoked (C1).

    When *timeout_seconds* is given, the resolved timeout (in the
    integration's native unit) is appended as a 4th argument so the dispatcher
    and inner runner honor the per-handler timeout instead of a fixed 120s cap
    that would kill a handler configured for longer (S4).
    """
    if target_os == "host":
        interpreter = _resolve_interpreter(project_root)
    else:
        interpreter = _resolve_interpreter_for_target(target_os)
    q_interp = _shell_quote(interpreter, target_os)
    q_command = _shell_quote(command_name, target_os)
    q_event = _shell_quote(event_name, target_os)
    if integration.key == "claude":
        # C2: double-quote so ${CLAUDE_PROJECT_DIR} still expands (double
        # quotes allow variable expansion in POSIX shells) but a project path
        # containing spaces doesn't word-split.
        dispatcher = '"${CLAUDE_PROJECT_DIR}/' + EVENTS_DISPATCHER_REL + '"'
    else:
        dispatcher = _shell_quote(EVENTS_DISPATCHER_REL, target_os)
    # C1: PowerShell won't invoke a single-quoted command without the call
    # operator. Prefix & for the explicit windows target only.
    prefix = "& " if target_os == "windows" else ""
    base = f"{prefix}{q_interp} {dispatcher} {q_command} {q_event}"
    if timeout_seconds is not None:
        # R2: the dispatcher interprets this argument as seconds, so pass the
        # raw seconds — NOT _native_timeout(...) (which converts to ms for
        # Gemini/Qwen/Tabnine and would yield 60000 seconds). The buffer is
        # applied to the native hook timeout field (in the adapter formatters)
        # so the agent's outer cap fires after the inner subprocess timeout.
        base += f" {_shell_quote(str(int(timeout_seconds)), target_os)}"
    return base


def install_integration_events(
    integration: IntegrationBase,
    project_root: Path,
    manifest: IntegrationManifest,
    events: ResolvedEvents,
) -> list[Path]:
    """Generate dispatcher, merge native config, return created files.

    ``events`` maps each canonical event to an ordered list of handler configs
    (#2); every handler is emitted as a separate native hook entry so two
    extensions declaring ``session_start`` both run.
    """
    canonical_to_native = getattr(integration, "CANONICAL_TO_NATIVE", {})
    if not canonical_to_native:
        return []

    # Filter to only supported events, preserving all handlers per event.
    filtered: ResolvedEvents = {}
    for ev, handlers in events.items():
        if not isinstance(handlers, list):
            continue
        if ev in canonical_to_native:
            filtered[ev] = handlers
        else:
            print(
                f"\u26a0\ufe0f  {integration.key} does not support '{ev}' events; skipping",
                file=sys.stderr,
            )

    # #3: an empty resolved map (--events false, or override disabling events)
    # must still strip prior Specify hooks from this integration's native
    # config rather than leaving them active. S3: also run the shared-
    # dispatcher refcount cleanup so an --events false upgrade of the last
    # event integration doesn't orphan .specify/events.py permanently (the
    # new manifest no longer claims it and stale cleanup excludes it).
    if not filtered:
        _remove_native_event_hooks(integration, project_root, manifest)
        _cleanup_shared_dispatcher(integration, project_root, manifest)
        return []

    created: list[Path] = []

    # 1. Generate events.py dispatcher script (#12: validate destination first)
    dispatcher_dir = project_root / EVENTS_DISPATCHER_DIR
    dispatcher_path = dispatcher_dir / EVENTS_DISPATCHER_FILENAME
    _ensure_safe_destination(dispatcher_path)
    dispatcher_dir.mkdir(parents=True, exist_ok=True)
    dispatcher_path.write_text(_EVENTS_DISPATCHER_TEMPLATE, encoding="utf-8")
    dispatcher_path.chmod(0o755)
    manifest.record_file(
        str(dispatcher_path.relative_to(project_root)),
        dispatcher_path.read_bytes(),
    )
    created.append(dispatcher_path)

    # 2. Format-specific merge/write
    fmt = getattr(integration, "events_format", "json-nested")
    config_file = getattr(integration, "events_config_file", None)
    if not config_file:
        return created

    config_path = project_root / config_file

    if fmt == "ts-plugin":
        # Opencode TS plugin custom merge
        plugin_rel = ".opencode/plugin/speckit-events.ts"
        plugin_path = project_root / plugin_rel
        _ensure_safe_destination(plugin_path)
        plugin_path.parent.mkdir(parents=True, exist_ok=True)
        plugin_path.write_text(
            _build_opencode_plugin(filtered, canonical_to_native),
            encoding="utf-8",
        )
        manifest.record_file(
            plugin_rel,
            plugin_path.read_bytes(),
        )
        created.append(plugin_path)

        # Merge plugin path into opencode.json. S5: only track the config
        # file when the merge actually wrote; a skipped merge (JSONC/malformed)
        # must not be tracked or manifest.uninstall() would later delete the
        # user's untouched file.
        if _merge_opencode_plugin_ref(config_path, f"./{plugin_rel}"):
            rel = str(config_path.relative_to(project_root))
            if rel not in manifest.files:
                manifest.record_existing(rel)
            created.append(config_path)

    elif fmt == "copilot-json":
        # Copilot dedicated .github/hooks/speckit.json. Each handler becomes
        # its own entry in the native event's list (#2). The bash and
        # powershell variants get independent OS-targeted interpreters (#S4)
        # so a config generated on Linux doesn't write a POSIX venv path into
        # the PowerShell hook (and vice-versa). Entries carry the ownership
        # marker so a pre-existing user-authored file is merged (owned entries
        # replaced) rather than overwritten (#8), and teardown removes only
        # owned entries.
        copilot_hooks: dict[str, list[dict[str, Any]]] = {}
        for ev, handlers in filtered.items():
            native = canonical_to_native[ev]
            entries: list[dict[str, Any]] = []
            for cfg in handlers:
                command = cfg.get("command", "")
                bash_cmd = _dispatcher_command(
                    integration, project_root, command, ev, target_os="posix",
                    timeout_seconds=cfg.get("timeout", 60),
                )
                ps_cmd = _dispatcher_command(
                    integration, project_root, command, ev, target_os="windows",
                    timeout_seconds=cfg.get("timeout", 60),
                )
                entries.append(
                    {
                        "type": "command",
                        "bash": bash_cmd,
                        "powershell": ps_cmd,
                        "timeoutSec": _native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER),
                        _SPECKIT_MARKER: True,
                    }
                )
            copilot_hooks[native] = entries
        # S5: only track when the merge wrote (skips on JSONC/malformed).
        if _merge_copilot_json(config_path, copilot_hooks):
            rel = str(config_path.relative_to(project_root))
            if rel not in manifest.files:
                manifest.record_existing(rel)
            created.append(config_path)

    elif fmt == "toml":
        # Codex config.toml custom merge. One [[hooks.<native>.hooks]] block
        # per handler so multiple handlers per event all emit (#2).
        lines: list[str] = []
        for ev, handlers in filtered.items():
            native = canonical_to_native[ev]
            for cfg in handlers:
                command = cfg.get("command", "")
                dispatcher_cmd = _dispatcher_command(integration, project_root, command, ev, timeout_seconds=cfg.get("timeout", 60))
                lines.append(f'[[hooks.{native}]]')
                lines.append(f'matcher = {_toml_quote(str(cfg.get("matcher", "*")))}')
                lines.append('')
                lines.append(f'[[hooks.{native}.hooks]]')
                lines.append('type = "command"')
                lines.append(f'command = {_toml_quote(dispatcher_cmd)}')
                lines.append(f'timeout = {_native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER)}')
                lines.append('speckit_marker = true')
                lines.append('')
        _merge_toml_fragment(config_path, "\n".join(lines))
        rel = str(config_path.relative_to(project_root))
        if rel not in manifest.files:
            manifest.record_existing(rel)
        created.append(config_path)

    elif fmt == "json-flat":
        # Cursor hooks.json custom merge. Flat command-string entries, one
        # per handler (#2), single resolved command string (#6/#16).
        cursor_hooks: dict[str, list[dict[str, Any]]] = {}
        for ev, handlers in filtered.items():
            native = canonical_to_native[ev]
            entries: list[dict[str, Any]] = []
            for cfg in handlers:
                command = cfg.get("command", "")
                dispatcher_cmd = _dispatcher_command(integration, project_root, command, ev, timeout_seconds=cfg.get("timeout", 60))
                entries.append(
                    {
                        "command": dispatcher_cmd,
                        "type": "command",
                        "timeout": _native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER),
                        "matcher": cfg.get("matcher", "*"),
                        _SPECKIT_MARKER: True,
                    }
                )
            cursor_hooks[native] = entries
        # #7: Cursor's .cursor/hooks.json schema requires top-level
        # "version": 1; ensure it (preserving a user's value if present).
        # S5: only track when the merge wrote (skips on JSONC/malformed).
        if _merge_json_fragment(config_path, cursor_hooks, version=1):
            rel = str(config_path.relative_to(project_root))
            if rel not in manifest.files:
                manifest.record_existing(rel)
            created.append(config_path)

    elif fmt == "json-nested":
        # Claude/Qwen/Gemini/Devin/Tabnine nested config JSON merge.
        # Native schema is a single ``command`` string per hook (not
        # command+args), so each handler renders one complete dispatcher
        # invocation (#6). Gemini timeouts are converted to ms (#7).
        # Handlers are grouped by distinct matcher so each matcher gets its
        # own matcher-group (S3); previously all handlers were placed under
        # the first handler's matcher, so two extensions registering the same
        # event with different matchers both ran for the first matcher and
        # neither for the later.
        nested_hooks: dict[str, list[dict[str, Any]]] = {}
        for ev, handlers in filtered.items():
            native = canonical_to_native[ev]
            by_matcher: dict[str, list[dict[str, Any]]] = {}
            for cfg in handlers:
                matcher = cfg.get("matcher", "*")
                command = cfg.get("command", "")
                dispatcher_cmd = _dispatcher_command(integration, project_root, command, ev, timeout_seconds=cfg.get("timeout", 60))
                by_matcher.setdefault(matcher, []).append(
                    {
                        "type": "command",
                        "command": dispatcher_cmd,
                        "timeout": _native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER),
                        _SPECKIT_MARKER: True,
                    }
                )
            nested_hooks[native] = [
                {"matcher": matcher, "hooks": inner}
                for matcher, inner in by_matcher.items()
            ]
        # S5: only track when the merge wrote (skips on JSONC/malformed).
        if _merge_json_fragment(config_path, nested_hooks):
            rel = str(config_path.relative_to(project_root))
            if rel not in manifest.files:
                manifest.record_existing(rel)
            created.append(config_path)

    elif fmt == "json-root-nested":
        # Devin hooks.v1.json: a root event map ({"PreToolUse": [...]}) with
        # no top-level "hooks" wrapper (U2). Same matcher-grouping and single
        # command string as json-nested, but written to the root.
        root_hooks: dict[str, list[dict[str, Any]]] = {}
        for ev, handlers in filtered.items():
            native = canonical_to_native[ev]
            by_matcher: dict[str, list[dict[str, Any]]] = {}
            for cfg in handlers:
                matcher = cfg.get("matcher", "*")
                command = cfg.get("command", "")
                dispatcher_cmd = _dispatcher_command(integration, project_root, command, ev, timeout_seconds=cfg.get("timeout", 60))
                by_matcher.setdefault(matcher, []).append(
                    {
                        "type": "command",
                        "command": dispatcher_cmd,
                        "timeout": _native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER),
                        _SPECKIT_MARKER: True,
                    }
                )
            root_hooks[native] = [
                {"matcher": matcher, "hooks": inner}
                for matcher, inner in by_matcher.items()
            ]
        if _merge_json_root(config_path, root_hooks):
            rel = str(config_path.relative_to(project_root))
            if rel not in manifest.files:
                manifest.record_existing(rel)
            created.append(config_path)

    return created


def _remove_native_event_hooks(
    integration: IntegrationBase,
    project_root: Path,
    manifest: IntegrationManifest,
) -> None:
    """Remove Specify-authored hooks from *this* integration's native config.

    Used both by full teardown and by the empty-resolved-map install path (#3).
    Does NOT touch the shared dispatcher (another integration may still
    reference it — #10).
    """
    fmt = getattr(integration, "events_format", None)
    config_file = getattr(integration, "events_config_file", None)
    if not config_file:
        return
    config_path = project_root / config_file
    if not config_path.exists():
        return
    _ensure_safe_destination(config_path)
    if fmt == "copilot-json":
        _remove_copilot_entries(config_path)
    elif fmt == "toml":
        _remove_toml_entries(config_path)
    elif fmt in ("json-nested", "json-flat"):
        _remove_json_entries(config_path)
    elif fmt == "json-root-nested":
        _remove_json_root_entries(config_path)
    elif fmt == "ts-plugin":
        _remove_opencode_entries(config_path)
    # Always drop this integration's manifest claim on the native config,
    # whether the file was deleted or retained with user content (S9). If we
    # kept a retained file tracked, teardown()'s manifest.uninstall(force=True)
    # would delete the entire user-owned settings file. After cleanup the file
    # is either gone or contains only user content, so this integration must
    # no longer claim it for teardown purposes.
    manifest.remove(config_file)


def _other_event_integrations_reference_dispatcher(
    project_root: Path, excluding_key: str
) -> bool:
    """Return True if another installed event-capable integration still
    references the shared ``.specify/events.py`` dispatcher (#10).

    Inspects each installed integration's manifest (excluding *excluding_key*)
    for the dispatcher path so uninstalling one multi-install event-capable
    integration doesn't delete the dispatcher the others still rely on.
    """
    from .integrations._helpers import _read_integration_json
    from .integrations.manifest import IntegrationManifest
    from .integration_state import installed_integration_keys

    state = _read_integration_json(project_root)
    for key in installed_integration_keys(state):
        if key == excluding_key:
            continue
        try:
            manifest = IntegrationManifest.load(key, project_root)
        except Exception:
            continue
        if EVENTS_DISPATCHER_REL in manifest.files:
            return True
    return False


def _cleanup_shared_dispatcher(
    integration: IntegrationBase, project_root: Path, manifest: IntegrationManifest
) -> None:
    """Drop this integration's manifest claim on the shared dispatcher and
    delete the file only when no other installed event-capable integration
    still references it (#10, S3).

    The manifest.remove() runs in both branches (S1): if we retain the file
    but leave it tracked, the subsequent manifest.uninstall() in teardown()
    sees the matching hash and deletes the file another integration still
    depends on. Used by full teardown and by the empty-resolved-map install
    path so an ``--events false`` upgrade of the last event integration
    doesn't orphan ``.specify/events.py`` permanently (S3).
    """
    dispatcher_rel = EVENTS_DISPATCHER_REL
    # Drop this integration's manifest claim if present. The remove() is
    # conditional (S1): an upgrade passes a *fresh* manifest that may never
    # have claimed the dispatcher, so the key may be absent — that's a no-op.
    if dispatcher_rel in manifest.files:
        manifest.remove(dispatcher_rel)
    # S2: run the no-other-references deletion independently of whether the
    # new manifest currently contains the key. An ``integration upgrade
    # --events false`` passes a fresh manifest that never recorded the
    # dispatcher, so gating the deletion on its presence orphans the file the
    # old on-disk manifest owned — and stale cleanup excludes it (C3). If no
    # other installed event-capable integration references the dispatcher,
    # delete it; otherwise leave it for them.
    if not _other_event_integrations_reference_dispatcher(project_root, integration.key):
        dispatcher_path = project_root / dispatcher_rel
        if dispatcher_path.exists():
            _ensure_safe_destination(dispatcher_path)
            dispatcher_path.unlink(missing_ok=True)


def remove_integration_events(
    integration: IntegrationBase, project_root: Path, manifest: IntegrationManifest
) -> None:
    """Remove Specify-authored event entries from native config.

    The shared ``.specify/events.py`` dispatcher is deleted only when no other
    installed event-capable integration still references it (#10); otherwise
    it is left in place so multi-install setups don't lose the dispatcher
    mid-stream.
    """
    _remove_native_event_hooks(integration, project_root, manifest)
    _cleanup_shared_dispatcher(integration, project_root, manifest)

    # Clean up opencode TS plugin (owned solely by the opencode integration).
    if integration.key == "opencode":
        plugin_rel = ".opencode/plugin/speckit-events.ts"
        if plugin_rel in manifest.files:
            plugin_path = project_root / plugin_rel
            if plugin_path.exists():
                _ensure_safe_destination(plugin_path)
                plugin_path.unlink(missing_ok=True)
            manifest.remove(plugin_rel)


def events_stale_exclusions(integration_key: str) -> set[str]:
    """Return project-relative paths to protect from stale cleanup."""
    from .integrations import get_integration
    integration = get_integration(integration_key)
    if not integration:
        return set()
    exclusions = set()
    config_file = getattr(integration, "events_config_file", None)
    if config_file:
        exclusions.add(config_file)
    if integration_key == "opencode":
        exclusions.add(".opencode/plugin/speckit-events.ts")
    # C3: the shared dispatcher is written into every event-capable
    # integration's manifest but is reference-counted across them. An upgrade
    # with --events false omits events.py from the new manifest, so the generic
    # stale pass would delete it without the refcount check, breaking any other
    # installed event-capable integration. Protect it here; its deletion is
    # left to remove_integration_events(), which checks the refcount.
    exclusions.add(EVENTS_DISPATCHER_REL)
    return exclusions


class EventRefreshError(RuntimeError):
    """Raised when refreshing one or more integrations' event config failed.

    Aggregates per-integration failures so a lifecycle command
    (extension add/remove/enable/disable) can surface that an extension was
    not fully deactivated — a stale native hook may still be active (R3).
    """

    def __init__(self, failures: list[tuple[str, str]]) -> None:
        self.failures = failures
        details = "; ".join(f"{key}: {detail}" for key, detail in failures)
        super().__init__(
            f"event refresh failed for {len(failures)} integration(s): {details}"
        )


def refresh_integration_events(project_root: Path) -> None:
    """Re-resolve and re-emit native event config for every installed
    event-capable integration (#1).

    Called after extension state changes (install/uninstall/enable/disable)
    so that extension-declared events are regenerated in each installed
    integration's native config — otherwise the documented install-after-
    ``specify init`` flow is inert and disabled/removed extension events stay
    active. Each integration is refreshed independently; a failure for one
    is logged and accumulated but does not abort the others. If any
    integration failed, :class:`EventRefreshError` is raised at the end so
    the lifecycle command can't claim the extension was fully deactivated
    while a stale native hook may still be active (R3).
    """
    from .integrations import get_integration
    from .integrations._helpers import _read_integration_json, _resolve_integration_options
    from .integrations.manifest import IntegrationManifest
    from .integration_state import installed_integration_keys

    state = _read_integration_json(project_root)
    failures: list[tuple[str, str]] = []
    for key in installed_integration_keys(state):
        integration = get_integration(key)
        if integration is None or not integration.supports_events():
            continue
        try:
            manifest = IntegrationManifest.load(key, project_root)
        except Exception as exc:
            logger.warning("Could not load manifest for '%s'; skipping event refresh: %s", key, exc)
            failures.append((key, f"manifest load: {exc}"))
            continue
        try:
            # C12: resolve first, then call install_integration_events once.
            # The previous flow ran _remove_native_event_hooks *before*
            # resolution, so any later failure (invalid destination, write
            # error, formatter error) destroyed the working native config
            # before the new one was written. install_integration_events
            # already removes stale Specify-marked entries and handles an
            # empty map (stripping prior hooks), so the destructive pre-step
            # is both unsafe and redundant.
            # S7: resolve this integration's persisted parsed_options so a
            # stored --events false is honored across extension lifecycle
            # changes; passing None would re-enable events the user disabled.
            _, parsed_options = _resolve_integration_options(integration, state, key, None)
            events_map = resolve_events(
                key, integration.config, project_root, parsed_options
            )
            # install_integration_events handles both the populated case
            # (writes new config, stripping stale owned entries) and the empty
            # case (strips prior hooks for --events false / disabled override).
            install_integration_events(integration, project_root, manifest, events_map)
            manifest.save()
        except Exception as exc:
            logger.warning("Failed to refresh events for '%s': %s", key, exc)
            failures.append((key, str(exc)))

    if failures:
        raise EventRefreshError(failures)


# -- Manifest validation ---------------------------------------------------

def validate_events(data: dict[str, Any]) -> None:
    """Validate ``events`` field in extension manifest data."""
    from .extensions import ValidationError

    events = data.get("events")
    if "events" in data and not isinstance(events, dict):
        raise ValidationError("Invalid events: expected a mapping")
    if events:
        for event_name, event_config in events.items():
            if not isinstance(event_config, dict):
                raise ValidationError(
                    f"Invalid event '{event_name}': expected a mapping"
                )
            command = event_config.get("command")
            # #17: command must be a non-empty string. A truthy non-string
            # (e.g. command: [foo]) would pass a bare truthiness check and
            # later render into invalid native configuration.
            if not isinstance(command, str) or not command.strip():
                raise ValidationError(
                    f"Event '{event_name}' missing required 'command' string"
                )
            if event_name not in CANONICAL_EVENTS:
                raise ValidationError(
                    f"Unknown event '{event_name}': "
                    f"must be one of {sorted(CANONICAL_EVENTS)}"
                )
            # C10: matcher must be a string (or absent). A non-string matcher
            # such as `matcher: []` would later crash by_matcher.setdefault.
            matcher = event_config.get("matcher")
            if matcher is not None and not isinstance(matcher, str):
                raise ValidationError(
                    f"Event '{event_name}' has invalid 'matcher': must be a string"
                )
            timeout = event_config.get("timeout")
            if timeout is not None:
                if not isinstance(timeout, int) or isinstance(timeout, bool) or timeout <= 0:
                    raise ValidationError(
                        f"Event '{event_name}' has invalid 'timeout': must be a positive integer"
                    )


def has_events(data: dict[str, Any]) -> bool:
    """Return True if ``events`` is present and non-empty."""
    return bool(data.get("events"))


# -- Helper merging functions ----------------------------------------------

def _toml_quote(value: str) -> str:
    """Render *value* as a TOML basic string via the shared escaper."""
    from ._toml_string import escape_toml_basic
    return escape_toml_basic(value)


def _build_opencode_plugin(
    filtered_events: ResolvedEvents,
    canonical_to_native: dict[str, str],
) -> str:
    """Render the opencode TS plugin for the resolved event set.

    Each canonical event may carry multiple handlers (#2); all handlers for a
    native event are invoked from one generated function. The dispatcher and
    interpreter are resolved per-project at plugin load from the ``directory``
    OpenCode passes (C8); the dispatcher is launched with ``execFileSync`` and
    an argv array (C9). Both the ``input`` and ``output`` callback arguments
    are forwarded to ``runEvent`` (C7) so pre_tool_use can inspect tool
    arguments and post_tool_use can inspect the result.
    """
    event_entries: list[str] = []
    plugin_returns: list[str] = []
    event_handlers: list[str] = []

    for ev, handlers in filtered_events.items():
        native = canonical_to_native[ev]
        # S1: serialize every interpolated value as a JSON string literal so a
        # quote/backslash/backtick in a command or matcher can't break the
        # generated TypeScript or inject code. json.dumps produces a valid
        # TS/JS string literal (double-quoted, fully escaped).
        ev_lit = json.dumps(ev)
        native_lit = json.dumps(native)

        # Build the body: one runEvent() call per handler wrapped in try/catch,
        # forwarding both input and output (C7). An optional tool-name matcher
        # guard applies to tool.execute.* hooks. All handlers execute before
        # any aggregate error is thrown.
        body_lines: list[str] = ["    const errors: string[] = [];"]
        for cfg in handlers:
            command = str(cfg.get("command", ""))
            command_lit = json.dumps(command)
            matcher = cfg.get("matcher", "*")
            # S3: thread the per-handler timeout (seconds) to runEvent so the
            # execFileSync cap and dispatcher arg match the configuration
            # instead of a fixed 60000ms / 120s.
            timeout_sec = int(cfg.get("timeout", 60))
            if native.startswith("tool.execute."):
                if matcher and matcher != "*":
                    tools = [t.strip().strip('"') for t in matcher.split("|")]
                    checks = " || ".join(
                        f"input.tool === {json.dumps(t.lower())}" for t in tools
                    )
                    body_lines.append(
                        f"    try {{ if ({checks}) {{ runEvent({command_lit}, {ev_lit}, input, output, {timeout_sec}); }} }} catch (e) {{ errors.push((e as Error).message); }}"
                    )
                else:
                    body_lines.append(
                        f"    try {{ runEvent({command_lit}, {ev_lit}, input, output, {timeout_sec}); }} catch (e) {{ errors.push((e as Error).message); }}"
                    )
            else:
                body_lines.append(
                    f"    try {{ runEvent({command_lit}, {ev_lit}, input, output, {timeout_sec}); }} catch (e) {{ errors.push((e as Error).message); }}"
                )
        body_lines.append("    if (errors.length > 0) { throw new Error(errors.join('; ')); }")

        if native.startswith("tool.execute."):
            ts_hook = native
            event_entries.append(
                f"function _{ev}(input: any, output: any) {{\n"
                + "\n".join(body_lines) + "\n"
                "  }"
            )
            plugin_returns.append(
                f"    {json.dumps(ts_hook)}: async (input: any, output: any) => {{\n"
                f"      _{ev}(input, output);\n"
                f"    }},"
            )
        else:
            event_entries.append(
                f"function _{ev}(input: any, output: any) {{\n"
                + "\n".join(body_lines) + "\n"
                "  }"
            )
            event_handlers.append(
                f"      if (event.type === {native_lit}) {{ _{ev}(event, event); }}"
            )

    if event_handlers:
        plugin_returns.append(
            "    event: async ({ event }) => {\n"
            + "\n".join(event_handlers) + "\n"
            "    },"
        )

    return _TS_PLUGIN_TEMPLATE.format(
        buffer=EVENT_TIMEOUT_BUFFER,
        event_entries="\n\n".join(event_entries),
        plugin_returns="\n".join(plugin_returns),
    )


def _merge_opencode_plugin_ref(config_path: Path, ref: str) -> bool:
    """Merge the speckit-events plugin ref into opencode.json.

    Aborts with a warning (#23) when the file cannot be parsed (e.g. JSONC or
    malformed JSON) instead of resetting user configuration to ``{}``. Returns
    False when skipped so callers avoid tracking the untouched file (S5).
    """
    existing = _load_user_json(config_path)
    if existing is None:
        return False
    plugins = existing.get("plugin", [])
    if not isinstance(plugins, list):
        plugins = []
    if ref not in plugins:
        plugins.append(ref)
    existing["plugin"] = plugins
    _safe_write_json(config_path, existing)
    return True


def _remove_opencode_entries(config_path: Path) -> bool:
    """Remove the speckit-events plugin ref from opencode.json (#23).

    Returns True if the file was deleted (now empty of user content), False
    otherwise. Aborts without writing when the file cannot be parsed.
    """
    _ensure_safe_destination(config_path)
    existing = _load_user_json(config_path)
    if existing is None:
        return False
    plugins = existing.get("plugin", [])
    if isinstance(plugins, list):
        ref = "./.opencode/plugin/speckit-events.ts"
        plugins = [p for p in plugins if p != ref]
        if plugins:
            existing["plugin"] = plugins
        else:
            existing.pop("plugin", None)
    if not existing:
        config_path.unlink(missing_ok=True)
        return True
    _safe_write_json(config_path, existing)
    return False


def _merge_toml_fragment(dst: Path, fragment: str) -> None:
    _ensure_safe_destination(dst)
    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) -> bool:
    """Remove Specify-marked TOML entries; delete the file if now empty (#14).

    Returns True if the file was deleted (no user content remained).
    """
    if not dst.exists():
        return False
    # R3: validate the destination before reading/writing so a symlink swap of
    # the config after install can't make teardown overwrite a file outside
    # the project (the merge/write path already validates; teardown must too).
    _ensure_safe_destination(dst)
    existing = dst.read_text(encoding="utf-8")
    cleaned = re.sub(
        r'\[\[hooks\.\w+\]\]\n(?:(?!\[\[hooks\.\w+\]\]).)*?speckit_marker = true\n*',
        "",
        existing,
        flags=re.DOTALL,
    )
    # If only whitespace/comments remain, the file had no user content —
    # delete it rather than leaving an empty stub that confuses uninstall.
    stripped = "\n".join(
        line for line in cleaned.splitlines()
        if line.strip() and not line.strip().startswith("#")
    )
    if not stripped:
        dst.unlink(missing_ok=True)
        return True
    dst.write_text(cleaned, encoding="utf-8")
    return False


def _merge_copilot_json(dst: Path, new_hooks: dict[str, list]) -> bool:
    """Merge Specify-owned hooks into Copilot's dedicated hooks JSON (#8).

    A pre-existing user-authored ``.github/hooks/speckit.json`` is merged
    (owned entries replaced via markers) rather than overwritten, and a
    parse failure aborts instead of resetting user content (#22). Returns
    False when skipped so callers avoid tracking the untouched file (S5).
    """
    existing = _load_user_json(dst)
    if existing is None:
        return False
    if not isinstance(existing, dict):
        existing = {}
    existing.setdefault("version", 1)
    existing_hooks = existing.get("hooks", {})
    if not isinstance(existing_hooks, dict):
        existing_hooks = {}
    # #11: strip ALL Specify-marked entries from every event first.
    cleaned_hooks: dict[str, list] = {}
    for event, entries in existing_hooks.items():
        if not isinstance(entries, list):
            continue
        kept_entries = _drop_marked_entries(entries)
        if kept_entries:
            cleaned_hooks[event] = kept_entries
    for event, entries in new_hooks.items():
        cleaned_hooks.setdefault(event, []).extend(entries)
    if cleaned_hooks:
        existing["hooks"] = cleaned_hooks
    else:
        existing.pop("hooks", None)
    _safe_write_json(dst, existing)
    return True


def _remove_copilot_entries(dst: Path) -> bool:
    """Remove Specify-owned hooks from Copilot's hooks JSON (#8, #14).

    Deletes the file when no user-authored hooks remain; otherwise keeps the
    file with user content. Aborts (no write) on parse failure (#22).
    """
    _ensure_safe_destination(dst)
    existing = _load_user_json(dst)
    if existing is None:
        return False
    if not isinstance(existing, dict):
        return False
    hooks = existing.get("hooks", {})
    if not isinstance(hooks, dict):
        hooks = {}
    cleaned: dict[str, list] = {}
    for event, entries in hooks.items():
        if not isinstance(entries, list):
            continue
        kept_entries = _drop_marked_entries(entries)
        if kept_entries:
            cleaned[event] = kept_entries
    if cleaned:
        existing["hooks"] = cleaned
    else:
        existing.pop("hooks", None)
    # Dedicated Spec-Kit file: delete when only the (Spec-Kit-invented)
    # ``version`` key would remain — no user content to preserve.
    user_keys = {k for k in existing if k != "version"}
    if not user_keys:
        dst.unlink(missing_ok=True)
        return True
    _safe_write_json(dst, existing)
    return False


def _merge_json_fragment(dst: Path, new_hooks: dict, *, version: int | None = None) -> bool:
    """Merge Specify-authored hook entries into a native JSON config.

    Idempotent: removes ALL prior Specify-marked entries from every event in
    the existing config first (#11), so an override that drops an event (e.g.
    ``pre_tool_use`` → ``stop``) doesn't leave stale marked entries behind.
    Marker detection recurses into nested ``hooks`` arrays (#9) so a
    matcher-group containing Specify-owned inner hooks is recognized and
    replaced rather than duplicated on every upgrade.

    Aborts with a warning (no write) when the existing file cannot be parsed
    (#22) — e.g. JSONC with comments — instead of resetting user content to
    ``{}``. Returns False when the merge was skipped so callers avoid tracking
    the untouched file (S5: otherwise manifest.uninstall() later deletes the
    user's JSONC/malformed file).

    When *version* is given, the top-level ``version`` field is ensured
    (preserving a user's value if present) so formats that require it — e.g.
    Cursor's ``.cursor/hooks.json`` schema (``version: 1``) — stay valid on a
    freshly generated file (#7).
    """
    existing = _load_user_json(dst)
    if existing is None:
        return False
    if not isinstance(existing, dict):
        existing = {}

    if version is not None:
        existing.setdefault("version", version)

    hooks_key = "hooks"
    existing_hooks = existing.get(hooks_key, {})
    if not isinstance(existing_hooks, dict):
        existing_hooks = {}

    # #11: strip ALL Specify-marked entries from every event first.
    cleaned_hooks: dict[str, list] = {}
    for event, entries in existing_hooks.items():
        if not isinstance(entries, list):
            continue
        kept_entries = _drop_marked_entries(entries)
        if kept_entries:
            cleaned_hooks[event] = kept_entries

    # Then add the newly resolved set.
    for event, entries in new_hooks.items():
        cleaned_hooks.setdefault(event, []).extend(entries)

    if cleaned_hooks:
        existing[hooks_key] = cleaned_hooks
    else:
        existing.pop(hooks_key, None)
    _safe_write_json(dst, existing)
    return True


def _merge_json_root(dst: Path, new_hooks: dict) -> bool:
    """Merge Specify-authored hooks into a root-nested JSON config (Devin U2).

    Devin's ``.devin/hooks.v1.json`` is a root event map
    (``{"PreToolUse": [...]}``) with no ``hooks`` wrapper, so the event keys
    are top-level. Same idempotent strip-all-marked-then-add semantics and
    JSONC-abort behavior as ``_merge_json_fragment``.
    """
    existing = _load_user_json(dst)
    if existing is None:
        return False
    if not isinstance(existing, dict):
        existing = {}

    # #11: strip ALL Specify-marked entries from every root event first.
    cleaned: dict[str, list] = {}
    for event, entries in existing.items():
        if not isinstance(entries, list):
            # Preserve non-list user fields at the root (Devin has none, but
            # be defensive against a mixed user file).
            cleaned[event] = entries  # type: ignore[assignment]
            continue
        kept_entries = _drop_marked_entries(entries)
        if kept_entries:
            cleaned[event] = kept_entries

    # Then add the newly resolved set (list values only).
    for event, entries in new_hooks.items():
        cleaned.setdefault(event, []).extend(entries)

    if cleaned:
        existing = cleaned
    else:
        existing = {}
    if not existing:
        dst.unlink(missing_ok=True)
        return True
    _safe_write_json(dst, existing)
    return True


def _remove_json_root_entries(dst: Path) -> bool:
    """Remove Specify-authored entries from a root-nested JSON config (Devin U2).

    Deletes the file when no user content remains (C5/#14 mirror).
    """
    _ensure_safe_destination(dst)
    existing = _load_user_json(dst)
    if existing is None:
        return False
    if not isinstance(existing, dict):
        return False
    cleaned: dict[str, Any] = {}
    for event, entries in existing.items():
        if not isinstance(entries, list):
            cleaned[event] = entries
            continue
        kept_entries = _drop_marked_entries(entries)
        if kept_entries:
            cleaned[event] = kept_entries
    if not cleaned:
        dst.unlink(missing_ok=True)
        return True
    _safe_write_json(dst, cleaned)
    return False


def _drop_marked_entries(entries: list) -> list:
    """Return *entries* with Specify-marked hooks removed, preserving user hooks.

    Handles both flat entries (marker on the entry itself) and nested entries
    (marker on inner ``hooks`` elements). A nested matcher-group whose inner
    hooks are all Specify-owned is dropped; one with surviving user inner
    hooks is kept with only the user hooks retained (#9).
    """
    kept: list = []
    for entry in entries:
        if not isinstance(entry, dict):
            kept.append(entry)
            continue
        inner = entry.get("hooks")
        if isinstance(inner, list):
            kept_inner = [h for h in inner if not _has_marker(h)]
            if kept_inner:
                entry["hooks"] = kept_inner
                kept.append(entry)
            # else: outer group was entirely Specify-owned → drop
        elif _has_marker(entry):
            pass  # flat Specify-owned entry → drop
        else:
            kept.append(entry)
    return kept


def _load_user_json(path: Path) -> dict | None:
    """Load a user-owned JSON file, aborting (None) on parse failure (#22/#23).

    Returns the parsed dict, or ``None`` when the file is missing or cannot be
    parsed (e.g. JSONC with comments, or temporarily malformed JSON). Callers
    must skip the merge rather than resetting user content to ``{}``.
    """
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, ValueError) as exc:
        logger.warning(
            "Could not parse %s (may contain JSONC comments or be malformed); "
            "skipping event-config merge to preserve user content.",
            path,
        )
        logger.debug("Parse error detail: %s", exc)
        return None
    if not isinstance(data, dict):
        logger.warning("%s is not a JSON object; skipping event-config merge.", path)
        return None
    return data


def _safe_write_json(dst: Path, data: dict) -> None:
    """Write *data* as JSON to *dst* after validating the destination (#12)."""
    _ensure_safe_destination(dst)
    dst.parent.mkdir(parents=True, exist_ok=True)
    dst.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")


def _ensure_safe_destination(dst: Path) -> None:
    """Validate a write target is a regular path inside the project (#12).

    Walks each path component and rejects symlinks (which could escape the
    project — e.g. a symlinked ``.claude`` or ``.specify`` directory pointing
    outside the repo would redirect writes to external files). Then validates
    lexical containment so ``..`` traversal is also rejected.
    """
    from .agents import CommandRegistrar

    # Walk each component so a symlinked ancestor (e.g. ``.claude`` → outside)
    # cannot be silently followed. Mirrors IntegrationManifest.record_existing.
    walked = dst.anchor and Path(dst.anchor) or Path("/")
    for part in dst.relative_to(dst.anchor).parts if dst.anchor else dst.parts:
        walked = walked / part
        if walked.is_symlink():
            raise ValueError(
                f"Refusing to write event config through a symlink: {walked}"
            )

    # Containment check against the nearest existing ancestor directory.
    base = dst.parent
    while not base.exists() and base != base.parent:
        base = base.parent
    CommandRegistrar._ensure_inside(dst, base)


def _remove_json_entries(dst: Path) -> bool:
    """Remove Specify-authored entries; delete the file if now empty (#14).

    Returns True if the file was deleted (Spec Kit created it and no user
    content remains), False otherwise.
    """
    _ensure_safe_destination(dst)
    existing = _load_user_json(dst)
    if existing is None:
        return False
    hooks = existing.get("hooks", {})
    if not isinstance(hooks, dict):
        return False
    cleaned: dict[str, list] = {}
    for event, entries in hooks.items():
        if not isinstance(entries, list):
            continue
        kept_entries = _drop_marked_entries(entries)
        if kept_entries:
            cleaned[event] = kept_entries
    if cleaned:
        existing["hooks"] = cleaned
    else:
        existing.pop("hooks", None)
    # #14/C5: if the config is now empty of user content, delete the file
    # rather than leaving a stub that confuses manifest.uninstall(). A
    # Spec-Kit-created Cursor file retains {"version": 1} after all owned
    # hooks are removed (we added the version field); treat the version-only
    # case as empty too, mirroring _remove_copilot_entries, so clean teardown
    # doesn't leave a generated stub behind.
    user_keys = {k for k in existing if k != "version"}
    if not user_keys:
        dst.unlink(missing_ok=True)
        return True
    _safe_write_json(dst, existing)
    return False


def _has_marker(entry: Any) -> bool:
    """Return True if *entry* (or any nested inner hook) is Specify-marked (#9).

    Flat entries carry the marker directly; nested matcher-groups carry it on
    their inner ``hooks`` elements, so detection recurses one level to
    recognize groups that are (wholly or partly) Specify-owned.
    """
    if not isinstance(entry, dict):
        return False
    if entry.get(_SPECKIT_MARKER, False) is True:
        return True
    inner = entry.get("hooks")
    if isinstance(inner, list):
        return any(_has_marker(h) for h in inner)
    return False