codec_hooks.py

"""CODEC Plugin Lifecycle Hooks — unified surface across all 5 execution paths.

Local Python files in ~/.codec/plugins/*.py register lifecycle handlers
(pre_tool, post_tool, on_error, on_operation_start, on_operation_end)
that fire identically from crew, voice, chat pre-LLM, chat post-LLM tag,
and MCP (stdio + HTTP) tool invocations.

Discovery: AST parse at startup mirrors codec_skill_registry — broken
plugins don't break startup; module imports are deferred to first hook
fire.

Audit: every successful hook fire emits `hook_fired`; hook-internal
exceptions emit `hook_error` (level=warning, never `error` — operation
still succeeded). pre_tool veto emits `tool_vetoed`. correlation_id
inherits from the wrapping operation per Step 1 §1.4 — never regenerated.

Trust model: hooks are local Python written or vetted by the user. No
marketplace, no auto-install, no isolation. Same as skills.

See docs/PHASE1-STEP2-DESIGN.md for the full contract.
"""
from __future__ import annotations

import ast
import hashlib
import importlib.util
import json
import logging
import os
import queue
import stat
import threading
import time
from dataclasses import dataclass, field, replace
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Union

from codec_audit import log_event as _log_event, _PREVIEW_MAX, _truncate

log = logging.getLogger("codec_hooks")

# ── Storage ────────────────────────────────────────────────────────────────────
_PLUGINS_DIR_DEFAULT = os.path.expanduser("~/.codec/plugins")
_PLUGINS_ALLOWLIST_DEFAULT = os.path.expanduser("~/.codec/plugins.allowlist")

# PR-2F (D-18): hook timeout default. Set high enough not to false-positive on
# legitimate hooks (self_improve.py's on_operation_end does a buffer snapshot +
# spawns a daemon thread for the slow Qwen call — synchronous body is <10ms).
# Operator can override via ~/.codec/config.json:plugin_hook_timeout_ms.
_HOOK_TIMEOUT_DEFAULT_S = 0.5


def _hook_timeout_seconds() -> float:
    """Read the hook timeout from config. Falls through to default on any
    error. Read lazily so config changes take effect on next hook fire
    (no restart needed for tuning)."""
    try:
        from codec_config import cfg
        ms = cfg.get("plugin_hook_timeout_ms", _HOOK_TIMEOUT_DEFAULT_S * 1000)
        if isinstance(ms, (int, float)) and ms > 0:
            return float(ms) / 1000.0
    except Exception:
        pass
    return _HOOK_TIMEOUT_DEFAULT_S

# Lifecycle hook names. AST discovery looks for top-level def's matching
# any of these. Plugins implement any subset.
_HOOK_NAMES = (
    "pre_tool",
    "post_tool",
    "on_error",
    "on_operation_start",
    "on_operation_end",
)

# Fields a pre_tool hook MAY mutate via its return dict. Anything else
# returned in the dict is dropped with a warning. Per design §6 + §11 Q2.
_MUTABLE_PRE_TOOL_FIELDS = ("task", "context")

# Identity fields a pre_tool hook MUST NOT mutate. Listed explicitly so
# the warning log message can name which one was attempted. Per §11 Q2.
_IMMUTABLE_IDENTITY_FIELDS = (
    "tool_name",
    "transport",
    "agent",
    "correlation_id",
    "client_id",
    "operation_id",
)

_DEFAULT_PRIORITY = 100

# Default identifier used when a plugin omits PLUGIN_NAME — the file stem.
_PLUGIN_FILE_SUFFIX = ".py"


# ── PR-2F (D-18): SHA-256 allowlist + AST gate ────────────────────────────────
#
# Plugins are local Python with full process privileges. Before PR-2F, any
# file dropped into ~/.codec/plugins/*.py would auto-load on next restart.
# Now a plugin's SHA-256 must be in `~/.codec/plugins.allowlist` (operator-
# managed) before `spec.loader.exec_module` runs. Mirrors PR-1A's hash-pinned
# `skills/.manifest.json` chokepoint: hash-pinned trust IS the security
# decision, not the AST check.
#
# AST check (via `codec_config.is_dangerous_skill_code`) is still run when
# the file is NOT in the allowlist — but only to enrich the `plugin_load_blocked`
# audit event with a specific reason ("ast_dangerous: subprocess.run" vs
# "not_in_allowlist"). Either reason → refuse. Both checks must pass for a
# plugin to load — hash match OR (well, the hash match alone is sufficient;
# AST is only run when hash misses, for forensic clarity).
#
# Migration: existing plugins at `~/.codec/plugins/*.py` on first run after
# PR-2F are grandfathered — their current hashes are written to the
# allowlist with `approved_by: "initial_migration"` and an `info`-level
# audit event. This avoids surprising the operator on upgrade.

_ALLOWLIST_LOCK = threading.Lock()


def _default_allowlist_path_for(plugins_dir: str) -> str:
    """Derive the allowlist path from the plugins dir. Production:
    `~/.codec/plugins/` → `~/.codec/plugins.allowlist`. Tests pointing
    at a tmp plugins dir get a sibling allowlist in the same tmp tree,
    so they don't touch the operator's real allowlist file."""
    parent = os.path.dirname(os.path.abspath(plugins_dir.rstrip(os.sep)))
    return os.path.join(parent, "plugins.allowlist")


def _allowlist_path_for(plugins_dir: str) -> Path:
    """Resolved allowlist path for a given plugins dir."""
    return Path(_default_allowlist_path_for(plugins_dir))


def _read_allowlist(path: Optional[Path] = None) -> Dict[str, Dict[str, Any]]:
    """Load the allowlist as a dict keyed by plugin filename. Returns {} on
    any error (missing file, parse error, wrong shape) — fail-closed: no
    file = no allowed plugins."""
    p = path if path is not None else Path(_PLUGINS_ALLOWLIST_DEFAULT)
    if not p.exists():
        return {}
    try:
        raw = p.read_text(encoding="utf-8")
        obj = json.loads(raw)
        if not isinstance(obj, dict):
            log.warning("[plugins] allowlist root is not a dict; treating as empty")
            return {}
        # Validate shape — each entry must have a sha256 hex string.
        clean: Dict[str, Dict[str, Any]] = {}
        for fname, entry in obj.items():
            if not isinstance(entry, dict):
                continue
            h = entry.get("sha256")
            if isinstance(h, str) and len(h) == 64 and all(c in "0123456789abcdef" for c in h.lower()):
                clean[fname] = entry
        return clean
    except (OSError, json.JSONDecodeError) as e:
        log.warning("[plugins] allowlist read failed: %s", e)
        return {}


def _write_allowlist(allowlist: Dict[str, Dict[str, Any]],
                     path: Optional[Path] = None) -> bool:
    """Atomic-write the allowlist with 0600 perms. Returns True on success."""
    p = path if path is not None else Path(_PLUGINS_ALLOWLIST_DEFAULT)
    try:
        p.parent.mkdir(parents=True, exist_ok=True)
        tmp = p.with_suffix(p.suffix + ".tmp")
        tmp.write_text(json.dumps(allowlist, indent=2, sort_keys=True),
                       encoding="utf-8")
        os.chmod(tmp, stat.S_IRUSR | stat.S_IWUSR)
        os.replace(tmp, p)
        try:
            os.chmod(p, stat.S_IRUSR | stat.S_IWUSR)
        except OSError:
            pass
        return True
    except OSError as e:
        log.warning("[plugins] allowlist write failed: %s", e)
        return False


def _file_sha256(path: str) -> Optional[str]:
    """SHA-256 hex digest of file contents. None on read error."""
    try:
        with open(path, "rb") as f:
            return hashlib.sha256(f.read()).hexdigest()
    except OSError:
        return None


def _maybe_grandfather_existing_plugins(plugins_dir: str,
                                          allowlist_path: Optional[Path] = None) -> None:
    """One-shot migration: if no allowlist file exists yet AND the plugins
    dir has .py files, write their current hashes to the allowlist with
    `approved_by: "initial_migration"`. Idempotent — runs once at the
    upgrade boundary, then becomes a no-op.

    `allowlist_path` defaults to the sibling of `plugins_dir` (i.e.
    `<dirname(plugins_dir)>/plugins.allowlist`) so tests pointing at a
    tmp plugins dir get a tmp allowlist, not the real one."""
    if allowlist_path is None:
        allowlist_path = _allowlist_path_for(plugins_dir)
    if allowlist_path.exists():
        return
    if not os.path.isdir(plugins_dir):
        return
    pys = [f for f in os.listdir(plugins_dir)
           if f.endswith(_PLUGIN_FILE_SUFFIX) and not f.startswith("_")]
    if not pys:
        # No plugins to grandfather — still create an empty allowlist so
        # subsequent loads don't re-attempt migration on every restart.
        _write_allowlist({}, allowlist_path)
        return
    seed: Dict[str, Dict[str, Any]] = {}
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    for fname in pys:
        fpath = os.path.join(plugins_dir, fname)
        h = _file_sha256(fpath)
        if h is None:
            continue
        seed[fname] = {
            "sha256": h,
            "approved_at": now,
            "approved_by": "initial_migration",
        }
    if _write_allowlist(seed, allowlist_path):
        log.info("[plugins] grandfathered %d existing plugin(s) into allowlist", len(seed))
        try:
            _log_event(
                "plugin_allowlist_migrated", "codec-hooks",
                f"grandfathered {len(seed)} plugin(s)",
                extra={"plugin_count": len(seed), "filenames": sorted(seed.keys())},
                level="info", outcome="ok",
            )
        except Exception:
            pass


def _is_plugin_allowed(filepath: str,
                       allowlist_path: Optional[Path] = None) -> tuple[bool, str]:
    """Return (allowed, reason). `allowed` True only if the file's current
    SHA-256 matches an entry in the allowlist keyed by basename. Tamper
    detection: a previously-approved plugin whose content changed will
    have a hash mismatch and be refused until re-approved."""
    if allowlist_path is None:
        allowlist_path = _allowlist_path_for(os.path.dirname(filepath))
    fname = os.path.basename(filepath)
    h = _file_sha256(filepath)
    if h is None:
        return False, "file_unreadable"
    allowlist = _read_allowlist(allowlist_path)
    entry = allowlist.get(fname)
    if entry is None:
        return False, "not_in_allowlist"
    if entry.get("sha256") != h:
        return False, "hash_mismatch"
    return True, ""


