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

.changeset/tidy-mailboxes-wait.md

@@ -0,0 +1,23 @@
+---
+"@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, including for chats whose most recent turn was completed by an older version of the SDK.
+
+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);
   }