feat(code): integrate Hooks v2 client lifecycle events

libs/code/deepagents_code/_cli_context.py

@@ -51,6 +51,12 @@ class CLIContextSchema:
 
     offload_tool_call_id: str | None = None
 
+    hooks_snapshot_id: str | None = None
+
+    hooks_server_events: list[str] = field(default_factory=list)
+
+    prompt_id: str | None = None
+
 
 class CLIContext(TypedDict, total=False):
     """Client-facing builder for the per-run graph context payload.
@@ -107,3 +113,20 @@ class CLIContext(TypedDict, total=False):
     This is set by the client, not graph state, so model-generated calls cannot
     grant themselves permission to execute during the hidden compaction turn.
     """
+
+    hooks_snapshot_id: str | None
+    """Canonical Hooks v2 configuration hash for this session.
+
+    Server-owned lifecycle middleware includes this id on interrupt requests so
+    the client can reject mismatched resumes.
+    """
+
+    hooks_server_events: list[str]
+    """Server-owned HookEvent names that have configured handlers.
+
+    Middleware only interrupts for events listed here, avoiding a round-trip
+    when the session snapshot has no matching handlers.
+    """
+
+    prompt_id: str | None
+    """Optional per-turn prompt id projected into hook context."""

libs/code/deepagents_code/agent.py

@@ -1929,6 +1929,14 @@ def _should_interrupt_tool_call(
     Returns:
         `True` to interrupt, or `False` for Auto/YOLO bypass.
     """
+    from deepagents_code.hooks.server_middleware import pre_tool_behavior
+
+    tool_call = getattr(request, "tool_call", None)
+    tool_call_id = str(tool_call.get("id") or "") if isinstance(tool_call, dict) else ""
+    hook_behavior = pre_tool_behavior(getattr(request, "state", None), tool_call_id)
+    if hook_behavior in {"allow", "deny"}:
+        return False
+
     runtime = getattr(request, "runtime", None)
     mode = _async_routing_mode(getattr(request, "state", None))
     if mode is None:
@@ -2447,6 +2455,20 @@ def _subagent_cli_middleware(
             middleware.append(_GlmTerminalStallRecovery())
         if restrictive_shell_allow_list is not None:
             middleware.append(ShellAllowListMiddleware(restrictive_shell_allow_list))
+        # Server-owned hooks must wrap subagent tools too; otherwise Pre/Post
+        # ToolUse only fire on the parent graph. Disable Stop so finishing a
+        # subagent does not emit the main-agent Stop event (SubagentStop still
+        # fires from the parent wrap around `task`).
+        from deepagents_code.hooks.server_middleware import ServerHooksMiddleware
+
+        hooks_cwd = Path(effective_cwd) if effective_cwd is not None else Path.cwd()
+        middleware.append(
+            ServerHooksMiddleware(
+                cwd=hooks_cwd,
+                emit_stop=False,
+                mcp_tools=mcp_tools,
+            )
+        )
         # Subagents share the on-disk filesystem backend and can edit the user
         # AGENTS.md, so they get the same managed onboarding-name block guard as
         # the main agent. Gated on memory because the block only exists when
@@ -2741,24 +2763,19 @@ def _subagent_cli_middleware(
             fs_tools=fs_tools,
         )
 
-    interrupt_on: dict[str, bool | InterruptOnConfig] | None
+    interrupt_on: dict[str, bool | InterruptOnConfig] = {}
     auto_mode_config: tuple[Path, list[str]] | None = None
-    if resolved_interrupt_on is None:
-        interrupt_on = {}
-    else:
-        interrupt_on = resolved_interrupt_on  # ty: ignore[invalid-assignment]  # InterruptOnConfig is compatible at runtime
-        if auto_mode_enabled:
-            configured_allow_list = shell_allow_list or settings.shell_allow_list
-            narrow_allow_list = (
-                configured_allow_list if isinstance(configured_allow_list, list) else []
-            )
-            trusted_root = (
-                project_context.project_root
-                if project_context is not None
-                and project_context.project_root is not None
-                else effective_cwd or Path.cwd()
-            )
-            auto_mode_config = (Path(trusted_root), narrow_allow_list)
+    if resolved_interrupt_on is not None and auto_mode_enabled:
+        configured_allow_list = shell_allow_list or settings.shell_allow_list
+        narrow_allow_list = (
+            configured_allow_list if isinstance(configured_allow_list, list) else []
+        )
+        trusted_root = (
+            project_context.project_root
+            if project_context is not None and project_context.project_root is not None
+            else effective_cwd or Path.cwd()
+        )
+        auto_mode_config = (Path(trusted_root), narrow_allow_list)
 
     # Set up composite backend with routing.
     if sandbox is None:
@@ -2818,6 +2835,20 @@ def _subagent_cli_middleware(
                 trusted_compaction_tool=compaction_middleware.tools[0],
             )
         )
+    elif resolved_interrupt_on is not None:
+        # `AutoModeHITLMiddleware` reports the same `HumanInTheLoopMiddleware`
+        # name, so installing both would trip `create_agent`'s duplicate-name
+        # assertion. Auto mode's specialized replacement wins when active.
+        agent_middleware.append(AsyncApprovalHITLMiddleware(resolved_interrupt_on))
+
+    # Server-owned Hooks v2 lifecycle events (Pre/Post tool, Stop, subagent).
+    # Gated at runtime by `hooks_server_events` on the per-run context so idle
+    # sessions without configured handlers pay no interrupt round-trip. Appended
+    # after the HITL middleware so `PreToolUse` resolves before approval routing.
+    from deepagents_code.hooks.server_middleware import ServerHooksMiddleware
+
+    hooks_cwd = Path(effective_cwd) if effective_cwd is not None else Path.cwd()
+    agent_middleware.append(ServerHooksMiddleware(cwd=hooks_cwd, mcp_tools=mcp_tools))
 
     if fs_tools is not None:
         # `fs_tools` is an explicit allowlist here (`--allow-fs-tools all` and an