def _emit_plugin_load_blocked(plugin_name: str, filepath: str,
                              reason: str, extra_detail: str = "") -> None:
    """Audit emit for plugin load refusal. Fire-and-forget."""
    extra: Dict[str, Any] = {
        "plugin_name": plugin_name,
        "plugin_path": filepath,
        "reason": reason,
    }
    if extra_detail:
        extra["detail"] = extra_detail[:200]
    try:
        _log_event(
            "plugin_load_blocked", "codec-hooks",
            f"Plugin {plugin_name!r} refused: {reason}",
            extra=extra,
            level="warning", outcome="error",
        )
    except Exception:
        pass


# ── PR-2F (D-18): thread-with-timeout for hook execution ───────────────────────
#
# Before PR-2F, hook functions ran in the calling thread. A slow / malicious
# `pre_tool` could block the entire chat / voice / crew turn. Now every hook
# fires inside a daemon thread with a hard timeout (default 500ms, configurable
# via `~/.codec/config.json:plugin_hook_timeout_ms`). On timeout, the calling
# thread receives a sentinel and the audit log records `plugin_hook_timeout`.
# The daemon thread keeps running in the background; daemon=True means it
# doesn't block process shutdown.


class _HookTimedOut:
    """Sentinel returned by `_run_hook_with_timeout` when the timeout
    elapsed. Distinct from None (which means "no return"); callers check
    `isinstance(result, _HookTimedOut)` to decide whether to log timeout."""
    __slots__ = ("hook_name", "plugin_name", "timeout_s")

    def __init__(self, *, hook_name: str, plugin_name: str, timeout_s: float):
        self.hook_name = hook_name
        self.plugin_name = plugin_name
        self.timeout_s = timeout_s


def _run_hook_with_timeout(
    fn: Callable,
    args: tuple,
    *,
    hook_name: str,
    plugin_name: str,
    timeout_s: Optional[float] = None,
) -> tuple[Any, Optional[BaseException]]:
    """Run `fn(*args)` in a daemon thread with a hard timeout. Returns
    `(result, exception)`:
      - on success → `(retval, None)`
      - on hook exception → `(None, exception)` (caller emits hook_error)
      - on timeout → `(_HookTimedOut(...), None)` (caller emits plugin_hook_timeout)
    Never raises. The daemon thread keeps running on timeout; daemon=True
    so process shutdown isn't blocked.
    """
    if timeout_s is None:
        timeout_s = _hook_timeout_seconds()
    result_q: "queue.Queue" = queue.Queue(maxsize=1)
    exc_q: "queue.Queue" = queue.Queue(maxsize=1)

    def _runner():
        try:
            result_q.put(fn(*args))
        except BaseException as e:
            try:
                exc_q.put(e)
            except Exception:
                pass

    t = threading.Thread(target=_runner, daemon=True,
                          name=f"codec-hook:{plugin_name}.{hook_name}")
    t.start()
    t.join(timeout=timeout_s)

    if t.is_alive():
        # Abandon — thread keeps running but caller doesn't wait.
        return _HookTimedOut(
            hook_name=hook_name, plugin_name=plugin_name,
            timeout_s=timeout_s,
        ), None
    if not exc_q.empty():
        return None, exc_q.get_nowait()
    if not result_q.empty():
        return result_q.get_nowait(), None
    return None, None


def _emit_plugin_hook_timeout(plugin_name: str, hook_name: str,
                              tool_name: Optional[str], transport: str,
                              correlation_id: str, timeout_s: float) -> None:
    """Audit emit for hook timeout. level=warning per Step 2 §7.5 contract
    (operational signal, not operation failure)."""
    try:
        _log_event(
            "plugin_hook_timeout", "codec-hooks",
            f"plugin {plugin_name}.{hook_name} exceeded {timeout_s * 1000:.0f}ms",
            extra={
                "plugin_name": plugin_name,
                "hook_name": hook_name,
                "tool_name": tool_name,
                "timeout_ms": int(timeout_s * 1000),
            },
            level="warning", outcome="error",
            transport=transport, tool=tool_name or "",
            correlation_id=correlation_id,
        )
    except Exception as e:
        log.debug("plugin_hook_timeout emit failed: %s", e)


# ── HookCtx + HookVeto ─────────────────────────────────────────────────────────
@dataclass(frozen=True)
class HookCtx:
    """Read-only context passed to every hook. Frozen — mutation via return only.

    Per design §1.4. The fields populated depend on which hook is firing:
        pre_tool / post_tool / on_error: tool_name, task, context, agent, client_id
        on_operation_start / on_operation_end: operation_id; on_operation_end
            also sets duration_ms + outcome
    `transport`, `correlation_id`, `plugin_name`, `timestamp_utc` are always set.
    """
    transport: str
    correlation_id: str
    plugin_name: str
    timestamp_utc: str

    tool_name: Optional[str] = None
    task: Optional[str] = None
    context: Optional[str] = None
    agent: Optional[str] = None
    client_id: Optional[str] = None

    operation_id: Optional[str] = None
    duration_ms: Optional[float] = None
    outcome: Optional[str] = None


class HookVeto:
    """Sentinel returned by pre_tool to abort a tool invocation. Not raised."""
    __slots__ = ("reason", "plugin_name")

    def __init__(self, reason: str, *, plugin_name: Optional[str] = None):
        self.reason = (reason or "")[:200]
        self.plugin_name = plugin_name

    def __repr__(self) -> str:  # pragma: no cover — debug aid only
        return f"HookVeto(plugin={self.plugin_name!r}, reason={self.reason!r})"


# ── Plugin metadata + registry ─────────────────────────────────────────────────
@dataclass
class _PluginMeta:
    name: str
    description: str
    priority: int
    tool_filter: Optional[List[str]]   # exact tool names; None = all tools
    file_path: str
    # Hook names declared at module top-level via AST. Functions are loaded
    # lazily on first hook fire — module not imported until needed.
    declared_hooks: List[str] = field(default_factory=list)

    def applies_to(self, tool_name: Optional[str]) -> bool:
        """Does this plugin apply to the given tool? None tool_name = always (operation hook)."""
        if self.tool_filter is None:
            return True
        if tool_name is None:
            return True   # operation hooks fire regardless of tool_name
        return tool_name in self.tool_filter


def _extract_metadata(filepath: str) -> Optional[_PluginMeta]:
    """AST-parse a plugin file; return _PluginMeta or None.

    Mirrors codec_skill_registry._extract_metadata. Never executes the
    module. Looks for top-level constants (PLUGIN_*) and top-level def's
    matching the lifecycle hook names.
    """
    try:
        with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
            source = f.read()
        tree = ast.parse(source, filename=filepath)
    except (SyntaxError, OSError) as e:
        log.warning("Plugin metadata parse error (%s): %s", filepath, e)
        return None

    name: Optional[str] = None
    description = ""
    priority = _DEFAULT_PRIORITY
    tool_filter: Optional[List[str]] = None
    declared_hooks: List[str] = []

    for node in ast.iter_child_nodes(tree):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if not isinstance(target, ast.Name):
                    continue
                tid = target.id
                if tid == "PLUGIN_NAME":
                    try:
                        name = ast.literal_eval(node.value)
                    except (ValueError, TypeError):
                        pass
                elif tid == "PLUGIN_DESCRIPTION":
                    try:
                        description = ast.literal_eval(node.value) or ""
                    except (ValueError, TypeError):
                        pass
                elif tid == "PLUGIN_PRIORITY":
                    try:
                        v = ast.literal_eval(node.value)
                        if isinstance(v, int):
                            priority = v
                    except (ValueError, TypeError):
                        pass
                elif tid == "PLUGIN_TOOL_FILTER":
                    try:
                        v = ast.literal_eval(node.value)
                        if v is None:
                            tool_filter = None
                        elif isinstance(v, (list, tuple)):
                            tool_filter = [str(x) for x in v]
                    except (ValueError, TypeError):
                        pass
        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if node.name in _HOOK_NAMES:
                declared_hooks.append(node.name)

    if not declared_hooks:
        # No lifecycle functions → not a plugin. Skip silently.
        return None

    if name is None:
        # Default: filename stem (matches skill convention).
        name = os.path.basename(filepath)
        if name.endswith(_PLUGIN_FILE_SUFFIX):
            name = name[:-len(_PLUGIN_FILE_SUFFIX)]

    return _PluginMeta(
        name=name,
        description=description,
        priority=priority,
        tool_filter=tool_filter,
        file_path=filepath,
        declared_hooks=declared_hooks,
    )


class PluginRegistry:
    """AST-discover plugins; lazy-load modules on first hook fire.

    Exposed for tests; production uses the module-level _registry instance.
    """

    def __init__(self, plugins_dir: str,
                 allowlist_path: Optional[str] = None):
        self.plugins_dir = plugins_dir
        # PR-2F: each registry owns its allowlist path. Default derives from
        # plugins_dir's parent, so tests with tmp plugins dirs get a tmp
        # allowlist (no pollution of the operator's real ~/.codec/plugins.allowlist).
        self.allowlist_path = Path(allowlist_path or
                                     _default_allowlist_path_for(plugins_dir))
        # Sort key: (priority asc, filename asc) per §5.1
        self._plugins: List[_PluginMeta] = []
        # name → loaded module (populated on first fire)
        self._modules: Dict[str, Any] = {}
        # Plugin names that failed to import — skipped on every fire.
        self._broken: set[str] = set()
        self._lock = threading.Lock()

    def scan(self) -> int:
        """AST-parse every plugin file; cache metadata. Cheap — no imports.

        PR-2F (D-18): runs the one-shot grandfather migration on first
        encounter (no allowlist file + plugins dir has .py files →
        seed allowlist with current hashes). Idempotent — once the
        allowlist file exists, migration is a no-op."""
        _maybe_grandfather_existing_plugins(self.plugins_dir, self.allowlist_path)
        plugins: List[_PluginMeta] = []
        if os.path.isdir(self.plugins_dir):
            for fname in sorted(os.listdir(self.plugins_dir)):
                if not fname.endswith(_PLUGIN_FILE_SUFFIX):
                    continue
                if fname.startswith("_"):
                    continue
                fpath = os.path.join(self.plugins_dir, fname)
                meta = _extract_metadata(fpath)
                if meta is not None:
                    plugins.append(meta)
        # Sort by (priority, filename) — lower priority runs first.
        plugins.sort(key=lambda m: (m.priority, os.path.basename(m.file_path)))
        with self._lock:
            self._plugins = plugins
        log.info("Plugin registry: %d plugins discovered (metadata only)", len(plugins))
        return len(plugins)

    def all(self) -> List[_PluginMeta]:
        with self._lock:
            return list(self._plugins)

    def for_hook(self, hook_name: str, tool_name: Optional[str] = None) -> List[_PluginMeta]:
        """Plugins that declared `hook_name` AND apply to `tool_name`."""
        with self._lock:
            plugins = list(self._plugins)
        return [p for p in plugins
                if hook_name in p.declared_hooks and p.applies_to(tool_name)
                and p.name not in self._broken]

    def get_fn(self, plugin: _PluginMeta, hook_name: str) -> Optional[Callable]:
        """Lazy-load the plugin module and return the named hook function.

        PR-2F (D-18): a two-stage gate runs BEFORE `exec_module`:
          1. SHA-256 hash of the file must match an entry in
             `~/.codec/plugins.allowlist` keyed by basename.
          2. If the hash doesn't match (not in allowlist OR content changed),
             also run `is_dangerous_skill_code` for forensic clarity in the
             refusal audit event. Either way, refuse the load.

        On refusal the plugin is marked broken and `plugin_load_blocked`
        is emitted with the specific reason. Returns None.

        On success the module is cached for the process lifetime.
        """
        if plugin.name in self._broken:
            return None
        mod = self._modules.get(plugin.name)
        if mod is None:
            # ── Two-stage trust gate ──
            allowed, reason = _is_plugin_allowed(
                plugin.file_path, self.allowlist_path)
            if not allowed:
                # Run AST check too, so the audit event captures the SPECIFIC
                # dangerous pattern if there is one — helps the operator
                # decide whether to approve the file.
                detail = ""
                try:
                    with open(plugin.file_path, "r", encoding="utf-8",
                              errors="ignore") as f:
                        src = f.read()
                    from codec_config import is_dangerous_skill_code
                    dangerous, ast_reason = is_dangerous_skill_code(src)
                    if dangerous:
                        detail = f"ast_dangerous: {ast_reason}"
                except Exception:
                    pass
                _emit_plugin_load_blocked(
                    plugin.name, plugin.file_path, reason, detail,
                )
                self._broken.add(plugin.name)
                return None

            try:
                import sys
                module_name = f"codec_plugin_{plugin.name}"
                spec = importlib.util.spec_from_file_location(
                    module_name, plugin.file_path)
                if spec is None or spec.loader is None:
                    raise ImportError(f"could not build spec for {plugin.file_path}")
                mod = importlib.util.module_from_spec(spec)
                # Register in sys.modules BEFORE exec_module so the plugin can
                # import itself transitively without re-loading. Standard
                # importlib pattern; mirrors codec_skill_registry's caching.
                sys.modules[module_name] = mod
                try:
                    spec.loader.exec_module(mod)
                except BaseException:
                    sys.modules.pop(module_name, None)  # roll back on failure
                    raise
                self._modules[plugin.name] = mod
                log.info("Lazy-loaded plugin: %s (sha256 allowlist-approved)",
                         plugin.name)
            except Exception as e:
                log.warning("Plugin import error (%s): %s", plugin.name, e)
                self._broken.add(plugin.name)
                return None
        fn = getattr(mod, hook_name, None)
        if not callable(fn):
            return None
        return fn


# Module-level registry. Tests can monkeypatch _registry to point at a temp dir.
_registry: PluginRegistry = PluginRegistry(_PLUGINS_DIR_DEFAULT)
_registry.scan()


# ── Audit emit helper ──────────────────────────────────────────────────────────
def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds")


def _emit_hook_fired(*, plugin_name: str, hook_name: str,
                     tool_name: Optional[str], transport: str,
                     correlation_id: str, duration_ms: float,
                     mutated: bool, vetoed: bool) -> None:
    """Fire-and-forget hook_fired emit. Never raises."""
    extra = {
        "plugin_name": plugin_name,
        "hook_name": hook_name,
        "tool_name": tool_name,
        "mutated": bool(mutated),
        "vetoed": bool(vetoed),
    }
    try:
        _log_event(
            "hook_fired", "codec-hooks",
            extra=extra,
            outcome="ok",
            level="info",
            transport=transport,
            duration_ms=duration_ms,
            correlation_id=correlation_id,
        )
    except Exception as e:
        log.debug("hook_fired emit failed: %s", e)


