fix(chat,sdk): stop an unconsumed control record wedging the mailbox

`chat.messages.next()` and `hasPending()` only inspect the head of the
`.in` buffer. A control record whose kind has no consumer on this boot
therefore parked at the head forever, and every message queued behind it
became undeliverable with no error: `hasPending()` stayed false and
`next()` timed out on every call.

Records of a kind nothing on the run consumes are now discarded at
dispatch instead. Consuming at dispatch keeps the resume cursor exact,
since the record never enters the buffer and so leaves no unconsumed
barrier for `lastDispatchedSeqNum()` to clamp behind. `message` is always
claimed, and handover kinds are claimed for the window in which a
handover-prepare boot is actually waiting for them.

The mixed-kinds test now builds its blocked-head state on a
handover-prepare boot, where the handover kind is claimed and so stays
buffered for the raw read it asserts.

Author: ericallam

.changeset/tidy-mailboxes-wait.md

@@ -4,3 +4,5 @@
 ---
 
 Custom agent loops can now inspect pending chat input without consuming it and consume one mailbox record at a time with `chat.messages.hasPending()` and `chat.messages.next()`. Mailbox records include stable identifiers for tracing and redelivery.
+
+A control record that nothing on the run consumes is now discarded rather than left at the head of the `.in` channel, where it would have made every message queued behind it undeliverable. `chat.messages.next()` returning `undefined` now always means the mailbox is idle.

docs/ai-chat/custom-agents.mdx

@@ -267,6 +267,11 @@ comes before a message, `hasPending()` stays `false` and `next()` leaves the
 control record for its own consumer. After that record is handled, the message
 becomes pending.
 
+A control record that nothing on the run consumes is discarded rather than left
+at the head of the channel. `hasPending()` and `next()` only look at the head, so
+a record parked there would make every message behind it undeliverable. This
+means `next()` returning `undefined` always means the mailbox is idle.
+
 A complete loop:
 
 ```ts trigger/my-chat-raw.ts

packages/trigger-sdk/src/v3/ai.ts

@@ -1885,17 +1885,118 @@ async function waitForHandover(options: {
   spanName?: string;
 }): Promise<HandoverSignal | null> {
   if (options.payload.trigger !== "handover-prepare") return null;
-  const result = await handoverInput.waitWithIdleTimeout({
-    idleTimeoutInSeconds:
-      options.idleTimeoutInSeconds ?? options.payload.idleTimeoutInSeconds ?? 60,
-    timeout: options.timeout,
-    spanName: options.spanName ?? "waiting for handover signal",
+  try {
+    const result = await handoverInput.waitWithIdleTimeout({
+      idleTimeoutInSeconds:
+        options.idleTimeoutInSeconds ?? options.payload.idleTimeoutInSeconds ?? 60,
+      timeout: options.timeout,
+      spanName: options.spanName ?? "waiting for handover signal",
+    });
+    // Non-ok = idle timeout or the warm handler crashed without signaling.
+    if (!result.ok) return null;
+    return result.output;
+  } finally {
+    // The handover window is over either way. A signal arriving after this
+    // point has no consumer, so hand it to the drain rather than letting it
+    // park at the head of the channel.
+    releaseChatInputKinds(CHAT_HANDOVER_KINDS);
+  }
+}
+
+/**
+ * Record kinds on `session.in` that some consumer on THIS boot is responsible
+ * for. `"message"` is always claimed; handover kinds are claimed only for the
+ * window in which `waitForHandover` is actually waiting for them.
+ *
+ * Anything not in this set has no consumer on this boot, so leaving it buffered
+ * would park it at the head of the channel forever — `chat.messages.next()` and
+ * `hasPending()` only ever inspect the head, so every record queued behind it
+ * becomes undeliverable with no error. The drain below discards unclaimed kinds
+ * instead.
+ * @internal
+ */
+const chatClaimedKindsKey = locals.create<Set<string>>("chat.claimedKinds");
+
+/** Kinds carried on `.in` that are not user messages. @internal */
+const CHAT_HANDOVER_KINDS = ["handover", "handover-skip"] as const;
+
+/** The run's attached drain subscription, so it can be re-offered the buffer. @internal */
+const chatInputDrainKey = locals.create<{ off: () => void }>("chat.inputDrain");
+
+function chatClaimedKinds(): Set<string> {
+  let claimed = locals.get(chatClaimedKindsKey);
+  if (!claimed) {
+    claimed = new Set<string>(["message"]);
+    locals.set(chatClaimedKindsKey, claimed);
+  }
+  return claimed;
+}
+
+/**
+ * Attach the unclaimed-control drain for this run.
+ *
+ * Consuming at dispatch (returning `true`) is what makes this safe for the
+ * resume cursor: the record is never buffered, so it leaves no unconsumed
+ * marker and `lastDispatchedSeqNum()` stays exact rather than being clamped
+ * behind a record nobody will ever take.
+ *
+ * `#dispatch` resolves a matching `once()` waiter BEFORE invoking handlers, so
+ * this can never take a record out from under a claimed consumer that is
+ * actively waiting for it.
+ *
+ * MUST be attached after `seedSessionInResumeCursorForCustomLoop`, like every
+ * other `.in` listener — attaching first would replay from seq 0.
+ * @internal
+ */
+function attachUnclaimedChatInputDrain(): { off: () => void } {
+  return getChatSession().in.on<ChatInputChunk>((chunk) => {
+    const kind = (chunk as { kind?: unknown } | undefined)?.kind;
+    // Malformed record: nothing can consume it, so don't let it wedge the head.
+    if (typeof kind !== "string") {
+      logger.warn("chat: discarded a malformed session.in record with no usable kind");
+      return true;
+    }
+    if (chatClaimedKinds().has(kind)) return undefined;
+    logger.warn("chat: discarded a session.in record that no consumer handled on this boot", {
+      kind,
+    });
+    return true;
   });
-  // Non-ok = idle timeout or the warm handler crashed without signaling.
-  if (!result.ok) return null;
-  return result.output;
 }
 
+/**
+ * Release claimed kinds and re-offer the buffer to the drain.
+ *
+ * Re-attaching is the sweep: `on()` re-offers every buffered record to the
+ * newly attached handler, so a record that was buffered while its kind was
+ * still claimed (the waiter-gap window between `once()` iterations) is
+ * discarded now and the cursor advances past it.
+ * @internal
+ */
+function releaseChatInputKinds(kinds: readonly string[]): void {
+  const claimed = chatClaimedKinds();
+  let changed = false;
+  for (const kind of kinds) {
+    if (claimed.delete(kind)) changed = true;
+  }
+  if (!changed) return;
+
+  const drain = locals.get(chatInputDrainKey);
+  if (!drain) return;
+  drain.off();
+  locals.set(chatInputDrainKey, attachUnclaimedChatInputDrain());
+}
+
+/**
+ * Declare that this run will consume the given `session.in` record kinds
+ * itself (via raw `session.in` reads). Claimed kinds are never discarded by
+ * the unclaimed-control drain: they stay buffered, block
+ * `chat.messages.next()` at the head of the channel, and hold the resume
+ * cursor behind them until consumed. Call `release()` when the loop stops
+ * consuming them — any still-buffered records of those kinds are then
+ * discarded and the cursor advances.
+ */
+
 /**
  * Per-turn deferred promises. Registered via `chat.defer()`, awaited
  * before `onTurnComplete` fires. Reset each turn.
@@ -5428,6 +5529,14 @@ function chatCustomAgent<
       // listener — otherwise a continuation boot replays already-answered
       // messages into the loop's first wait.
       await seedSessionInResumeCursorForCustomLoop(payload);
+      // Claim the kinds this boot actually has a consumer for, then attach the
+      // drain for everything else. Handover kinds are only claimed on a
+      // handover-prepare boot, which is the only boot that waits for them.
+      const claimed = chatClaimedKinds();
+      if (payload.trigger === "handover-prepare") {
+        for (const kind of CHAT_HANDOVER_KINDS) claimed.add(kind);
+      }
+      locals.set(chatInputDrainKey, attachUnclaimedChatInputDrain());
       return userRun(payload, runOptions);
     },
   });
@@ -10765,6 +10874,7 @@ export const chat = {
   response: chatResponse,
   /** Pre-built input stream for receiving messages from the transport. */
   messages: messagesInput,
+  /** Declare `session.in` record kinds this run consumes itself. See {@link chatClaimInputKinds}. */
   /** Create a managed stop signal wired to the stop input stream. See {@link createStopSignal}. */
   createStopSignal,
   /** Signal the frontend that the current turn is complete. See {@link chatWriteTurnComplete}. */

packages/trigger-sdk/test/chat-messages-mailbox.test.ts

@@ -64,7 +64,7 @@ describe("chat.messages mailbox", () => {
 
     await runInMockTaskContext(async (drivers) => {
       const runPromise = run(
-        { chatId, trigger: "preload" },
+        { chatId, trigger: "handover-prepare" },
         { ctx: drivers.ctx, signal: new AbortController().signal }
       );
       await ready.promise;
@@ -156,7 +156,7 @@ describe("chat.messages mailbox", () => {
 
     await runInMockTaskContext(async (drivers) => {
       const runPromise = run(
-        { chatId, trigger: "preload" },
+        { chatId, trigger: "handover-prepare" },
         { ctx: drivers.ctx, signal: new AbortController().signal }
       );
       await ready.promise;
@@ -229,7 +229,7 @@ describe("chat.messages mailbox", () => {
 
     await runInMockTaskContext(async (drivers) => {
       const runPromise = run(
-        { chatId, trigger: "preload" },
+        { chatId, trigger: "handover-prepare" },
         { ctx: drivers.ctx, signal: new AbortController().signal }
       );
       await ready.promise;