Add PendingHookEvent/ResolvedHookEvent/CancelledHookEvent

docs/ai-python/content/docs/basics/durable-execution.mdx

@@ -116,7 +116,7 @@ assistant message before tools are scheduled. In a streaming loop, `merge`
 interleaves model events and tool results while the model is still producing
 output.
 
-Consider using the serverless hook flow for approvals. When a deferred `HookEvent`
+Consider using the serverless hook flow for approvals. When a `PendingHookEvent`
 appears, call `ai.defer_hook(event.hook)`, persist `stream.messages`,
 and return the approval request to the client. On the next workflow turn, call
 `ai.resolve_hook(...)` inside the `agent.run(...)` block before iterating the

docs/ai-python/content/docs/basics/human-in-the-loop.mdx

@@ -20,16 +20,13 @@ async def notify_mothership(message: str) -> str:
     return f"Sent: {message}"
 ```
 
-When the model calls the tool, the agent emits a hook event. Resolve it with
-`ai.resolve_hook`:
+When the model calls the tool, the agent emits a `PendingHookEvent`. Resolve
+it with `ai.resolve_hook`:
 
 ```python
 async with agent.run(model, messages) as stream:
     async for event in stream:
-        if (
-            isinstance(event, ai.events.HookEvent)
-            and event.hook.status == "pending"
-        ):
+        if isinstance(event, ai.events.PendingHookEvent):
             print(event.hook.hook_id, event.hook.metadata)
             ai.resolve_hook(
                 event.hook,
@@ -138,10 +135,7 @@ async with agent.run(model, messages) as stream:
         )
 
     async for event in stream:
-        if (
-            isinstance(event, ai.events.HookEvent)
-            and event.hook.status == "pending"
-        ):
+        if isinstance(event, ai.events.PendingHookEvent):
             # interrupt the loop
             ai.defer_hook(event.hook)
 

docs/ai-python/content/docs/basics/messages-and-events.mdx

@@ -135,7 +135,7 @@ StreamEnd
 # Agent runs can also interleave:
 ToolCallResult          # emitted after a Python tool finishes
 PartialToolCallResult   # emitted while a streaming tool or subagent yields
-HookEvent               # emitted when a hook is deferred, resolved, or cancelled
+HookEvent               # PendingHookEvent / ResolvedHookEvent / CancelledHookEvent
 ```
 
 ```python

docs/ai-python/content/docs/reference/events.mdx

@@ -65,7 +65,10 @@ Agents can emit event values that are not raw provider stream events.
 - `ToolCallResult`: Carries the tool result message and result parts.
 - `PartialToolCallResult`: Carries values yielded by streaming tools and
   `yield_from`.
-- `HookEvent`: Carries an internal hook message and `HookPart`.
+- `HookEvent`: Carries an internal hook message and `HookPart`. Every
+  instance is one of the per-status subclasses — `PendingHookEvent`,
+  `ResolvedHookEvent`, or `CancelledHookEvent` — so match on the subclass
+  instead of checking `hook.status`.
 - `RunBlocked`: Emitted when the run is blocked on deferred hooks.
 - `Retry`: Discards everything since the most recent `StreamStart`: the
   in-progress assistant message and any tool results for it. The
@@ -79,7 +82,9 @@ Agent event union:
   and `Retry`.
   The union is annotated with a Pydantic discriminator on the `kind` field,
   so use it as a field or `TypeAdapter` type when deserializing events and
-  Pydantic picks the right event class from `kind`.
+  Pydantic picks the right event class from `kind`. Hook events all share
+  `kind: "hook"`; validating against `HookEvent` dispatches to the
+  per-status subclass from `hook.status`.
 
 ## Aggregator
 

examples/agents/custom_hook.py

@@ -44,10 +44,7 @@ async def main() -> None:
         async for event in stream:
             if isinstance(event, ai.events.TextDelta):
                 print(event.chunk, end="", flush=True)
-            elif (
-                isinstance(event, ai.events.HookEvent)
-                and event.hook.status == "pending"
-            ):
+            elif isinstance(event, ai.events.PendingHookEvent):
                 print(f"\n[deferred] {event.hook.hook_id}")
                 ai.resolve_hook(
                     event.hook,
@@ -57,10 +54,7 @@ async def main() -> None:
                         reason="change window is open",
                     ),
                 )