def _emit_hook_error(*, plugin_name: str, hook_name: str,
                     tool_name: Optional[str], transport: str,
                     correlation_id: str, duration_ms: float,
                     exc: BaseException) -> None:
    """Per design §7.5. level='warning' (operation still succeeded)."""
    err_type = type(exc).__name__
    err_msg = _truncate(str(exc), _PREVIEW_MAX)
    try:
        _log_event(
            "hook_error", "codec-hooks",
            f"plugin {plugin_name}.{hook_name} raised {err_type}",
            extra={
                "plugin_name": plugin_name,
                "hook_name": hook_name,
                "tool_name": tool_name,
            },
            outcome="error",
            level="warning",       # NOT "error" — operation still succeeded
            transport=transport,
            duration_ms=duration_ms,
            error_type=err_type,
            error=err_msg,
            tool=tool_name or "",
            correlation_id=correlation_id,
        )
    except Exception as e:
        log.debug("hook_error emit failed: %s", e)


def _emit_tool_vetoed(*, tool_name: str, transport: str, correlation_id: str,
                      veto: HookVeto, duration_ms: float,
                      task_preview: Optional[str] = None) -> None:
    """Per design §4.3."""
    extra = {
        "veto_reason": (veto.reason or "")[:_PREVIEW_MAX],
        "plugin_name": veto.plugin_name,
    }
    if task_preview is not None:
        extra["task_preview"] = _truncate(task_preview, _PREVIEW_MAX)
    try:
        _log_event(
            "tool_vetoed", "codec-hooks",
            f"{tool_name} vetoed by {veto.plugin_name}",
            extra=extra,
            outcome="denied",
            level="warning",
            transport=transport,
            tool=tool_name,
            duration_ms=duration_ms,
            correlation_id=correlation_id,
        )
    except Exception as e:
        log.debug("tool_vetoed emit failed: %s", e)


def _fire_one_pre_tool(plugin: _PluginMeta, ctx: HookCtx) -> Any:
    """Run one plugin's pre_tool. Returns None / dict / HookVeto / sentinel-skip.

    PR-2F (D-18): wrapped in daemon-thread timeout. On timeout, emits
    `plugin_hook_timeout` audit and returns None (operation continues
    with unmutated state — never block the calling thread on a slow plugin).
    """
    fn = _registry.get_fn(plugin, "pre_tool")
    if fn is None:
        return None
    plugin_ctx = replace(ctx, plugin_name=plugin.name)
    t0 = time.monotonic()
    ret, exc = _run_hook_with_timeout(
        fn, (plugin_ctx,),
        hook_name="pre_tool", plugin_name=plugin.name,
    )
    elapsed = (time.monotonic() - t0) * 1000.0
    if isinstance(ret, _HookTimedOut):
        _emit_plugin_hook_timeout(
            plugin_name=plugin.name, hook_name="pre_tool",
            tool_name=ctx.tool_name, transport=ctx.transport,
            correlation_id=ctx.correlation_id, timeout_s=ret.timeout_s,
        )
        return None
    if exc is not None:
        _emit_hook_error(
            plugin_name=plugin.name, hook_name="pre_tool",
            tool_name=ctx.tool_name, transport=ctx.transport,
            correlation_id=ctx.correlation_id, duration_ms=elapsed, exc=exc,
        )
        return None
    mutated = isinstance(ret, dict) and bool(ret)
    vetoed = isinstance(ret, HookVeto)
    if vetoed and ret.plugin_name is None:
        # Stamp the plugin name so callers know who vetoed.
        ret.plugin_name = plugin.name
    _emit_hook_fired(
        plugin_name=plugin.name, hook_name="pre_tool",
        tool_name=ctx.tool_name, transport=ctx.transport,
        correlation_id=ctx.correlation_id, duration_ms=elapsed,
        mutated=mutated, vetoed=vetoed,
    )
    return ret


def _fire_one_post_tool(plugin: _PluginMeta, ctx: HookCtx, result: str) -> Any:
    """Run one plugin's post_tool. Wrapped in daemon-thread timeout (PR-2F)."""
    fn = _registry.get_fn(plugin, "post_tool")
    if fn is None:
        return None
    plugin_ctx = replace(ctx, plugin_name=plugin.name)
    t0 = time.monotonic()
    ret, exc = _run_hook_with_timeout(
        fn, (plugin_ctx, result),
        hook_name="post_tool", plugin_name=plugin.name,
    )
    elapsed = (time.monotonic() - t0) * 1000.0
    if isinstance(ret, _HookTimedOut):
        _emit_plugin_hook_timeout(
            plugin_name=plugin.name, hook_name="post_tool",
            tool_name=ctx.tool_name, transport=ctx.transport,
            correlation_id=ctx.correlation_id, timeout_s=ret.timeout_s,
        )
        return None
    if exc is not None:
        _emit_hook_error(
            plugin_name=plugin.name, hook_name="post_tool",
            tool_name=ctx.tool_name, transport=ctx.transport,
            correlation_id=ctx.correlation_id, duration_ms=elapsed, exc=exc,
        )
        return None
    mutated = isinstance(ret, str)
    _emit_hook_fired(
        plugin_name=plugin.name, hook_name="post_tool",
        tool_name=ctx.tool_name, transport=ctx.transport,
        correlation_id=ctx.correlation_id, duration_ms=elapsed,
        mutated=mutated, vetoed=False,
    )
    return ret


def _fire_one_observe(plugin: _PluginMeta, hook_name: str, ctx: HookCtx,
                      *extra_args: Any) -> None:
    """Run one observe-only hook (on_error / on_operation_*). Return ignored.
    Wrapped in daemon-thread timeout (PR-2F)."""
    fn = _registry.get_fn(plugin, hook_name)
    if fn is None:
        return
    plugin_ctx = replace(ctx, plugin_name=plugin.name)
    t0 = time.monotonic()
    ret, exc = _run_hook_with_timeout(
        fn, (plugin_ctx, *extra_args),
        hook_name=hook_name, plugin_name=plugin.name,
    )
    elapsed = (time.monotonic() - t0) * 1000.0
    if isinstance(ret, _HookTimedOut):
        _emit_plugin_hook_timeout(
            plugin_name=plugin.name, hook_name=hook_name,
            tool_name=ctx.tool_name, transport=ctx.transport,
            correlation_id=ctx.correlation_id, timeout_s=ret.timeout_s,
        )
        return
    if exc is not None:
        _emit_hook_error(
            plugin_name=plugin.name, hook_name=hook_name,
            tool_name=ctx.tool_name, transport=ctx.transport,
            correlation_id=ctx.correlation_id, duration_ms=elapsed, exc=exc,
        )
        return
    _emit_hook_fired(
        plugin_name=plugin.name, hook_name=hook_name,
        tool_name=ctx.tool_name, transport=ctx.transport,
        correlation_id=ctx.correlation_id, duration_ms=elapsed,
        mutated=False, vetoed=False,
    )


# ── Mutation-contract enforcement ──────────────────────────────────────────────
def _apply_pre_tool_mutation(plugin_name: str, ret: Any,
                             task: str, context: str) -> tuple[str, str]:
    """Validate + apply a pre_tool dict return. Drops immutable-identity
    fields with a warning. Per design §6 + §11 Q2.
    """
    if not isinstance(ret, dict):
        # Anything else (False, [], a frozen ctx) is a typo — log + ignore.
        if ret is not None:
            log.warning("[hooks] plugin %s pre_tool returned %s; ignored",
                        plugin_name, type(ret).__name__)
        return task, context
    new_task = task
    new_context = context
    for k, v in ret.items():
        if k in _IMMUTABLE_IDENTITY_FIELDS:
            log.warning("[hooks] plugin %s tried to mutate immutable field %r; "
                        "ignored", plugin_name, k)
            continue
        if k == "task":
            if isinstance(v, str):
                new_task = v
            else:
                log.warning("[hooks] plugin %s pre_tool returned non-str task; "
                            "ignored", plugin_name)
        elif k == "context":
            if isinstance(v, str):
                new_context = v
            else:
                log.warning("[hooks] plugin %s pre_tool returned non-str "
                            "context; ignored", plugin_name)
        else:
            # Unknown key — just warn; not necessarily harmful but signals a typo.
            log.warning("[hooks] plugin %s pre_tool returned unknown key %r; "
                        "ignored", plugin_name, k)
    return new_task, new_context


# ── Public emitters ────────────────────────────────────────────────────────────
def run_with_hooks(
    *,
    tool_name: str,
    task: str,
    context: str = "",
    transport: str,
    agent: Optional[str] = None,
    client_id: Optional[str] = None,
    correlation_id: str,
    invoke: Callable[[str, str], str],
) -> Union[str, HookVeto]:
    """Orchestrate pre/post/on_error hooks around invoke(task, context).

    Per design §3.1. Never raises in the hook layer. If invoke raises,
    on_error fires and the exception is re-raised so the call site's
    existing audit emit + error-formatting paths are unchanged.

    Returns the post-hook result string, or a HookVeto sentinel if any
    pre_tool returned one. The first veto wins; subsequent pre_tool
    hooks in the chain do not fire after a veto.
    """
    ctx = HookCtx(
        transport=transport,
        correlation_id=correlation_id,
        plugin_name="",                # set per-fire by _fire_one_*
        timestamp_utc=_now_iso(),
        tool_name=tool_name,
        task=task,
        context=context,
        agent=agent,
        client_id=client_id,
    )

    # 1. pre_tool chain
    plugins_pre = _registry.for_hook("pre_tool", tool_name)
    cur_task = task
    cur_context = context
    veto_t0 = time.monotonic()
    for p in plugins_pre:
        # Each fire sees the current (possibly mutated) task/context
        ctx_now = replace(ctx, task=cur_task, context=cur_context)
        ret = _fire_one_pre_tool(p, ctx_now)
        if isinstance(ret, HookVeto):
            _emit_tool_vetoed(
                tool_name=tool_name, transport=transport,
                correlation_id=correlation_id, veto=ret,
                duration_ms=(time.monotonic() - veto_t0) * 1000.0,
                task_preview=cur_task,
            )
            return ret
        cur_task, cur_context = _apply_pre_tool_mutation(
            p.name, ret, cur_task, cur_context)

    # 2. invoke. If it raises, fire on_error, then re-raise.
    try:
        result = invoke(cur_task, cur_context)
    except BaseException as exc:
        plugins_err = _registry.for_hook("on_error", tool_name)
        ctx_err = replace(ctx, task=cur_task, context=cur_context)
        for p in plugins_err:
            _fire_one_observe(p, "on_error", ctx_err, exc)
        raise

    # 3. post_tool chain — chain mutations: A's output is B's input
    plugins_post = _registry.for_hook("post_tool", tool_name)
    cur_result = result if isinstance(result, str) else (str(result) if result is not None else "")
    for p in plugins_post:
        ctx_now = replace(ctx, task=cur_task, context=cur_context)
        ret = _fire_one_post_tool(p, ctx_now, cur_result)
        if ret is None:
            continue
        if isinstance(ret, str):
            cur_result = ret
        else:
            log.warning("[hooks] plugin %s post_tool returned %s; ignored",
                        p.name, type(ret).__name__)

    return cur_result


def emit_operation_start(
    *,
    operation_id: str,
    transport: str,
    correlation_id: str,
    agent: Optional[str] = None,
    client_id: Optional[str] = None,
) -> None:
    """Fire on_operation_start hooks for every registered plugin.

    Called from voice WebSocket session start, crew run start, chat
    request handler. Not fired for individual MCP tool calls (those
    don't form an operation envelope).
    """
    ctx = HookCtx(
        transport=transport,
        correlation_id=correlation_id,
        plugin_name="",
        timestamp_utc=_now_iso(),
        agent=agent,
        client_id=client_id,
        operation_id=operation_id,
    )
    for p in _registry.for_hook("on_operation_start", tool_name=None):
        _fire_one_observe(p, "on_operation_start", ctx)


def emit_operation_end(
    *,
    operation_id: str,
    transport: str,
    correlation_id: str,
    duration_ms: float,
    outcome: str = "ok",
    agent: Optional[str] = None,
    client_id: Optional[str] = None,
) -> None:
    """Fire on_operation_end hooks. Mirrors emit_operation_start."""
    ctx = HookCtx(
        transport=transport,
        correlation_id=correlation_id,
        plugin_name="",
        timestamp_utc=_now_iso(),
        agent=agent,
        client_id=client_id,
        operation_id=operation_id,
        duration_ms=duration_ms,
        outcome=outcome,
    )
    for p in _registry.for_hook("on_operation_end", tool_name=None):
        _fire_one_observe(p, "on_operation_end", ctx)


def approve_plugin(filename: str, *, approved_by: str = "operator") -> dict:
    """Public API for the `plugin_approve` skill (PR-2F, D-18).

    Resolves `filename` to `<plugins_dir>/<filename>`, computes its SHA-256,
    writes/updates the entry in `~/.codec/plugins.allowlist`, and returns
    a summary dict: `{ok, sha256, last8, filename, reason}`.

    Caller MUST be the operator (the skill is `SKILL_MCP_EXPOSE=False`).
    """
    # Reject path traversal — only accept a basename
    if "/" in filename or ".." in filename or filename.startswith("."):
        return {"ok": False, "reason": "invalid_filename: must be a basename"}
    if not filename.endswith(_PLUGIN_FILE_SUFFIX):
        return {"ok": False, "reason": f"must end with {_PLUGIN_FILE_SUFFIX}"}

    # Use the singleton registry's plugins_dir + allowlist_path so the
    # operator can wire in a custom registry (tests do).
    plugins_dir = _registry.plugins_dir
    allowlist_path = _registry.allowlist_path

    fpath = os.path.join(plugins_dir, filename)
    if not os.path.exists(fpath):
        return {"ok": False, "reason": f"plugin file not found: {filename}"}

    h = _file_sha256(fpath)
    if h is None:
        return {"ok": False, "reason": "could not read plugin file"}

    with _ALLOWLIST_LOCK:
        allowlist = _read_allowlist(allowlist_path)
        allowlist[filename] = {
            "sha256": h,
            "approved_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
            "approved_by": approved_by,
        }
        if not _write_allowlist(allowlist, allowlist_path):
            return {"ok": False, "reason": "allowlist write failed"}

    # Invalidate the broken-plugin cache so the plugin is re-attempted on
    # next fire (the operator just approved it; clear the prior refusal).
    plugin_name = filename[:-len(_PLUGIN_FILE_SUFFIX)]
    with _registry._lock:
        _registry._broken.discard(plugin_name)
        _registry._modules.pop(plugin_name, None)

    try:
        _log_event(
            "plugin_approved", "codec-hooks",
            f"Plugin {filename} approved",
            extra={"filename": filename, "sha256_last8": h[-8:],
                   "approved_by": approved_by},
            level="info", outcome="ok",
        )
    except Exception:
        pass

    return {
        "ok": True, "sha256": h, "last8": h[-8:],
        "filename": filename, "approved_by": approved_by,
    }


__all__ = [
    "HookCtx",
    "HookVeto",
    "PluginRegistry",
    "run_with_hooks",
    "emit_operation_start",
    "emit_operation_end",
    "approve_plugin",
]