From 4cc6eea28df78ef37a2d9b297be6ab00cfbdf637 Mon Sep 17 00:00:00 2001
From: "Michael J. Sullivan" <sully@msully.net>
Date: Fri, 17 Jul 2026 11:40:11 +0200
Subject: [PATCH] Add PendingHookEvent/ResolvedHookEvent/CancelledHookEvent

This makes matching on them much nicer looking.
---
 .../content/docs/basics/durable-execution.mdx |  2 +-
 .../content/docs/basics/human-in-the-loop.mdx | 14 +--
 .../docs/basics/messages-and-events.mdx       |  2 +-
 .../content/docs/reference/events.mdx         |  9 +-
 examples/agents/custom_hook.py                | 10 +-
 examples/agents/tool_approval.py              | 10 +-
 examples/apps/coding_agent/main.py            | 19 ++--
 examples/apps/web_agent/README.md             |  2 +-
 examples/apps/web_agent/backend/main.py       |  5 +-
 pyproject.toml                                |  1 +
 src/ai/agents/agent.py                        | 15 ++-
 src/ai/agents/runtime.py                      |  4 +-
 src/ai/types/events.py                        | 99 ++++++++++++++++---
 tests/agents/test_run_blocked.py              |  2 +-
 tests/types/test_events.py                    | 19 +++-
 tests/ui/ai_sdk/test_outbound_stream.py       |  6 +-
 16 files changed, 145 insertions(+), 74 deletions(-)

diff --git a/docs/ai-python/content/docs/basics/durable-execution.mdx b/docs/ai-python/content/docs/basics/durable-execution.mdx
index 74a03381..7735c055 100644
--- a/docs/ai-python/content/docs/basics/durable-execution.mdx
+++ b/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
diff --git a/docs/ai-python/content/docs/basics/human-in-the-loop.mdx b/docs/ai-python/content/docs/basics/human-in-the-loop.mdx
index 3352ee4e..b0e118dd 100644
--- a/docs/ai-python/content/docs/basics/human-in-the-loop.mdx
+++ b/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)
 
diff --git a/docs/ai-python/content/docs/basics/messages-and-events.mdx b/docs/ai-python/content/docs/basics/messages-and-events.mdx
index 9d72a197..d13594ce 100644
--- a/docs/ai-python/content/docs/basics/messages-and-events.mdx
+++ b/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
diff --git a/docs/ai-python/content/docs/reference/events.mdx b/docs/ai-python/content/docs/reference/events.mdx
index 871d3edd..a4c50ce3 100644
--- a/docs/ai-python/content/docs/reference/events.mdx
+++ b/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
 
diff --git a/examples/agents/custom_hook.py b/examples/agents/custom_hook.py
index 6e5baf3d..87eb1090 100644
--- a/examples/agents/custom_hook.py
+++ b/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()
 
diff --git a/examples/agents/tool_approval.py b/examples/agents/tool_approval.py
index 60278f33..31d40518 100644
--- a/examples/agents/tool_approval.py
+++ b/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)
diff --git a/examples/apps/coding_agent/main.py b/examples/apps/coding_agent/main.py
index df7a47e9..0ef77cad 100644
--- a/examples/apps/coding_agent/main.py
+++ b/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(
diff --git a/examples/apps/web_agent/README.md b/examples/apps/web_agent/README.md
index 9eab4aa6..af139b83 100644
--- a/examples/apps/web_agent/README.md
+++ b/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
diff --git a/examples/apps/web_agent/backend/main.py b/examples/apps/web_agent/backend/main.py
index 42ca0edd..94423f17 100644
--- a/examples/apps/web_agent/backend/main.py
+++ b/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
 
diff --git a/pyproject.toml b/pyproject.toml
index 02ba10fe..292586b2 100644
--- a/pyproject.toml
+++ b/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]
diff --git a/src/ai/agents/agent.py b/src/ai/agents/agent.py
index 92bfb271..81795cd6 100644
--- a/src/ai/agents/agent.py
+++ b/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:
diff --git a/src/ai/agents/runtime.py b/src/ai/agents/runtime.py
index 0e1cbca2..5cfb8211 100644
--- a/src/ai/agents/runtime.py
+++ b/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()
diff --git a/src/ai/types/events.py b/src/ai/types/events.py
index 56a36a1d..07ff0eea 100644
--- a/src/ai/types/events.py
+++ b/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
 
diff --git a/tests/agents/test_run_blocked.py b/tests/agents/test_run_blocked.py
index c28dbf88..c6a3372a 100644
--- a/tests/agents/test_run_blocked.py
+++ b/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
     )
 
diff --git a/tests/types/test_events.py b/tests/types/test_events.py
index f746c8e2..657e66a0 100644
--- a/tests/types/test_events.py
+++ b/tests/types/test_events.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 import pydantic
 import pytest
 
@@ -220,7 +222,7 @@ def test_message_hydrator_tracks_multiple_messages() -> None:
         events.TextDelta(message=assistant_1, block_id="text-1", chunk="first"),
         events.StreamEnd(message=assistant_1),
         events.ToolCallResult(message=tool_message, results=[]),
-        events.HookEvent(
+        events.PendingHookEvent(
             message=hook_message,
             hook=messages.HookPart(
                 hook_id="hook", hook_type="test", status="pending"
@@ -265,7 +267,7 @@ def test_non_model_events_keep_message() -> None:
     )
     source: list[events.AgentEvent] = [
         events.ToolCallResult(message=tool_message, results=[]),
-        events.HookEvent(
+        events.PendingHookEvent(
             message=hook_message,
             hook=messages.HookPart(
                 hook_id="hook", hook_type="test", status="pending"
@@ -297,6 +299,19 @@ def test_omitted_model_event_validates_with_dummy_message() -> None:
     assert restored.message.parts == []
 
 
+@pytest.mark.parametrize("dump", [False, True])
+def test_hook_event_validation_dispatches_on_status(dump: bool) -> None:
+    message = messages.Message(id="hook-message", role="internal", parts=[])
+    hook: messages.HookPart[Any] = messages.HookPart(
+        hook_id="hook", hook_type="test", status="resolved"
+    )
+    data = {"message": message, "hook": hook.model_dump() if dump else hook}
+
+    restored = events.HookEvent.model_validate(data)
+
+    assert isinstance(restored, events.ResolvedHookEvent)
+
+
 def test_retry_round_trips_through_agent_event_union() -> None:
     adapter: pydantic.TypeAdapter[events.AgentEvent] = pydantic.TypeAdapter(
         events.AgentEvent
diff --git a/tests/ui/ai_sdk/test_outbound_stream.py b/tests/ui/ai_sdk/test_outbound_stream.py
index fdf272cf..97ac482c 100644
--- a/tests/ui/ai_sdk/test_outbound_stream.py
+++ b/tests/ui/ai_sdk/test_outbound_stream.py
@@ -206,7 +206,7 @@ async def test_finish_metadata_tracks_tool_and_internal_messages() -> None:
                 message=tool,
                 results=tool.tool_results,
             ),
-            agent_events_.HookEvent(message=internal, hook=hook),
+            agent_events_.HookEvent.for_hook(message=internal, hook=hook),
         ]
     )
 
@@ -435,7 +435,7 @@ async def test_approval_request_hook_emits_approval_event() -> None:
                 ),
             ),
             # Hook requesting approval
-            agent_events_.HookEvent(
+            agent_events_.HookEvent.for_hook(
                 message=messages_.Message(
                     role="internal",
                     parts=[
@@ -638,7 +638,7 @@ async def test_resolved_approval_hook_emits_response_event() -> None:
 
     out = await _collect(
         [
-            agent_events_.HookEvent(
+            agent_events_.HookEvent.for_hook(
                 message=messages_.Message(
                     id="turn-1:internal:0",
                     turn_id="turn-1",