feat(chat): add mailbox helpers for custom agents

.changeset/tidy-mailboxes-wait.md

@@ -0,0 +1,8 @@
+---
+"@trigger.dev/core": patch
+"@trigger.dev/sdk": patch
+---
+
+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` means no message became consumable before the timeout.

docs/ai-chat/custom-agents.mdx

@@ -213,14 +213,68 @@ 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 buffer head when it is a message, without consuming it; otherwise return `undefined` |
+| `hasPending()` | Resolve `true` when the buffer head is a message; 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 the local, already-delivered buffer head is a
+message that `next()` can consume immediately. 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 handovers. If one
+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.
+
+`next()` still returns `undefined` whenever no message became consumable before
+the timeout, including while a control record that does have its own consumer
+sits at the head.
+
 A complete loop:
 
 ```ts trigger/my-chat-raw.ts

docs/ai-chat/reference.mdx

@@ -506,7 +506,7 @@ All methods available on the `chat` object from `@trigger.dev/sdk/ai`.
 | `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.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()` |

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/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";
@@ -43,10 +49,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 +94,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);
   }