feat(ch12): phase 0 — snapshot rewire + workspace-trust gate

pyproject.toml

@@ -50,7 +50,7 @@ dev = [
     # ``pyproject.toml`` declares ``asyncio_mode = "auto"`` (below) which
     # requires the pytest-asyncio plugin. Without it, async tests fail at
     # run time with "async def functions are not natively supported."
-    "pytest-asyncio>=0.23",
+    "pytest-asyncio>=1.3.0",
 ]
 
 [project.scripts]

requirements.txt

@@ -13,3 +13,4 @@ pathspec>=0.11
 build>=1.0.0
 twine>=5.0.0
 pytest>=8.0.0
+pytest-asyncio>=1.3.0

src/hooks/hook_executor.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import time
+import warnings
 from typing import Any, AsyncGenerator
 from uuid import uuid4
 
@@ -28,6 +29,7 @@
     HookEvent,
     HookResult,
 )
+from src.hooks.trust_gate import should_skip_hook_due_to_trust
 from src.types.messages import (
     create_attachment_message,
     create_progress_message,
@@ -36,38 +38,87 @@
 logger = logging.getLogger(__name__)
 
 
-def _get_hooks_from_settings(tool_use_context: Any) -> dict[str, list[HookConfig]]:
+def _get_hooks_from_snapshot(tool_use_context: Any) -> dict[str, list[HookConfig]]:
+    """Read hooks from the frozen snapshot held on the active HookConfigManager.
+
+    Mirrors typescript/src/utils/hooks/hooksConfigSnapshot.ts:119-124
+    (``getHooksConfigFromSnapshot``). settings.json is never re-read implicitly;
+    snapshot updates flow only through ``HookConfigManager.load()`` (startup) or
+    ``HookConfigManager.reload_if_changed()`` (explicit /hooks command).
+
+    **Back-compat:** if ``hook_config_manager`` is not set on the context but
+    ``options.hooks`` is, fall back to the legacy read path with a
+    ``DeprecationWarning``. This keeps existing callers / tests working during
+    one CHANGELOG cycle. After the deprecation cycle, the fallback is removed.
+    """
+    manager = getattr(tool_use_context, "hook_config_manager", None)
+    if manager is not None:
+        snapshot = getattr(manager, "snapshot", None)
+        if snapshot is None:
+            # Manager exists but hasn't been loaded — treat as empty.
+            return {}
+        # Defensive copy: callers MUST NOT mutate the returned dict.
+        return {ev: list(hooks) for ev, hooks in snapshot.hooks.items()}
+
+    # Legacy fallback — emit DeprecationWarning if options.hooks carries data.
+    return _get_hooks_from_options_legacy(tool_use_context)
+
+
+def _get_hooks_from_options_legacy(tool_use_context: Any) -> dict[str, list[HookConfig]]:
+    """Legacy read path: ``tool_use_context.options.hooks``.
+
+    Bypasses the snapshot freezing semantic introduced in WI-0.1. Preserved for
+    one CHANGELOG cycle so existing callers / tests do not break in lockstep
+    with the rewire. Emits a ``DeprecationWarning`` when it actually returns
+    data (silent no-op when options.hooks is empty/None).
+    """
     try:
         options = getattr(tool_use_context, "options", None)
         if options is None:
             return {}
         hooks_config = getattr(options, "hooks", None)
-        if hooks_config is None:
+        if hooks_config is None or not hooks_config:
+            return {}
+        if not isinstance(hooks_config, dict):
             return {}
-        if isinstance(hooks_config, dict):
-            result: dict[str, list[HookConfig]] = {}
-            for event_name, hook_list in hooks_config.items():
-                if isinstance(hook_list, list):
-                    configs = []
-                    for h in hook_list:
-                        if isinstance(h, dict):
-                            configs.append(HookConfig(
-                                type=h.get("type", "command"),
-                                command=h.get("command", ""),
-                                timeout=h.get("timeout"),
-                                matcher=h.get("matcher"),
-                            ))
-                        elif isinstance(h, HookConfig):
-                            configs.append(h)
-                    result[event_name] = configs
-            return result
+        result: dict[str, list[HookConfig]] = {}
+        for event_name, hook_list in hooks_config.items():
+            if isinstance(hook_list, list):
+                configs = []
+                for h in hook_list:
+                    if isinstance(h, dict):
+                        configs.append(HookConfig(
+                            type=h.get("type", "command"),
+                            command=h.get("command", ""),
+                            timeout=h.get("timeout"),
+                            matcher=h.get("matcher"),
+                        ))
+                    elif isinstance(h, HookConfig):
+                        configs.append(h)
+                result[event_name] = configs
+        if result:
+            warnings.warn(
+                "Reading hooks from tool_use_context.options.hooks is deprecated; "
+                "wire a HookConfigManager onto tool_use_context.hook_config_manager "
+                "and call .load() at bootstrap. The legacy path bypasses the snapshot "
+                "security model (chapter §'The Snapshot Security Model'). Will be "
+                "removed two CHANGELOG entries after the rename.",
+                DeprecationWarning,
+                stacklevel=3,
+            )
+        return result
     except Exception:
-        pass
-    return {}
+        return {}
+
+
+# Back-compat alias — preserved for any external test fixtures still importing
+# ``_get_hooks_from_settings`` directly. New code uses ``_get_hooks_from_snapshot``.
+def _get_hooks_from_settings(tool_use_context: Any) -> dict[str, list[HookConfig]]:
+    return _get_hooks_from_snapshot(tool_use_context)
 
 
 def has_hook_for_event(event: str, tool_use_context: Any) -> bool:
-    hooks = _get_hooks_from_settings(tool_use_context)
+    hooks = _get_hooks_from_snapshot(tool_use_context)
     return bool(hooks.get(event))
 
 
@@ -207,9 +258,25 @@ async def _run_hooks_for_event(
     abort_signal: Any | None = None,
     timeout_ms: int = TOOL_HOOK_EXECUTION_TIMEOUT_MS,
 ) -> AsyncGenerator[dict[str, Any], None]:
-    hooks = _get_hooks_from_settings(tool_use_context)
+    # WI-0.2 — workspace-trust gate. Skip non-policy hooks while the workspace
+    # is untrusted. The per-hook policy check happens below since policy-source
+    # identification is per-HookConfig.
+    trust_skip = should_skip_hook_due_to_trust(tool_use_context)
+
+    # WI-0.1 — read from the frozen snapshot, not from options.hooks. The
+    # snapshot is built once at startup by HookConfigManager.load() and is
+    # immune to settings.json mutation between trust acceptance and tool calls.
+    hooks = _get_hooks_from_snapshot(tool_use_context)
     event_hooks = hooks.get(event, [])
 
+    if trust_skip:
+        # Drop everything that isn't a policy-source hook. ``HookConfig.source``
+        # is a ``HookSource`` enum; any non-POLICY value is gated. Imported
+        # locally to avoid pulling the enum into module-init paths that don't
+        # need it.
+        from src.hooks.hook_types import HookSource
+        event_hooks = [h for h in event_hooks if h.source == HookSource.POLICY]
+
     tool_use_id = stdin_data.get("tool_use_id", str(uuid4()))
     parent_tool_use_id = ""
 

src/hooks/trust_gate.py

@@ -0,0 +1,37 @@
+"""Workspace-trust gate for hook execution.
+
+Mirrors the TypeScript ``shouldSkipHookDueToTrust`` predicate (referenced from
+``typescript/src/utils/hooks/hooksConfigManager.ts``). The chapter
+(``ch12-extensibility.md`` §"The Snapshot Security Model") describes this as a
+centralized gate at the top of ``executeHooks()`` — introduced after two
+vulnerabilities where hooks fired in lifecycle states the user had not consented
+to.
+
+Failure mode without this gate: a project's ``.claude/settings.json`` defines a
+hook that fires before every tool call. The user opens the project, declines the
+workspace-trust dialog, and the hook still fires once because ``SessionStart``
+runs before the trust check. The gate closes that window.
+
+Policy hooks (``HookSource.POLICY_SETTINGS``) are NOT subject to this gate per
+chapter §"The Snapshot Security Model" final paragraph: "the policy layer
+always wins."
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def should_skip_hook_due_to_trust(tool_use_context: Any) -> bool:
+    """Return True if non-policy hooks must be skipped because the workspace
+    isn't trusted.
+
+    Reads ``tool_use_context.workspace_trusted`` (a bool added on ``ToolContext``
+    in WI-0.2). Defaults to ``False`` if the attribute is missing — fail-safe:
+    unknown trust state is treated as untrusted.
+
+    Returns ``True`` (skip) iff workspace_trusted is False. Callers must still
+    let policy-source hooks through; the per-hook policy check happens at the
+    caller, not here, because policy-source identification is per-``HookConfig``.
+    """
+    return not getattr(tool_use_context, "workspace_trusted", False)

src/tool_system/context.py

@@ -141,6 +141,25 @@ class ToolContext:
     # string in that case.
     session_id: str | None = None
 
+    # Chapter-12 / Phase 0 / WI-0.1 — frozen snapshot of hook config.
+    # The snapshot is built once at startup by ``HookConfigManager.load()``
+    # and updated only via explicit channels (the ``/hooks`` command or
+    # an explicit ``reload_if_changed()`` call). Hook execution reads from
+    # ``hook_config_manager.snapshot`` instead of ``options.hooks`` so a
+    # malicious post-trust mutation of ``settings.json`` cannot affect
+    # in-flight tool calls.
+    #
+    # ``options.hooks`` survives as a deprecated fallback for one release
+    # cycle (see ``_get_hooks_from_snapshot`` in ``src/hooks/hook_executor.py``):
+    # callers that still pass hooks via options get a ``DeprecationWarning``
+    # but their behavior is preserved.
+    hook_config_manager: Any | None = None
+    # Chapter-12 / Phase 0 / WI-0.2 — workspace-trust gate. Bootstrap flips
+    # this to ``True`` after the user accepts the trust dialog. Hooks (other
+    # than ``HookSource.POLICY``) are skipped while the workspace is
+    # untrusted, mirroring TS' ``shouldSkipHookDueToTrust`` gate.
+    workspace_trusted: bool = False
+
     def __post_init__(self) -> None:
         self.workspace_root = Path(self.workspace_root).resolve()
         if self.cwd is None: