feat(ch01/round4): bootstrap the Hooks abstraction — configured hooks fire in production

src/agent/subagent_context.py

@@ -184,6 +184,14 @@ def create_subagent_context(
         agent_id=agent_id,
         agent_type=agent_type,
         user_modified=parent_context.user_modified,
+        # ch01 round-4 WI-1 — hooks apply to sub-agents exactly as to the
+        # parent: same config snapshot, same workspace-trust verdict.
+        # Without this, PreToolUse/PostToolUse (incl. enterprise policy
+        # hooks) silently skip every sub-agent tool call — a policy bypass
+        # via the Agent tool — and the PostSampling trust filter would
+        # treat sub-agent loops as untrusted in a trusted workspace.
+        hook_config_manager=parent_context.hook_config_manager,
+        workspace_trusted=parent_context.workspace_trusted,
     )
 
 

src/entrypoints/headless.py

@@ -213,6 +213,27 @@ def run_headless(options: HeadlessOptions) -> int:
         abort_controller=abort_controller,
     )
     tool_context.options.is_non_interactive_session = True
+    # ch01 round-4 WI-1 — load settings hooks into the executor-visible
+    # snapshot + global registry. Safe here: run_headless is sync and its
+    # asyncio.run happens later. Never raises.
+    from src.hooks.config_manager import bootstrap_hook_config_manager
+
+    tool_context.hook_config_manager = bootstrap_hook_config_manager(
+        cwd=str(workspace_root),
+    )
+    # workspace_trusted feeds the hook trust gate (trust_gate WI-0.2); it
+    # defaulted False with no production setter, so even configured hooks
+    # were silently skipped. Same source of truth as the CLI startup gate.
+    try:
+        from src.services.startup_gates import check_trust_accepted
+
+        tool_context.workspace_trusted = check_trust_accepted(workspace_root)
+    except Exception:  # noqa: BLE001 — unknown trust stays untrusted
+        import logging
+
+        logging.getLogger(__name__).debug(
+            "headless trust check failed", exc_info=True,
+        )
     if options.skip_permissions or effective_mode == "bypassPermissions":
         tool_context.allow_docs = True
         tool_context.permission_handler = None

src/hooks/config_manager.py

@@ -290,7 +290,30 @@ def load_hooks_from_settings(
                 translated = _translate_legacy_notification_entry(hook_raw)
                 if translated is not None:
                     target_event = translated
-            hooks.setdefault(target_event, []).append(_parse_hook_config(hook_raw))
+            # ch01 round-4 WI-1 — canonical Claude Code settings nest hook
+            # definitions in matcher groups: {"matcher": ..., "hooks": [...]}
+            # (TS schemas/hooks.ts HookMatcherSchema). Expand each group,
+            # propagating the group matcher onto inner entries that don't
+            # set their own. The flat form (a bare hook dict) stays
+            # supported. Without the expansion a real-world settings.json
+            # parsed into empty-command junk hooks.
+            inner = hook_raw.get("hooks")
+            if isinstance(inner, list):
+                group_matcher = hook_raw.get("matcher")
+                for inner_raw in inner:
+                    if not isinstance(inner_raw, dict):
+                        continue
+                    if group_matcher is not None and "matcher" not in inner_raw:
+                        inner_raw = {**inner_raw, "matcher": group_matcher}
+                    config = _parse_hook_config(inner_raw)
+                    if config.type == "command" and not config.command:
+                        continue  # malformed entry — never execute ""
+                    hooks.setdefault(target_event, []).append(config)
+                continue
+            config = _parse_hook_config(hook_raw)
+            if config.type == "command" and not config.command:
+                continue  # malformed entry — never execute ""
+            hooks.setdefault(target_event, []).append(config)
 
     return HookConfigSnapshot(
         hooks=hooks,
@@ -372,3 +395,89 @@ async def validate(self) -> list[HookValidationError]:
             )]
 
         return validate_hook_configs(hooks_raw)
+
+
+def bootstrap_hook_config_manager(
+    *,
+    cwd: str | Path | None = None,
+    settings_path: str | Path | None = None,
+) -> HookConfigManager | None:
+    """Build + load the manager that makes settings hooks live.
+
+    ch01 round-4 WI-1. This is the missing root of the Hooks abstraction:
+    the executors all read hook configs through
+    ``tool_use_context.hook_config_manager.snapshot``
+    (``hook_executor._get_hooks_from_snapshot``), and the router lane
+    (post-sampling / session hooks) reads the global ``AsyncHookRegistry``
+    — but nothing constructed or loaded a manager in production, so
+    configured hooks never fired. Both production ``ToolContext``
+    construction sites (agent-server ``_build_runtime``, headless
+    ``run_headless``) call this and attach the result.
+
+    ``load()`` populates BOTH read paths at once: the frozen snapshot
+    (context lane) and the global registry (router lane), because every
+    parsed config is registered into the registry the manager was
+    constructed with.
+
+    Sync-only by contract: both call sites are plain sync functions running
+    on threads with no live event loop (the agent-server builds runtimes in
+    ``run_in_executor``; headless bootstraps before its own
+    ``asyncio.run``). If a running loop is detected, log and return None
+    rather than deadlock — an async call site needs an async variant, not
+    this one.
+
+    Never raises: hooks must not be able to break startup. Returns None
+    when ``settings.hooks.enabled`` is False (the framework off-switch;
+    unreadable settings fail open to the default of enabled).
+
+    ``cwd`` asymmetry, deliberate: ``cwd`` scopes only the
+    ``hooks.enabled`` knob lookup (settings hierarchy); the hook *entries*
+    always load from the single user-scope file
+    (``$CLAUDE_CONFIG_DIR``/``~/.claude/settings.json``) regardless of
+    cwd. Multi-scope (project/local/policy) entry merging is the ch12
+    round-4 subject.
+    """
+    try:
+        from ..settings.settings import load_settings
+
+        if not load_settings(cwd=cwd).hooks.enabled:
+            logger.info("hooks disabled via settings.hooks.enabled")
+            return None
+    except Exception:  # noqa: BLE001 — the knob is an off-switch, not a gate
+        logger.debug("could not read settings.hooks.enabled; assuming enabled",
+                     exc_info=True)
+
+    import asyncio
+
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        pass  # no running loop — the expected state at both call sites
+    else:
+        # Documented sync-only contract: asyncio.run() below would raise, and
+        # probing first avoids creating a never-awaited load() coroutine.
+        logger.warning(
+            "bootstrap_hook_config_manager called from a running event loop; "
+            "hooks not loaded (use an async variant at this call site)",
+        )
+        return None
+
+    try:
+        from .registry import get_global_hook_registry
+
+        manager = HookConfigManager(
+            get_global_hook_registry(), settings_path=settings_path,
+        )
+        asyncio.run(manager.load())
+        snapshot = manager.snapshot
+        if snapshot is not None and not snapshot.is_empty:
+            count = sum(len(v) for v in snapshot.hooks.values())
+            logger.info(
+                "loaded %d hook config(s) from %s",
+                count, snapshot.source_path,
+            )
+        return manager
+    except Exception:  # noqa: BLE001 — hooks must never break startup
+        logger.warning("hook config bootstrap failed; continuing without hooks",
+                       exc_info=True)
+        return None

src/hooks/hook_types.py

@@ -9,7 +9,7 @@
 # Hook events — Chapter-12 / Phase-1 / WI-1.1
 # ---------------------------------------------------------------------------
 #
-# 25-event taxonomy promoted from the legacy 10-event Literal. Mirrors
+# 28-event taxonomy promoted from the legacy 10-event Literal. Mirrors
 # typescript/src/utils/hooks/hooksConfigManager.ts:26-267 plus the chapter's
 # reference table (``ch12-extensibility.md`` §"Five Most Important Lifecycle
 # Events" + §"Reference table — remaining events").

src/hooks/post_sampling_hooks.py

@@ -15,11 +15,19 @@ async def run_post_sampling_hooks(
     model: str = "",
     usage: dict[str, int] | None = None,
     stop_reason: str | None = None,
-    response_content: Any = None,
+    untrusted_workspace: bool = False,
 ) -> list[dict[str, Any]]:
     reg = registry or get_global_hook_registry()
     hooks = await reg.get_hooks_for_event("PostSampling")
 
+    if untrusted_workspace:
+        # Same trust rule the snapshot-lane executor applies
+        # (hook_executor._run_hooks_for_event / trust_gate WI-0.2): in an
+        # untrusted workspace only enterprise policy hooks may run. Without
+        # this, the registry lane would execute user-settings shell hooks
+        # that the tool-hook lane correctly blocks.
+        hooks = [h for h in hooks if h.source.is_policy]
+
     if not hooks:
         return []
 

src/hooks/registry.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-import asyncio
 import logging
+import threading
 from dataclasses import dataclass, field
 from typing import Any, AsyncGenerator
 
@@ -41,7 +41,16 @@ def __init__(self) -> None:
         self._hooks: dict[HookEvent, list[RegisteredHook]] = {
             event: [] for event in ALL_HOOK_EVENTS
         }
-        self._lock = asyncio.Lock()
+        # threading.Lock, not asyncio.Lock — deliberately (ch01 round-4).
+        # The registry is a process-global singleton touched from MULTIPLE
+        # event loops (the sync bootstrap's asyncio.run, each turn's
+        # asyncio.run on the agent-server worker thread) and from multiple
+        # session worker threads. An asyncio.Lock binds to whichever loop
+        # first awaits it and raises "bound to a different event loop" ever
+        # after; it also provides zero cross-thread safety. Every critical
+        # section below is pure in-memory dict/list work with no await, so a
+        # plain mutex is both correct and loop-agnostic.
+        self._lock = threading.Lock()
         self._counter = 0
 
     async def register(
@@ -50,7 +59,7 @@ async def register(
         config: HookConfig,
         source: HookSource | None = None,
     ) -> RegisteredHook:
-        async with self._lock:
+        with self._lock:
             effective_source = source or config.source
             self._counter += 1
             hook = RegisteredHook(
@@ -76,7 +85,7 @@ async def deregister(
         event: HookEvent,
         config: HookConfig,
     ) -> bool:
-        async with self._lock:
+        with self._lock:
             existing = self._hooks.get(event, [])
             temp_hook = RegisteredHook(event=event, config=config, source=config.source)
             dedup_key = temp_hook.dedup_key
@@ -90,7 +99,7 @@ async def get_hooks_for_event(
         event: HookEvent,
         tool_name: str | None = None,
     ) -> list[RegisteredHook]:
-        async with self._lock:
+        with self._lock:
             hooks = list(self._hooks.get(event, []))
 
         if tool_name is not None:
@@ -110,13 +119,13 @@ async def has_hooks_for_event(
         return len(hooks) > 0
 
     async def clear(self) -> None:
-        async with self._lock:
+        with self._lock:
             for event in ALL_HOOK_EVENTS:
                 self._hooks[event] = []
             self._counter = 0
 
     async def clear_source(self, source: HookSource) -> int:
-        async with self._lock:
+        with self._lock:
             removed = 0
             for event in ALL_HOOK_EVENTS:
                 before = len(self._hooks[event])

src/query/query.py

@@ -178,6 +178,68 @@ def _create_max_turns_attachment(max_turns: int, turn_count: int) -> SystemMessa
     )
 
 
+async def _fire_post_sampling_hooks(
+    assistant_messages: list[AssistantMessage],
+    provider: Any,
+    tool_use_context: Any,
+) -> None:
+    """Run configured ``PostSampling`` hooks for a completed model stream.
+
+    ch01 round-4 WI-2 — restores the dependency-graph edge "Query Loop
+    fires Hooks" for the one event TS fires from query.ts itself
+    (``executePostSamplingHooks``, query.ts:1079-1089). Reads the global
+    ``AsyncHookRegistry`` (populated at startup by
+    ``bootstrap_hook_config_manager``); with no ``PostSampling`` hooks
+    configured this returns after one in-memory lookup.
+
+    Deviation from TS, deliberate: awaited inline instead of
+    fire-and-forget. The agent-server drives each turn with
+    ``asyncio.run(...)``, so a task created on the loop's final iteration
+    would be cancelled at teardown — losing the hook on exactly the
+    turn-final response. Inline await trades stream/hook overlap for a
+    completion guarantee.
+
+    Hook failures are logged and swallowed — a hook must never kill the
+    turn (TS parity: logError + continue). ``additional_contexts`` from
+    hook results are debug-logged and dropped: TS post-sampling hooks
+    return void, so there are no injection semantics to mirror; a uniform
+    injection lane across events is the ch12 round-4 subject.
+    """
+    if not assistant_messages:
+        return
+    try:
+        from ..hooks.post_sampling_hooks import run_post_sampling_hooks
+        from ..hooks.trust_gate import should_skip_hook_due_to_trust
+
+        last = assistant_messages[-1]
+        results = await run_post_sampling_hooks(
+            model=(
+                getattr(last, "model", None)
+                or getattr(provider, "model", None)
+                or ""
+            ),
+            usage=getattr(last, "usage", None) or {},
+            stop_reason=getattr(last, "stop_reason", None),
+            # Same trust rule as the tool-hook lane: untrusted workspace →
+            # policy hooks only (trust_gate WI-0.2).
+            untrusted_workspace=should_skip_hook_due_to_trust(tool_use_context),
+        )
+        for entry in results:
+            injected = entry.get("injected_messages")
+            if injected:
+                logger.debug(
+                    "PostSampling hook additional_contexts dropped "
+                    "(no injection lane yet — ch12): %d block(s)",
+                    len(injected),
+                )
+    except (asyncio.CancelledError, AbortError):
+        # User intent wins — AbortError subclasses Exception, so without the
+        # explicit re-raise the blanket handler below would swallow it.
+        raise
+    except Exception:  # noqa: BLE001 — hook failure must never kill the turn
+        logger.error("PostSampling hook execution failed", exc_info=True)
+
+
 def _drain_pending_user_messages(tool_use_context: Any) -> list[UserMessage]:
     """Drain the running agent's ``pending_messages`` inbox, if any.
 
@@ -1007,6 +1069,10 @@ def _do_provider_call():
     assistant_msg = AssistantMessage(
         content=assistant_blocks if assistant_blocks else "",
         stop_reason=stop_reason,
+        # TS assistant messages carry the responding model (query.ts message
+        # assembly); consumers like the PostSampling hook payload and cost
+        # attribution read it. Same fallback chain as record_api_usage above.
+        model=getattr(response, "model", None) or getattr(provider, "model", None),
         usage=response.usage,
     )
     if response.reasoning_content:
@@ -1037,9 +1103,11 @@ async def query(
     See :func:`run_query` for a convenience helper that consumes the
     generator and returns ``(messages, terminal)``.
 
-    This PR (Phase A) introduces the typed Terminal infrastructure;
-    recovery integration, stop hooks, token budget, model fallback,
-    and continuation nudge land in subsequent PRs.
+    Recovery integration, stop hooks, token budget, and the continuation
+    nudge are wired (ch05 rounds 2-3). Model fallback (TS
+    FallbackTriggeredError → sticky model switch + tombstones) is NOT yet
+    ported — tracked as ch04/ch05 round-4 scope; see
+    my-docs/ch01-architecture-round4-gap-analysis.md §3 GAP B.
     """
     _diag = os.environ.get("CLAWCODEX_DEBUG", "").lower() in ("1", "true", "yes")
     holder = terminal_holder or TerminalHolder()
@@ -1363,6 +1431,16 @@ def _marking_chunk_cb(text: str) -> None:
             set_terminal(holder, natural_termination, Terminal(reason="aborted_streaming"))
             return
 
+        # ch01 round-4 WI-2 — PostSampling hook wire (TS query.ts:1079-1089).
+        # Placed after the abort check: TS fires before it, but its call is
+        # non-blocking (`void …`) so pre-abort is free there; an inline await
+        # before the abort return would delay ESC responsiveness by the hook
+        # runtime. Consequence: hooks do not fire for user-aborted streams.
+        if assistant_messages:
+            await _fire_post_sampling_hooks(
+                assistant_messages, params.provider, tool_use_context,
+            )
+
         if not needs_follow_up:
             last_message = assistant_messages[-1] if assistant_messages else None