fix(code): keep server hook state out of task results

libs/code/deepagents_code/agent.py

@@ -1929,12 +1929,11 @@ 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
+    from deepagents_code.hooks.server_middleware import hook_decided_permission
 
     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"}:
+    if hook_decided_permission(getattr(request, "state", None), tool_call_id):
         return False
 
     runtime = getattr(request, "runtime", None)

libs/code/deepagents_code/auto_mode.py

@@ -2733,12 +2733,12 @@ 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
+        from deepagents_code.hooks.server_middleware import hook_decided_permission
 
         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"}
+            if hook_decided_permission(state, _tool_call_id(call))
         }
         thread_key = _thread_key(runtime)
         # Derive the emission scope once for the whole node run. Deriving it

libs/code/deepagents_code/hooks/server_middleware.py

@@ -16,6 +16,7 @@
 from datetime import UTC, datetime, timedelta
 from typing import (
     TYPE_CHECKING,
+    Annotated,
     Any,
     Literal,
     NotRequired,
@@ -35,6 +36,7 @@
     AgentMiddleware,
     AgentState,
     ContextT,
+    PrivateStateAttr,
     ResponseT,
     hook_config,
 )
@@ -93,19 +95,60 @@
 _INVOCATION_NAMESPACE = UUID("f2896d18-cf2a-4e7d-b11a-d5b10fc0e335")
 
 PreToolBehavior: TypeAlias = Literal["allow", "deny", "none"]
+_DEFAULT_DENY_REASON = "Blocked by PreToolUse hook"
 
 
-class _PreToolState(TypedDict):
-    behavior: PreToolBehavior
-    reason: str | None
+class _PreToolDenied(TypedDict):
+    """Outcome for a call a hook refused. A denial always carries a reason."""
+
+    behavior: Literal["deny"]
+    reason: str
+    context: list[str]
+
+
+class _PreToolPassed(TypedDict):
+    """Outcome for a call a hook allowed or had no opinion on."""
+
+    behavior: Literal["allow", "none"]
     context: list[str]
 
 
+_PreToolState: TypeAlias = _PreToolDenied | _PreToolPassed
+
+
 class ServerHooksState(AgentState[Any]):
-    """Agent state extensions for server-owned hook middleware."""
+    """Agent state extensions for server-owned hook middleware.
+
+    Both fields are per-turn bookkeeping owned by `ServerHooksMiddleware` and
+    marked `PrivateStateAttr`: they are omitted from the public graph I/O schema,
+    and `SubAgentMiddleware` strips them from subagent result merges so parallel
+    `task` calls cannot produce two concurrent writes to these `LastValue`
+    channels.
+
+    `PrivateStateAttr` only omits the fields from the input and output schemas;
+    the channels stay ordinary checkpointed `LastValue` channels visible to every
+    node, so values still flow from `after_model` to `wrap_tool_call` and survive
+    interrupt/resume.
+
+    Note:
+        If either field ever needs a reducer, the reducer must be placed *after*
+        `PrivateStateAttr` in the `Annotated` metadata. LangGraph only inspects
+        the last metadata entry when detecting reducers, so a reducer added
+        before the marker is silently ignored.
+    """
 
-    _hooks_stop_continuation_count: NotRequired[int]
-    _hooks_pre_tool_outcomes: NotRequired[dict[str, _PreToolState]]
+    _hooks_stop_continuation_count: NotRequired[Annotated[int, PrivateStateAttr]]
+    """Stop-hook continuations in the current turn; reset to 0 when the loop ends."""
+
+    _hooks_pre_tool_outcomes: NotRequired[
+        Annotated[dict[str, _PreToolState], PrivateStateAttr]
+    ]
+    """Pre-execution hook verdicts keyed by tool-call id.
+
+    A full snapshot of the *current* turn's calls, not an accumulator: every
+    `_after_model` replaces the whole dict (including with `{}`) so stale ids
+    cannot survive into a later turn.
+    """
 
 
 class _SessionHookGate(TypedDict):
@@ -392,7 +435,7 @@ def _after_model(
                     reason = (
                         permission.reason
                         or decision.stop_reason
-                        or "Blocked by PreToolUse hook"
+                        or _DEFAULT_DENY_REASON
                     )
                 elif permission.behavior == "ask":
                     blocked = _ask_permission_via_hitl(call, permission)
@@ -408,11 +451,19 @@ def _after_model(
                         )
                 elif permission.behavior == "allow":
                     behavior = "allow"
-            outcomes[call.id] = {
-                "behavior": behavior,
-                "reason": reason,
-                "context": hook_context,
-            }
+            if behavior == "deny":
+                outcomes[call.id] = {
+                    "behavior": "deny",
+                    # Every deny path above resolves a reason; the guard keeps the
+                    # "a denial always explains itself" invariant checkable here.
+                    "reason": reason if reason is not None else _DEFAULT_DENY_REASON,
+                    "context": hook_context,
+                }
+            else:
+                outcomes[call.id] = {
+                    "behavior": behavior,
+                    "context": hook_context,
+                }
         return {_PRE_TOOL_STATE_KEY: outcomes}
 
     def _maybe_post_tool_use(
@@ -541,19 +592,23 @@ def _event_enabled(gate: _SessionHookGate | None, event: HookEvent) -> bool:
     return gate is not None and event.value in gate["events"]
 
 
-def pre_tool_behavior(state: object, tool_call_id: str) -> PreToolBehavior | None:
-    """Return the replayed pre-execution hook behavior for one call."""
+def hook_decided_permission(state: object, tool_call_id: str) -> bool:
+    """Report whether a pre-execution hook already settled permission for a call.
+
+    Args:
+        state: Agent state carrying the current turn's hook outcomes.
+        tool_call_id: Tool call to look up.
+
+    Returns:
+        `True` when a hook explicitly allowed or denied the call, so stock
+        approval flows must not prompt again. `False` when no hook ran, the hook
+        expressed no opinion, or no outcome was recorded -- in every one of those
+        cases normal approval still applies.
+    """
     outcome = _pre_tool_state(state, tool_call_id)
     if outcome is None:
-        return None
-    behavior = outcome.get("behavior")
-    if behavior == "allow":
-        return "allow"
-    if behavior == "deny":
-        return "deny"
-    if behavior == "none":
-        return "none"
-    return None
+        return False
+    return outcome.get("behavior") in {"allow", "deny"}
 
 
 def _pre_tool_state(state: object, tool_call_id: str) -> Mapping[str, object] | None: