feat(chat): custom agent mailbox helpers and session.in delivery fixes (#4644)

## Summary

Adds `chat.messages.hasPending()` and `chat.messages.next()` so a custom
agent
loop can inspect pending chat input without consuming it and take one
record at a
time, and fixes four ways a chat could mishandle input across a restart:
a
message silently lost, a recovered answer cut off by a stop the user had
already
pressed, a retried send answered twice, and a record the agent had no
consumer
for blocking every message queued behind it.

```ts
if (await chat.messages.hasPending()) {
  const record = await chat.messages.next({ timeoutInSeconds: 0 });
  if (record) handle(record.payload);
}
```

## Why the fixes came together

`session.in` carries records for consumers whose delivery needs differ.
A user
message must be delivered eventually, so it can wait arbitrarily long
for a turn
to take it. A stop only means anything to the turn that is live when it
lands.
Progress along the channel was tracked as one sequence number, and one
number
cannot say "control applied through 7, message 3 still owed" at the same
time.
Each of the bugs above is that mismatch surfacing somewhere different.

So instead of a rule per symptom, records are now classified once and
handed to
one route, and each route declares two things: whether it holds a record
when no
consumer is ready, and whether a record it never handled has to survive
into the
next boot. The resume cursor, the replay window and the
discard-the-unowned
behaviour are then derived from route state rather than maintained
beside it, and
`hasPending()` answers from the message queue instead of the head of a
buffer
shared with every other kind.

The wire is unchanged. Both cursors on the turn boundary keep their
meanings, so
existing chats resume as before and there is no webapp change.

## Behaviour worth calling out

`chat.writeTurnComplete()`'s `sessionInEventId` is the cursor that is
safe to
resume from, not the sequence of the record the turn answered. It is
held back
behind any message still waiting to be handled, so a value below the
record you
just handled is expected rather than a sign of a lost turn.

The stop fix also covers chats whose most recent turn was completed by
an older
SDK, by resolving the replay window from the channel when the boundary
does not
carry one. The trade there is deliberate: a stop that landed in the
moments
before boot and was never applied is dropped along with the replayed
ones,
because a stop the user can press again beats a stale one killing an
answer they
are waiting for.

## Verification

Thirteen reproductions against a local stack, each driving real runs
rather than
mocks, covering the documented `next()`/`hasPending()` loop, suspend and
resume, a
crash between consuming a message and writing turn-complete, a retried
send whose
idempotency claim is lost, and a continuation boot that must not replay
answered
messages. Where applicable each was also run against `main`, so the
fixes are
differences rather than assertions.

Five further legs on a deployed environment, which the earlier revisions
of this
branch did not cover at all: a message appended while the run is
genuinely
checkpointed, a message appended while the run is dead, the
stop-after-crash case
on the real crash path, and both version-skew directions (a newer worker
resuming
an older worker's turn boundary, and an older worker resuming a newer
one's).

Two of those restart fixes also have a browser-driven red and green pair
on a
deployed environment, staged identically on both sides and differing
only in the
SDK. For the lost-message fix, the unanswered message is replayed and
answered in
full here, and is never replayed at all on the released SDK. For the
stop fix,
both sides replay the message and diverge on the stop itself: it is
declined here
and the answer completes, while the released SDK re-applies it and the
recovered
answer dies before it streams.

The routing decision itself is a pure state machine, so it also has a
property
test over every interleaving of the record kinds crossed with each crash
point,
checked by mutation to confirm it fails when the cursor arithmetic or
the replay
window is broken.

## Known and not addressed here

The read of the woken record is unbounded, so a wake with nothing to
read makes
`wait()` outlive its own waitpoint. Tested and not a deadlock, since the
read
defers to the next record, but bounding it is a separate change with its
own
test.

Separately, and not caused by this branch: a run that crashes while a
message is
still queued is not replaced until the next inbound append, so that
message waits
rather than being recovered on its own. Worth its own issue.

Also not caused by this branch, but worth knowing when reading the
release note: a
chat page that stayed open across the crash keeps showing the partial
answer it
already received, so the recovered answer only appears after a reload.
The answer
itself is persisted correctly. The gap is on the client, which does not
apply a
re-delivered turn over a partial it already holds.

---------

Co-authored-by: Eric Allam <eric@trigger.dev>
Co-authored-by: Eric Allam <eallam@icloud.com>

Author: gtremper

.changeset/tidy-mailboxes-wait.md

@@ -0,0 +1,25 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+Fixes a case where a chat could silently lose a message. If a message arrived while the agent was between turns and a stop arrived after it, the cursor the next boot resumed from could point past that message, so it was never answered and no error was raised. This affected `chat.agent`, not just custom agents.
+
+Fixes a recovered answer being cut off. After a crash the agent replays the message it had not answered yet, but it was replaying the stop that arrived after that message too, so the turn answering it was aborted the moment it began. A stop is now only applied to the turn that was live when it arrived. That holds however the stop got there: sent after the last completed turn, or sent to a chat whose most recent turn was completed by an older version of the SDK.
+
+One limitation to know about: the recovered answer is persisted correctly, but a chat page that stayed open across the crash keeps showing the partial answer it had already received. Reload the page to see the full recovered answer.
+
+Also fixes a retried send being answered twice. When a send was retried and its idempotency claim was lost, the agent could consume the same message a second time.
+
+Custom agent loops can now inspect pending chat input without consuming it, and consume one record at a time, with `chat.messages.hasPending()` and `chat.messages.next()`. Records carry stable identifiers so a redelivery is recognisable.
+
+```ts
+if (await chat.messages.hasPending()) {
+  const record = await chat.messages.next({ timeoutInSeconds: 0 });
+  if (record) handle(record.payload);
+}
+```
+
+`hasPending()` answers for messages alone, so a message sitting behind a stop, or behind a record this version of the SDK does not recognise, still reports as pending and is still delivered. Anything the agent has no consumer for is discarded rather than left where it would make every message queued behind it undeliverable. `chat.messages.next()` returning `undefined` means no message became consumable before the timeout.
+
+`chat.writeTurnComplete()`'s `sessionInEventId` is the cursor that is safe to resume from, not the sequence of the record the turn answered. It is held back behind any message still waiting to be handled, so a value below the record you just handled is expected.

docs/ai-chat/custom-agents.mdx

@@ -179,6 +179,12 @@ for await (const turn of session) {
 
 The frontend stops a turn with [`transport.stopGeneration(chatId)`](/ai-chat/frontend#stop-generation), which writes a stop signal to the session's input stream. It aborts the current turn's generation but keeps the run alive, so the next message continues on the same session.
 
+A stop only applies to the turn that was live when it arrived. If the run crashes
+and a later run recovers a message that had not been answered yet, a stop that
+was already applied before the crash is not applied again, so the turn answering
+the recovered message runs to completion. A stop sent after the recovery is live
+and aborts that turn as normal.
+
 `turn.signal` is a combined stop-and-cancel `AbortSignal`, fresh each turn. Pass it to `streamText` so the stop reaches the model, then let `turn.complete()` finish the turn:
 
 ```ts trigger/my-chat.ts
@@ -213,14 +219,65 @@ For full control, skip `createSession` and compose the primitives directly:
 
 | Primitive                       | Description                                                                                  |
 | ------------------------------- | -------------------------------------------------------------------------------------------- |
-| `chat.messages`                 | Input stream for incoming messages — use `.waitWithIdleTimeout()` to wait for the next turn  |
+| `chat.messages`                 | Mailbox for incoming messages — inspect buffered input, consume one record, or suspend until the next turn |
 | `chat.createStopSignal()`       | Create a managed stop signal wired to the stop input stream                                  |
 | `chat.pipeAndCapture(result)`   | Pipe a stream and capture the response; returns `{ message, status, error }`                 |
 | `chat.writeTurnComplete()`      | Signal turn complete; returns `{ lastEventId, sessionInEventId }` resume cursors             |
 | `chat.MessageAccumulator`       | Accumulates conversation messages across turns                                               |
 | `chat.pipe(stream)`             | Pipe a stream to the frontend (no response capture)                                          |
 | `chat.cleanupAbortedParts(msg)` | Clean up incomplete parts from a stopped response                                            |
 
+### `chat.messages` mailbox
+
+`chat.messages` exposes the incoming message mailbox for hand-rolled loops:
+
+| Method | Behavior |
+| --- | --- |
+| `peek()` | Return the next queued message without consuming it, or `undefined` when none is queued |
+| `hasPending()` | Resolve `true` when a message is queued; does not consume it |
+| `next({ timeoutInSeconds? })` | Consume exactly one message record in channel order, or resolve `undefined` when the optional timeout elapses |
+| `on(handler)` | Consume messages as they arrive and invoke the handler |
+| `waitWithIdleTimeout(options)` | Wait warm, then suspend the run until the next message arrives |
+
+`hasPending()` checks whether a message has already been delivered locally and is
+waiting for `next()` to take it. It does not query the remote Session channel or
+start a subscription. Use `waitWithIdleTimeout()` when the loop needs to idle
+until future input arrives.
+
+`next({ timeoutInSeconds: 0 })` is also a local, non-blocking read. Call
+`next()` without a timeout, or with a positive timeout, to subscribe for future
+input.
+
+`next()` returns a readonly record envelope:
+
+```ts
+const record = await chat.messages.next({ timeoutInSeconds: 5 });
+if (record) {
+  console.log(record.id, record.seqNum);
+  currentPayload = record.payload;
+}
+```
+
+- `id` is the append's stable idempotency key.
+- `seqNum` is the monotonic sequence on this Session's `.in` channel.
+- `payload` is the existing `ChatTaskWirePayload` delivered by the other mailbox methods.
+
+Both identifiers remain the same if the record is delivered again after a
+reconnect. Each `next()` call commits only the record it returns, so a loop that
+owns its own turn sequencing never advances past input it has not taken. By
+contrast, `on()` commits a record as soon as it dispatches the handler; avoid
+mixing `on()` and `next()` when a single loop owns mailbox consumption.
+
+The Session `.in` channel also carries control records such as stops and
+handovers. Those are routed to their own consumers and never block messages: a
+message that arrived behind one is still reported by `hasPending()` and still
+returned by `next()`, in channel order. The same holds for a record kind this
+version of the SDK does not recognise, which is discarded rather than left where
+it would make every message behind it undeliverable.
+
+`next()` returns `undefined` when no message became consumable before the
+timeout.
+
 A complete loop:
 
 ```ts trigger/my-chat-raw.ts

docs/ai-chat/reference.mdx

@@ -504,9 +504,9 @@ All methods available on the `chat` object from `@trigger.dev/sdk/ai`.
 | `chat.createSession(payload, options)`      | Create an async iterator for chat turns                                                                                      |
 | `chat.pipe(source, options?)`               | Pipe a stream to the frontend (from anywhere inside a task)                                                                  |
 | `chat.pipeAndCapture(source, options?)`     | Pipe and capture the response; returns `{ message, status, error }`                                                         |
-| `chat.writeTurnComplete(options?)`          | Signal turn complete; returns `{ lastEventId, sessionInEventId }` resume cursors                                            |
+| `chat.writeTurnComplete(options?)`          | Signal turn complete; returns `{ lastEventId, sessionInEventId }` resume cursors. `sessionInEventId` is a lower bound, not the sequence of the record the turn answered |
 | `chat.createStopSignal()`                   | Create a managed stop signal wired to the stop input stream                                                                  |
-| `chat.messages`                             | Input stream for incoming messages — use `.waitWithIdleTimeout()`                                                            |
+| `chat.messages`                             | Incoming message mailbox; supports non-consuming `.peek()` / `.hasPending()`, single-record `.next()`, `.on()`, and suspend-aware `.waitWithIdleTimeout()` |
 | `chat.local<T>({ id })`                     | Create a per-run typed local (see [`chat.local`](/ai-chat/chat-local))                             |
 | `chat.createStartSessionAction(taskId, options?)` | Returns a server action that creates a chat Session + triggers the first run + returns a session-scoped PAT. Idempotent on `(env, externalId)`.   |
 | `chat.waitForHandover(options)`             | Wait for a [`chat.headStart`](/ai-chat/fast-starts#handover-with-custom-agents) handover signal in a custom loop. Returns the signal or `null`. `chat.MessageAccumulator` wraps this as `consumeHandover()` / `applyHandover()` |
@@ -645,7 +645,7 @@ The `onEvent` callback receives a `ChatTransportEvent` (exported from `@trigger.
 | `message-send-failed` | `messageId?`, `source`, `error`, `status?`, `durationMs`, `partId?`, `bodyBytes?` | A send definitively failed after internal retries. Fires in addition to `useChat`'s `onError`. |
 | `stream-connected` | `resumed`, `lastEventId?`, `messageId?` | The SSE subscription to the session's output stream started delivering. `resumed: true` when reconnecting from a stored cursor (page reload) rather than following a fresh send. `lastEventId` is the cursor it connected from. |
 | `first-chunk` | `chunkType?`, `lastEventId?`, `messageId?`, `sinceSendMs?` | The first response chunk of a turn arrived. `sinceSendMs` is the delta from the last turn-producing send — time to first token without any bookkeeping. |
-| `turn-completed` | `lastEventId?`, `sessionInEventId?`, `messageId?`, `sinceSendMs?` | The agent's turn-complete control record arrived — the "finished answering" signal. `sinceSendMs` is the full turn latency; `sessionInEventId` is the agent's committed input-stream cursor. |
+| `turn-completed` | `lastEventId?`, `sessionInEventId?`, `messageId?`, `sinceSendMs?` | The agent's turn-complete control record arrived — the "finished answering" signal. `sinceSendMs` is the full turn latency; `sessionInEventId` is the cursor the agent can safely resume its input stream from. Treat it as a lower bound: it is held back behind any message still waiting to be handled, so it can be below the sequence of the record this turn answered. Do not use it to decide whether a turn boundary belongs to your own send. |
 | `stream-error` | `error`, `status?` | The output stream failed unrecoverably. |
 
 `source` identifies the send path: `"submit-message"`, `"regenerate-message"`, `"steer"` (`sendPendingMessage`), `"action"` (`sendAction`), `"stop"` (`stopGeneration`), or `"head-start"`.

packages/core/src/v3/apiClient/runStream.test.ts

@@ -492,6 +492,7 @@ describe("SSEStreamSubscription v2 batch parsing — record kinds", () => {
   });
 
   type ParsedPart = {
+    recordId?: string;
     id: string;
     chunk: unknown;
     headers?: ReadonlyArray<readonly [string, string]>;
@@ -548,6 +549,7 @@ describe("SSEStreamSubscription v2 batch parsing — record kinds", () => {
     const parts = await sub.subscribe().then(drain);
 
     expect(parts).toHaveLength(1);
+    expect(parts[0]!.recordId).toBe("p1");
     expect(parts[0]!.id).toBe("5");
     expect(parts[0]!.chunk).toEqual({ type: "text-delta", delta: "hi" });
     expect(parts[0]!.headers).toEqual([]);

packages/core/src/v3/apiClient/runStream.ts

@@ -170,6 +170,9 @@ export interface StreamSubscriptionFactory {
 }
 
 export type SSEStreamPart<TChunk = unknown> = {
+  /** Stable logical record id from the S2 data envelope (`X-Part-Id` on append). */
+  recordId?: string;
+  /** S2 sequence number in decimal-string form. */
   id: string;
   chunk: TChunk;
   timestamp: number;
@@ -502,6 +505,7 @@ export class SSEStreamSubscription implements StreamSubscription {
                     chunkController.enqueue({
                       type: "part",
                       part: {
+                        recordId: parsedBody?.id,
                         id: record.seq_num.toString(),
                         chunk: parsedBody?.data,
                         timestamp: record.timestamp,

packages/core/src/v3/session-streams-api.ts

@@ -7,3 +7,4 @@ export const sessionStreams = SessionStreamsAPI.getInstance();
 export * from "./sessionStreams/types.js";
 export * from "./sessionStreams/wireProtocol.js";
 export * from "./sessionStreams/chatSnapshot.js";
+export * from "./sessionStreams/router.js";

packages/core/src/v3/sessionStreams/index.ts

@@ -1,6 +1,12 @@
 import { getGlobal, registerGlobal } from "../utils/globals.js";
 import { NoopSessionStreamManager } from "./noopManager.js";
-import type { InputStreamOncePromise, SessionChannelIO, SessionStreamManager } from "./types.js";
+import type {
+  InputStreamOncePromise,
+  SessionChannelIO,
+  SessionStreamManager,
+  SessionStreamRecord,
+  SessionStreamRecordPredicate,
+} from "./types.js";
 import type { InputStreamOnceOptions } from "../realtimeStreams/types.js";
 
 const API_NAME = "session-streams";
@@ -35,6 +41,18 @@ export class SessionStreamsAPI implements SessionStreamManager {
     return this.#getManager().on(sessionId, io, handler);
   }
 
+  public onRecord(
+    sessionId: string,
+    io: SessionChannelIO,
+    handler: (record: SessionStreamRecord) => void | boolean | Promise<void>
+  ): { off: () => void } {
+    const manager = this.#getManager();
+    if (!manager.onRecord) {
+      throw new Error("The configured Session stream manager does not support record handlers");
+    }
+    return manager.onRecord(sessionId, io, handler);
+  }
+
   public once(
     sessionId: string,
     io: SessionChannelIO,
@@ -43,10 +61,43 @@ export class SessionStreamsAPI implements SessionStreamManager {
     return this.#getManager().once(sessionId, io, options);
   }
 
+  public onceRecord(
+    sessionId: string,
+    io: SessionChannelIO,
+    options?: InputStreamOnceOptions
+  ): InputStreamOncePromise<SessionStreamRecord> {
+    const manager = this.#getManager();
+    if (!manager.onceRecord) {
+      throw new Error("The configured Session stream manager does not support record metadata");
+    }
+    return manager.onceRecord(sessionId, io, options);
+  }
+
+  public onceRecordWhere(
+    sessionId: string,
+    io: SessionChannelIO,
+    predicate: SessionStreamRecordPredicate,
+    options?: InputStreamOnceOptions
+  ): InputStreamOncePromise<SessionStreamRecord> {
+    const manager = this.#getManager();
+    if (!manager.onceRecordWhere) {
+      throw new Error("The configured Session stream manager does not support selective records");
+    }
+    return manager.onceRecordWhere(sessionId, io, predicate, options);
+  }
+
   public peek(sessionId: string, io: SessionChannelIO): unknown | undefined {
     return this.#getManager().peek(sessionId, io);
   }
 
+  public peekRecord(sessionId: string, io: SessionChannelIO): SessionStreamRecord | undefined {
+    const manager = this.#getManager();
+    if (!manager.peekRecord) {
+      throw new Error("The configured Session stream manager does not support record metadata");
+    }
+    return manager.peekRecord(sessionId, io);
+  }
+
   public lastSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
     return this.#getManager().lastSeqNum(sessionId, io);
   }
@@ -55,6 +106,14 @@ export class SessionStreamsAPI implements SessionStreamManager {
     this.#getManager().setLastSeqNum(sessionId, io, seqNum);
   }
 
+  public consumeRecord(sessionId: string, io: SessionChannelIO, seqNum: number): void {
+    const manager = this.#getManager();
+    if (!manager.consumeRecord) {
+      throw new Error("The configured Session stream manager does not support exact consumption");
+    }
+    manager.consumeRecord(sessionId, io, seqNum);
+  }
+
   public lastDispatchedSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
     return this.#getManager().lastDispatchedSeqNum(sessionId, io);
   }
@@ -75,6 +134,10 @@ export class SessionStreamsAPI implements SessionStreamManager {
     return this.#getManager().shiftBuffer(sessionId, io);
   }
 
+  public reconnectStream(sessionId: string, io: SessionChannelIO): void {
+    this.#getManager().reconnectStream?.(sessionId, io);
+  }
+
   public disconnectStream(sessionId: string, io: SessionChannelIO): void {
     this.#getManager().disconnectStream(sessionId, io);
   }