-            elif (
-                isinstance(event, ai.events.HookEvent)
-                and event.hook.status == "resolved"
-            ):
+            elif isinstance(event, ai.events.ResolvedHookEvent):
                 print(f"[resolved] {event.hook.hook_id}")
     print()
 

examples/agents/tool_approval.py

@@ -26,10 +26,7 @@ async def live(messages: list[ai.messages.Message]) -> None:
         async for event in stream:
             if isinstance(event, ai.events.TextDelta):
                 print(event.chunk, end="", flush=True)
-            elif (
-                isinstance(event, ai.events.HookEvent)
-                and event.hook.status == "pending"
-            ):
+            elif isinstance(event, ai.events.PendingHookEvent):
                 print(f"\n[deferred] {event.hook.hook_id}")
                 ai.resolve_hook(
                     event.hook,
@@ -57,10 +54,7 @@ async def stateless(
             if isinstance(event, ai.events.TextDelta):
                 text.append(event.chunk)
                 print(event.chunk, end="", flush=True)
-            elif (
-                isinstance(event, ai.events.HookEvent)
-                and event.hook.status == "pending"
-            ):
+            elif isinstance(event, ai.events.PendingHookEvent):
                 deferred.append(event.hook)
                 print(f"\n[deferred] {event.hook.hook_id}")
                 ai.defer_hook(event.hook)

examples/apps/coding_agent/main.py

@@ -402,8 +402,10 @@ async def _run_turn(self) -> None:
                                         part.result, is_error=part.is_error
                                     ),
                                 )
+                    elif isinstance(event, ai.events.PendingHookEvent):
+                        self._mount_prompt(event.hook)
                     elif isinstance(event, ai.events.HookEvent):
-                        self._on_hook(event.hook)
+                        self._drop_prompt(event.hook.hook_id)
             finally:
                 # ``stream.messages`` is always a clean prefix of
                 # completed rounds — keep it even on interrupt/error so
@@ -413,15 +415,12 @@ async def _run_turn(self) -> None:
 
     # -- approval hooks ------------------------------------------------------
 
-    def _on_hook(self, hook: ai.messages.HookPart[Any]) -> None:
-        """Mount or dismiss an approval prompt for a hook signal."""
-        if hook.status == "pending":
-            prompt = ApprovalPrompt(hook)
-            self.query_one("#dock").mount(prompt, before=self.composer)
-            if not isinstance(self.focused, ApprovalPrompt):
-                prompt.focus()
-        else:
-            self._drop_prompt(hook.hook_id)
+    def _mount_prompt(self, hook: ai.messages.HookPart[Any]) -> None:
+        """Mount an approval prompt for a pending hook."""
+        prompt = ApprovalPrompt(hook)
+        self.query_one("#dock").mount(prompt, before=self.composer)
+        if not isinstance(self.focused, ApprovalPrompt):
+            prompt.focus()
 
     def on_approval_prompt_decided(self, event: ApprovalPrompt.Decided) -> None:
         ai.resolve_hook(

examples/apps/web_agent/README.md

@@ -16,7 +16,7 @@ tool is gated behind user confirmation before execution.
 `ToolApproval` hook before executing it. The flow is:
 
 1. LLM emits a call to the gated tool
-2. The runtime emits a `HookEvent` with a deferred `HookPart`; the
+2. The runtime emits a `PendingHookEvent` with a deferred `HookPart`; the
    backend aborts the deferred hook (`ai.defer_hook`) so the turn
    ends and the deferred approval streams to the client
 3. The frontend renders Approve / Reject buttons via the

examples/apps/web_agent/backend/main.py

@@ -74,10 +74,7 @@ async def stream_response() -> AsyncGenerator[str]:
             # async generator that does this check and yields the events.
             async def process() -> AsyncGenerator[ai.events.AgentEvent]:
                 async for event in result:
-                    if (
-                        isinstance(event, ai.events.HookEvent)
-                        and event.hook.status == "pending"
-                    ):
+                    if isinstance(event, ai.events.PendingHookEvent):
                         ai.defer_hook(event.hook)
                     yield event
 

pyproject.toml

@@ -168,6 +168,7 @@ extend-ignore = [
     "PLW1514", # unspecified-encoding (UTF-8 is Python 3 default)
     "PLW2901", # redefined-loop-name (intentional pattern for line processing)
     "RUF029",  # async generators/protocol hooks may not need an await
+    "TC003",   # annoying and wrong if TYPE_CHECKING for stdlib imports
 ]
 
 [tool.ruff.lint.per-file-ignores]

src/ai/agents/agent.py

@@ -1139,15 +1139,14 @@ def _track(self, event: events_.AgentEvent) -> events_.AgentEvent:
         # consumer has seen so far.
         if isinstance(event, events_.RunBlocked):
             self._blocked = True
+        elif isinstance(event, events_.PendingHookEvent):
+            self._deferred_hooks[event.hook.hook_id] = event.hook
         elif isinstance(event, events_.HookEvent):
-            if event.hook.status == "pending":
-                self._deferred_hooks[event.hook.hook_id] = event.hook
-            else:
-                # A blocked run can only resume via a hook resolution
-                # or cancellation (see RunBlocked), so this is also the
-                # unblock signal.
-                self._deferred_hooks.pop(event.hook.hook_id, None)
-                self._blocked = False
+            # A blocked run can only resume via a hook resolution
+            # or cancellation (see RunBlocked), so this is also the
+            # unblock signal.
+            self._deferred_hooks.pop(event.hook.hook_id, None)
+            self._blocked = False
         return event
 
     def __aiter__(self) -> Self:

src/ai/agents/runtime.py

@@ -33,7 +33,9 @@ async def put_event(self, event: events_.AgentEvent) -> None:
 
     async def put_hook(self, hook_part: messages_.HookPart[Any]) -> None:
         msg = messages_.Message(role="internal", parts=[hook_part])
-        await self.put_event(events_.HookEvent(message=msg, hook=hook_part))
+        await self.put_event(
+            events_.HookEvent.for_hook(message=msg, hook=hook_part)
+        )
 
     async def signal_done(self) -> None:
         await self._event_queue.astop()

src/ai/types/events.py

@@ -1,7 +1,9 @@
+from __future__ import annotations
+
 import abc
 import contextvars
-from collections.abc import AsyncGenerator, Callable, Sequence
-from typing import Annotated, Any, Literal
+from collections.abc import AsyncGenerator, Callable, Mapping, Sequence
+from typing import Annotated, Any, Literal, cast
 
 import pydantic
 
@@ -608,14 +610,83 @@ class ToolCallResult(BaseEvent):
     kind: Literal["tool_call_result"] = "tool_call_result"
 
 
+def _get(obj: Any, key: Any) -> Any | None:
+    return obj.get(key) if isinstance(obj, dict) else getattr(obj, key, None)
+
+
 class HookEvent(BaseEvent):
-    """Emitted when a hook suspends, resolves, or is cancelled."""
+    """Emitted when a hook suspends, resolves, or is cancelled.
+
+    Every instance is one of the per-status subclasses —
+    :class:`PendingHookEvent`, :class:`ResolvedHookEvent`, or
+    :class:`CancelledHookEvent` — matching ``hook.status``.  Construct
+    via :meth:`for_hook` (or a subclass directly); the base class
+    cannot be instantiated, and validating against it dispatches to
+    the matching subclass.
+    """
 
     message: messages.Message
     hook: messages.HookPart[Any]
 
     kind: Literal["hook"] = "hook"
 
+    def __init__(self, /, **data: Any) -> None:
+        if type(self) is HookEvent:
+            raise TypeError(
+                "HookEvent cannot be instantiated directly; use "
+                "HookEvent.for_hook() or a per-status subclass"
+            )
+        super().__init__(**data)
+
+    @classmethod
+    def for_hook(
+        cls, message: messages.Message, hook: messages.HookPart[Any]
+    ) -> HookEvent:
+        """Construct the per-status subclass matching ``hook.status``."""
+        return _HOOK_EVENTS[hook.status](message=message, hook=hook)
+
+    @pydantic.model_validator(mode="wrap")
+    @classmethod
+    def _dispatch_status(
+        cls, value: Any, handler: pydantic.ValidatorFunctionWrapHandler
+    ) -> HookEvent:
+        if cls is not HookEvent or not isinstance(value, Mapping):
+            return cast("HookEvent", handler(value))
+        hook = _get(value, "hook")
+        status = _get(hook, "status") if hook is not None else None
+        # On an invalid status, we validate against PendingHookEvent, but
+        # pydantic will report the missing fields.
+        sub = _HOOK_EVENTS.get(status, PendingHookEvent)
+        return sub.model_validate(value)
+
+    @pydantic.model_validator(mode="after")
+    def _check_status(self) -> HookEvent:
+        if not isinstance(self, _HOOK_EVENTS[self.hook.status]):
+            raise ValueError(
+                f"{type(self).__name__} cannot carry a "
+                f"{self.hook.status!r} hook"
+            )
+        return self
+
+
+class PendingHookEvent(HookEvent):
+    """The hook suspended; the run needs its resolution to proceed."""
+
+
+class ResolvedHookEvent(HookEvent):
+    """The hook resolved; ``hook.resolution`` carries the value."""
+
+
+class CancelledHookEvent(HookEvent):
+    """The hook was cancelled without a resolution."""
+
+
+_HOOK_EVENTS: dict[object, type[HookEvent]] = {
+    "pending": PendingHookEvent,
+    "resolved": ResolvedHookEvent,
+    "cancelled": CancelledHookEvent,
+}
+
 
 class RunBlocked(BaseEvent):
     """The run is blocked on hooks.
@@ -631,11 +702,11 @@ class RunBlocked(BaseEvent):
 
     There is no mirror "unblocked" event because it would be redundant:
     a blocked run can only resume via a hook resolution (or
-    cancellation), so the next ``HookEvent`` with a non-``pending``
-    status *is* the unblock signal.  Note the converse does not hold —
-    a ``ToolCallResult`` carrying an ``is_hook_deferred`` placeholder
-    (serverless abort) arrives while the run stays blocked, and the run
-    then ends still blocked.
+    cancellation), so the next :class:`ResolvedHookEvent` or
+    :class:`CancelledHookEvent` *is* the unblock signal.  Note the
+    converse does not hold — a ``ToolCallResult`` carrying an
+    ``is_hook_deferred`` placeholder (serverless abort) arrives while
+    the run stays blocked, and the run then ends still blocked.
     """
 
     hooks: tuple[messages.HookPart[Any], ...] = ()
@@ -681,8 +752,9 @@ class RunStateTracker:
 
     The fold reads three things:
 
-    * hook state from :class:`HookEvent` (``pending`` adds, ``resolved``
-      / ``cancelled`` removes);
+    * hook state from :class:`HookEvent` (:class:`PendingHookEvent`
+      adds, :class:`ResolvedHookEvent` / :class:`CancelledHookEvent`
+      remove);
     * model-stream activity from :class:`StreamStart` / :class:`StreamEnd`;
     * in-flight tool calls from the assistant message on
       :class:`StreamEnd` (scheduled) and :class:`ToolCallResult`
@@ -743,11 +815,10 @@ def feed(self, event: AgentEvent) -> RunBlocked | None:
                 self._in_flight.difference_update(
                     r.tool_call_id for r in event.results
                 )
+            case PendingHookEvent():
+                self._deferred[event.hook.hook_id] = event.hook
             case HookEvent():
-                if event.hook.status == "pending":
-                    self._deferred[event.hook.hook_id] = event.hook
-                else:
-                    self._deferred.pop(event.hook.hook_id, None)
+                self._deferred.pop(event.hook.hook_id, None)
             case _:
                 return None
 

tests/agents/test_run_blocked.py

@@ -328,7 +328,7 @@ async def self_gated(x: int) -> str:
 
 
 def _hook_event(hook: messages_.HookPart[Any]) -> events_.HookEvent:
-    return events_.HookEvent(
+    return events_.HookEvent.for_hook(
         message=messages_.Message(role="internal", parts=[hook]), hook=hook
     )