feat(code): integrate Hooks v2 server 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

libs/code/deepagents_code/app.py

@@ -2267,6 +2267,8 @@ def __init__(
         # Assign the backing field directly: the setter reads `self._thread_id`
         # to detect a thread change, and it isn't set yet.
         self._thread_id = thread_id or _new_thread_id()
+        # Optional session-scoped Hooks v2 client runtime.
+        self.hooks_runtime = None
 
     @property
     def auto_approve(self) -> bool:
@@ -4291,10 +4293,25 @@ async def _init_session_state(self) -> None:
         """Create session state in a thread (imports deepagents_code.sessions)."""
 
         def _create() -> TextualSessionState:
-            return TextualSessionState(
+            from pathlib import Path
+
+            from deepagents_code.hooks.runtime import HooksRuntime
+
+            state = TextualSessionState(
                 approval_mode=self._approval_mode,
                 thread_id=self._lc_thread_id,
             )
+            try:
+                # Interactive sessions keep project hooks off until a dedicated
+                # workspace-trust prompt lands (design-doc security follow-up).
+                state.hooks_runtime = HooksRuntime.create(
+                    cwd=Path(self._cwd),
+                    workspace_trusted=False,
+                )
+            except Exception:
+                logger.exception("Failed to create HooksRuntime; server hooks disabled")
+                state.hooks_runtime = None
+            return state
 
         try:
             session_state = await asyncio.to_thread(_create)

libs/code/deepagents_code/auto_mode.py

@@ -2566,13 +2566,21 @@ async def aafter_model(
         )
         if ai_message is None or not ai_message.tool_calls:
             return {"_auto_decision_plan": None}
+        from deepagents_code.hooks.server_middleware import pre_tool_behavior
+
+        hook_bypass_ids = {
+            _tool_call_id(call)
+            for call in ai_message.tool_calls
+            if pre_tool_behavior(state, _tool_call_id(call)) in {"allow", "deny"}
+        }
         thread_key = _thread_key(runtime)
         plan = self._validated_plan(state, ai_message, thread_key)
         current_mode, current_mode_unavailable = await _live_mode(runtime)
         manual_ids = {
             _tool_call_id(call)
             for call in ai_message.tool_calls
             if call["name"] in self.interrupt_on
+            and _tool_call_id(call) not in hook_bypass_ids
         }
         if plan is None:
             if not manual_ids:
@@ -2622,6 +2630,9 @@ async def aafter_model(
                 current_mode = ApprovalMode.MANUAL
 
         if proposal_mode is ApprovalMode.MANUAL or current_mode is ApprovalMode.MANUAL:
+            review_ids = set(plan["manual_gated_ids"]) - hook_bypass_ids
+            if not review_ids:
+                return {"_auto_decision_plan": None}
             manual_fallback = plan["fallback_reason"] in {
                 "approval_mode_unavailable",
                 "control_state_unavailable",
@@ -2635,7 +2646,7 @@ async def aafter_model(
                 state,
                 runtime,
                 ai_message,
-                set(plan["manual_gated_ids"]),
+                review_ids,
                 fallback=manual_fallback,
                 counters=counters,
                 all_manual_ids=manual_ids,
@@ -2650,7 +2661,9 @@ async def aafter_model(
             return {"_auto_decision_plan": None}
 
         decision_by_id = {
-            decision["tool_call_id"]: decision for decision in plan["decisions"]
+            decision["tool_call_id"]: decision
+            for decision in plan["decisions"]
+            if decision["tool_call_id"] not in hook_bypass_ids
         }
         human_ids = {
             tool_id

libs/code/deepagents_code/client/non_interactive.py

@@ -87,6 +87,9 @@
     from deepagents import FsToolName
     from langchain_core.runnables import RunnableConfig
 
+    from deepagents_code.hooks.runtime import HooksRuntime
+    from deepagents_code.json_types import JsonObject
+
 logger = logging.getLogger(__name__)
 
 
@@ -354,6 +357,15 @@ class StreamState:
     Used to resume the agent after HITL processing.
     """
 
+    pending_hook_interrupts: dict[str, object] = field(default_factory=dict)
+    """Raw Hooks v2 invocation interrupt payloads awaiting client fulfillment."""
+
+    hook_response: dict[str, JsonObject] = field(default_factory=dict)
+    """Resume values for fulfilled Hooks v2 interrupts, keyed by interrupt id."""
+
+    hooks_runtime: HooksRuntime | None = None
+    """Optional session-scoped HooksRuntime used to fulfill server hook interrupts."""
+
     interrupt_occurred: bool = False
     """Flag indicating whether any HITL interrupt was received during the
     current stream pass."""
@@ -419,9 +431,15 @@ def _process_interrupts(
         state: Stream state to update with new pending interrupts.
         console: Rich console for user-visible warnings.
     """
+    from deepagents_code.hooks.interrupt import is_hook_interrupt_payload
+
     interrupts = data["__interrupt__"]
     if interrupts:
         for interrupt_obj in interrupts:
+            if is_hook_interrupt_payload(interrupt_obj.value):
+                state.pending_hook_interrupts[interrupt_obj.id] = interrupt_obj.value
+                state.interrupt_occurred = True
+                continue
             try:
                 validated_request = _HITL_REQUEST_ADAPTER.validate_python(
                     interrupt_obj.value
@@ -933,6 +951,36 @@ def _collect_action_request_warnings(action_request: ActionRequest) -> list[str]
     return warnings
 
 
+async def _fulfill_pending_hook_interrupts(state: StreamState) -> None:
+    """Execute pending server-owned hook interrupts on the client runtime.
+
+    Raises:
+        RuntimeError: If a hook interrupt arrives without a session runtime, or
+            if a payload cannot be parsed.
+        HooksSnapshotChangedError: If a hook resume cannot be applied because it
+            was made against a stale configuration snapshot.
+    """
+    if not state.pending_hook_interrupts:
+        return
+    from deepagents_code.hooks.client import (
+        HooksSnapshotChangedError,
+        fulfill_pending_hook_interrupts,
+    )
+
+    if state.hooks_runtime is None:
+        msg = "Received hook invocation interrupt without a HooksRuntime"
+        raise RuntimeError(msg)
+    pending = dict(state.pending_hook_interrupts)
+    state.pending_hook_interrupts.clear()
+    try:
+        state.hook_response.update(
+            await fulfill_pending_hook_interrupts(state.hooks_runtime, pending)
+        )
+    except ValueError as exc:
+        msg = f"Hook resume could not be applied: {exc}"
+        raise HooksSnapshotChangedError(msg) from exc
+
+
 def _process_hitl_interrupts(state: StreamState, console: Console) -> None:
     """Iterate over pending HITL interrupts and build approval/rejection responses.
 
@@ -1045,6 +1093,7 @@ async def _run_agent_loop(
     max_turns: int | None = None,
     rubric: str | None = None,
     show_rubric_iterations: bool = False,
+    trust_project_hooks: bool = False,
 ) -> None:
     """Run the agent and handle HITL interrupts until the task completes.
 
@@ -1077,6 +1126,11 @@ async def _run_agent_loop(
             `None` leaves it unset (no grading).
         show_rubric_iterations: Whether rubric lifecycle messages should include
             iteration numbers.
+        trust_project_hooks: When `True`, load project-scoped
+            `.deepagents/hooks.json` command handlers.
+
+            Defaults to `False` so untrusted checkouts cannot execute repository
+            hooks in CI without an explicit opt-in.
 
     Raises:
         HITLIterationLimitError: If the effective turn limit is exceeded.
@@ -1103,6 +1157,24 @@ async def _run_agent_loop(
     # unset in context rather than passing a blank string to model middleware.
     context_thread_id = thread_id if isinstance(thread_id, str) and thread_id else None
     context = CLIContext(thread_id=context_thread_id)
+
+    from pathlib import Path
+
+    from deepagents_code.hooks.context import apply_hooks_context
+    from deepagents_code.hooks.runtime import HooksRuntime
+
+    try:
+        # Project hooks require an explicit opt-in, matching `--trust-project-mcp`.
+        hooks_runtime = HooksRuntime.create(
+            cwd=Path.cwd(),
+            workspace_trusted=trust_project_hooks,
+        )
+    except Exception:
+        logger.exception("Failed to create HooksRuntime; server hooks disabled")
+        hooks_runtime = None
+    apply_hooks_context(context, hooks_runtime)
+    state.hooks_runtime = hooks_runtime
+
     await dispatch_hook("session.start", {"thread_id": thread_id})
 
     start_time = time.monotonic()
@@ -1137,8 +1209,11 @@ async def _run_agent_loop(
             turns += 1
             state.interrupt_occurred = False
             state.hitl_response.clear()
+            state.hook_response.clear()
+            await _fulfill_pending_hook_interrupts(state)
             _process_hitl_interrupts(state, console)
-            stream_input = Command(resume=state.hitl_response)
+            resume_payload = {**state.hook_response, **state.hitl_response}
+            stream_input = Command(resume=resume_payload)
             await _stream_agent(
                 agent, stream_input, config, state, console, file_op_tracker, context
             )
@@ -1366,6 +1441,7 @@ async def run_non_interactive(
     rubric_model: str | None = None,
     rubric_max_iterations: int | None = None,
     recursion_limit: int | None = None,
+    trust_project_hooks: bool = False,
 ) -> int:
     """Run a single task non-interactively and exit.
 
@@ -1442,6 +1518,11 @@ async def run_non_interactive(
             uses the middleware default.
         recursion_limit: Explicit main-agent `recursion_limit`; `None` resolves
             from env / `config.toml` / default at agent-build time.
+        trust_project_hooks: When `True`, load project-scoped
+            `.deepagents/hooks.json` handlers.
+
+            Defaults to `False` so untrusted repositories cannot execute hook
+            commands without an explicit `--trust-project-hooks` opt-in.
 
     Returns:
         Exit code: 0 for success, 1 for error, 124 when the `--max-turns`
@@ -1692,6 +1773,7 @@ def discover_all_skills() -> tuple[list[ExtendedSkillMetadata], list[Path]]:
                 max_turns=max_turns,
                 rubric=rubric,
                 show_rubric_iterations=rubric_max_iterations is not None,
+                trust_project_hooks=trust_project_hooks,
             )
 
     except KeyboardInterrupt: