fix(chat): stop losing a user message that arrived mid-turn (#4795)

Follow-up to
[#4644](#4644), now
rebased onto main so the diff is just these three commits.

## Summary

Two ways a chat could lose a user message, both pre-existing and both
raised while reviewing #4644.

A message arriving while a turn was streaming was handed to that turn's
push handler and parked in an in-memory array. The router counts a
record handed to a handler as terminally decided, so it stopped holding
the resume floor behind it, and the turn boundary published a cursor
past a message that existed only in that process. A crash before the
next turn lost it, silently. Measured: with the message at sequence 1,
the boundary published `session-in-event-id: 1`, so a resume skipped it.

Separately, a message the agent declined to inject was discarded with
the turn. Never injected, never written to the wire buffer, never
answered. That was also the documented default, since a
`pendingMessages` config without `shouldInject` declines every batch.

## Design

Notification and consumption are now separate concerns on the router.

`observe` reports that a record arrived without taking it, so the record
stays queued and keeps holding the floor. It is rejected on an
`at-arrival` route: an observer there would either have to count as a
listener, which would stop an unconsumed stop being discarded and bring
back a wedged mailbox, or watch records it cannot affect. `take` removes
exactly one queued record.

The managed loop and the `chat.createSession()` iterator now only
subscribe when there is a steering config to feed, and injection is the
point of consumption. A declined batch never reaches the take, so its
records stay queued and become later turns. Both in-memory wire buffers
are gone, so a message waiting for its turn is durable rather than
living in whichever worker received it.

The floor doubles as the wake cursor: `awaitWake` registers with it and
the server completes the waitpoint immediately if anything sits after
that sequence. An over-advanced floor was therefore also a missed wake.
It is now recorded on the wait span so a run that never woke can be
diagnosed from its trace.

## Verification

Both fixes have a red and green pair, each checked against the
unmodified source rather than only observed to pass:

- the resume cursor test fails on the parent branch and passes here
- the declined-message test fails without the second commit and passes
with it

Also 8 new router tests for `observe` and `take`. Suites green at 385
for the SDK and 886 for core.

## Not addressed

A `pendingMessages` config with no `chat.toStreamTextOptions()` spread
still swallows messages, because nothing drains the queue at all. Same
shape, different trigger, tracked separately.

Author: ericallam

.changeset/quiet-floors-hold.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fixes a message sent while the agent was mid-answer being lost if the run then crashed. The cursor written at the end of each turn could point past a message that had arrived during that turn but had not been answered yet, so the next boot skipped it and no error was raised anywhere. Such a message is now held until a turn actually takes it.
+
+This also removes the in-memory buffer those messages used to sit in, on both `chat.agent` and `chat.createSession()`, so a message waiting for its turn is durable rather than only present in the worker that received it.

.changeset/spry-steers-defer.md

@@ -0,0 +1,28 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+A message that arrives mid-turn and is not injected into that turn is now answered as the next turn, instead of being dropped. This is what the `pendingMessages` docs have always described, and it applies to the default too: configuring `pendingMessages` without a `shouldInject` declines every batch, which previously meant every mid-turn message was lost with no error at either end.
+
+```ts
+chat.agent({
+  id: "my-chat",
+  pendingMessages: {
+    onReceived: ({ message }) => logger.info("arrived mid-turn", { id: message.id }),
+    // Only interrupt once the agent has started calling tools.
+    shouldInject: ({ steps }) => steps.length > 0,
+  },
+  run: async ({ messages, signal }) =>
+    streamText({
+      model,
+      messages,
+      abortSignal: signal,
+      // Required for injection. Without it nothing injects, and every
+      // mid-turn message is answered as the next turn instead.
+      ...chat.toStreamTextOptions(),
+    }),
+});
+```
+
+A declined message keeps its place in the queue, so it survives a crash and is answered by whichever run picks the conversation up. An injected one is consumed at the moment it is injected, so it is never also answered as a later turn.

docs/ai-chat/client-protocol.mdx

@@ -953,9 +953,12 @@ You can send messages while the agent is still streaming a response. These are *
 
 The wire format is identical to a normal `kind: "message"` send — same `.in` channel, single `message` field. The difference is timing. What happens depends on the agent's `pendingMessages` configuration:
 
-- **With `pendingMessages.shouldInject`**: the message is injected into the model's context at the next `prepareStep` boundary. The agent sees it and can adjust its behavior mid-response.
+- **With `pendingMessages.shouldInject` returning `true`**: the message is injected into the model's context at the next `prepareStep` boundary. The agent sees it and can adjust its behavior mid-response.
+- **With a `pendingMessages` config that declines it**, either because `shouldInject` returned `false` or because it is absent: the message stays queued on the backend and is answered as the next turn.
 - **Without `pendingMessages` config**: the message queues for the next turn.
 
+In every case the message is answered. A declined message keeps its place in the queue, so it also survives a crash and is picked up by whichever run continues the conversation.
+
 See [Pending Messages](/ai-chat/pending-messages) for how to configure the agent side.
 
 <Note>

docs/ai-chat/pending-messages.mdx

@@ -10,7 +10,9 @@ When an AI agent is executing tool calls, users may want to send a message that
 
 By default (without `pendingMessages`), a message sent while the agent is responding never interrupts the in-flight response: it's buffered and processed as its own turn once the current turn completes, with multiple messages running sequentially in arrival order.
 
-The `pendingMessages` option enables steering instead, injecting user messages between tool-call steps via the AI SDK's `prepareStep`. Messages that arrive during streaming are queued and injected at the next step boundary. If there are no more step boundaries (single-step response or final text generation), the message becomes the next turn automatically.
+The `pendingMessages` option enables steering instead, injecting user messages between tool-call steps via the AI SDK's `prepareStep`. Messages that arrive during streaming are queued and injected at the next step boundary. A message that is not injected becomes the next turn instead, whether that is because `shouldInject` returned `false` or because there were no more step boundaries (single-step response or final text generation). The backend handles that, so no client-side re-send is involved.
+
+Injection is what needs wiring: the `pendingMessages` options only reach `streamText` if you spread `chat.toStreamTextOptions()` (or pass `prepareStep`). Without that, nothing injects, so every mid-turn message is answered as the next turn. Deferral does not depend on it.
 
 ## How it works
 
@@ -20,7 +22,7 @@ The `pendingMessages` option enables steering instead, injecting user messages b
 4. At the next `prepareStep` boundary (between tool-call steps), `shouldInject` is called
 5. If it returns `true`, the message is injected into the LLM's context
 6. A `data-pending-message-injected` stream chunk confirms injection to the frontend
-7. If `prepareStep` never fires (no tool calls), the message becomes the next turn
+7. If `shouldInject` returns `false`, or `prepareStep` never fires (no tool calls), the message stays queued on the backend and is answered as the next turn
 
 ## Backend: chat.agent
 
@@ -310,7 +312,7 @@ function Chat({ chatId }: { chatId: string }) {
 
 ### Message lifecycle
 
-- **Steering messages** are sent via `transport.sendPendingMessage()` immediately. They appear as purple pending bubbles. If injected, they disappear from the overlay and render inline at the injection point. If not injected (no more step boundaries), they auto-send as the next turn when the response finishes.
+- **Steering messages** are sent via `transport.sendPendingMessage()` immediately. They appear as purple pending bubbles. If injected, they disappear from the overlay and render inline at the injection point. If not injected, the backend answers them as the next turn once the response finishes; the client does not need to re-send them.
 
 - **Queued messages** stay client-side until the turn completes, then auto-send as the next turn via `sendMessage()`. They can be promoted to steering mid-stream by clicking "Steer instead".
 

docs/ai-chat/reference.mdx

@@ -396,7 +396,7 @@ Options for the `pendingMessages` field. See [Pending Messages](/ai-chat/pending
 
 | Option         | Type                                                                              | Required | Description                                                                               |
 | -------------- | --------------------------------------------------------------------------------- | -------- | ----------------------------------------------------------------------------------------- |
-| `shouldInject` | `(event: PendingMessagesBatchEvent) => boolean \| Promise<boolean>`               | No       | Decide whether to inject the batch between tool-call steps. If absent, no injection.      |
+| `shouldInject` | `(event: PendingMessagesBatchEvent) => boolean \| Promise<boolean>`               | No       | Decide whether to inject the batch between tool-call steps. If absent, nothing is injected and the messages are answered as the next turn. Only consulted when `chat.toStreamTextOptions()` (or `prepareStep`) reaches `streamText`; without that nothing is injected and every mid-turn message becomes the next turn. |
 | `prepare`      | `(event: PendingMessagesBatchEvent) => ModelMessage[] \| Promise<ModelMessage[]>` | No       | Transform the batch before injection. Default: convert each via `convertToModelMessages`. |
 | `onReceived`   | `(event: PendingMessageReceivedEvent) => void \| Promise<void>`                   | No       | Called when a message arrives during streaming (per-message).                             |
 | `onInjected`   | `(event: PendingMessagesInjectedEvent) => void \| Promise<void>`                  | No       | Called after a batch is injected via prepareStep.                                         |

packages/core/src/v3/sessionStreams/router.test.ts

@@ -407,3 +407,169 @@ describe("SessionChannelRouter: exactly-once across a crash", () => {
     }
   });
 });
+
+describe("SessionChannelRouter: observe", () => {
+  it("notifies without consuming, so the record still queues and holds the floor", () => {
+    const r = router();
+    const seen: number[] = [];
+    r.observe("messages", (record) => seen.push(record.seqNum));
+
+    r.ingest(rec(0, "message"));
+
+    expect(seen).toEqual([0]);
+    expect(r.hasPending("messages")).toBe(true);
+    expect(r.resumeFloor()).toBeUndefined();
+  });
+
+  it("does not satisfy a queue route's handler delivery", () => {
+    const r = router();
+    const observed: number[] = [];
+    const handled: number[] = [];
+    r.observe("messages", (record) => observed.push(record.seqNum));
+
+    r.ingest(rec(0, "message"));
+    expect(handled).toEqual([]);
+
+    r.on("messages", (record) => handled.push(record.seqNum));
+    expect(handled).toEqual([0]);
+    expect(observed).toEqual([0]);
+  });
+
+  it("rejects an at-arrival route, so a stop with only an observer is still discarded", () => {
+    const r = router();
+    expect(() => r.observe("stop", () => {})).toThrow(/at-arrival/);
+  });
+
+  it("does not re-offer records that were already queued when it attached", () => {
+    const r = router();
+    r.ingest(rec(0, "message"));
+
+    const seen: number[] = [];
+    r.observe("messages", (record) => seen.push(record.seqNum));
+    expect(seen).toEqual([]);
+
+    r.ingest(rec(1, "message"));
+    expect(seen).toEqual([1]);
+  });
+
+  it("stops notifying after off()", () => {
+    const r = router();
+    const seen: number[] = [];
+    const sub = r.observe("messages", (record) => seen.push(record.seqNum));
+
+    r.ingest(rec(0, "message"));
+    sub.off();
+    r.ingest(rec(1, "message"));
+
+    expect(seen).toEqual([0]);
+  });
+});
+
+describe("SessionChannelRouter: take", () => {
+  it("removes one queued record by sequence and releases the floor", () => {
+    const r = router();
+    r.ingest(rec(0, "message"));
+    r.ingest(rec(1, "message"));
+
+    expect(r.take("messages", 0)?.seqNum).toBe(0);
+    expect(r.pendingCount("messages")).toBe(1);
+    expect(r.peek("messages")?.seqNum).toBe(1);
+  });
+
+  it("reports false for a record that is no longer queued", () => {
+    const r = router();
+    r.ingest(rec(0, "message"));
+
+    expect(r.take("messages", 0)?.seqNum).toBe(0);
+    expect(r.take("messages", 0)).toBeUndefined();
+    expect(r.take("messages", 99)).toBeUndefined();
+  });
+
+  it("leaves an untaken observed record to be delivered as normal", async () => {
+    const r = router();
+    r.observe("messages", () => {});
+    r.ingest(rec(0, "message"));
+    r.ingest(rec(1, "message"));
+
+    r.take("messages", 0);
+
+    const next = await r.next("messages", { timeoutMs: 0 });
+    expect(next?.seqNum).toBe(1);
+  });
+});
+
+describe("SessionChannelRouter: clearRoute guard", () => {
+  it("refuses to clear a replayable route", () => {
+    const r = router();
+    r.ingest(rec(0, "message"));
+
+    expect(() => r.clearRoute("messages")).toThrow(/replayable/);
+    expect(r.hasPending("messages")).toBe(true);
+  });
+
+  it("still clears a non-replayable route", () => {
+    const r = router();
+    r.ingest(rec(0, "handover"));
+    expect(r.hasPending("handover")).toBe(true);
+
+    r.clearRoute("handover");
+    expect(r.hasPending("handover")).toBe(false);
+  });
+});
+
+describe("SessionChannelRouter: observe versus a waiting consumer", () => {
+  it("notifies the observer even when a parked puller takes the record", async () => {
+    const r = router();
+    const seen: number[] = [];
+    r.observe("messages", (record) => seen.push(record.seqNum));
+
+    const pull = r.next("messages");
+    r.ingest(rec(0, "message"));
+    const taken = await pull;
+
+    expect(seen).toEqual([0]);
+    expect(taken?.seqNum).toBe(0);
+    expect(r.hasPending("messages")).toBe(false);
+  });
+
+  it("reports a failed take for a record a puller already consumed", async () => {
+    const r = router();
+    const seen: number[] = [];
+    r.observe("messages", (record) => seen.push(record.seqNum));
+
+    const pull = r.next("messages");
+    r.ingest(rec(0, "message"));
+    await pull;
+
+    expect(seen).toEqual([0]);
+    expect(r.take("messages", 0)).toBeUndefined();
+  });
+});
+
+describe("SessionChannelRouter: untake", () => {
+  it("puts a claimed record back in sequence order", async () => {
+    const r = router();
+    r.ingest(rec(0, "message"));
+    r.ingest(rec(2, "message"));
+
+    const taken = r.take("messages", 0)!;
+    expect(r.peek("messages")?.seqNum).toBe(2);
+
+    r.untake("messages", taken);
+
+    expect(r.pendingCount("messages")).toBe(2);
+    expect(r.peek("messages")?.seqNum).toBe(0);
+    expect(r.resumeFloor()).toBeUndefined();
+  });
+
+  it("is idempotent, so a double return cannot duplicate a record", () => {
+    const r = router();
+    r.ingest(rec(0, "message"));
+    const taken = r.take("messages", 0)!;
+
+    r.untake("messages", taken);
+    r.untake("messages", taken);
+
+    expect(r.pendingCount("messages")).toBe(1);
+  });
+});

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

@@ -101,6 +101,7 @@ class RouteState {
   readonly queue: SessionStreamRecord[] = [];
   readonly waiters: QueueWaiter[] = [];
   readonly handlers = new Set<RouteHandler>();
+  readonly observers = new Set<RouteHandler>();
 
   constructor(readonly route: SessionRoute) {}
 
@@ -239,6 +240,14 @@ export class SessionChannelRouter {
       return this.#drop(record, "no-handler", routeName);
     }
 
+    for (const observer of state.observers) {
+      try {
+        observer(record);
+      } catch {
+        void 0;
+      }
+    }
+
     const waiter = state.waiters.shift();
     if (waiter) {
       if (waiter.timer) clearTimeout(waiter.timer);
@@ -356,6 +365,70 @@ export class SessionChannelRouter {
     };
   }
 
+  /**
+   * Watch a route without consuming from it.
+   *
+   * Notification and consumption are separate concerns: an observer is told
+   * that a record arrived and the record still queues, so the resume floor
+   * stays held behind it until something actually takes it. A push handler
+   * registered with {@link on} is the opposite, and a record handed to one
+   * counts as terminally decided.
+   *
+   * Only meaningful on a replayable route. On an `at-arrival` route an observer
+   * would either have to count as a listener, which would stop an unconsumed
+   * record being discarded, or watch records it cannot affect, so it is
+   * rejected rather than given one of those two meanings.
+   *
+   * Records already queued are not re-offered: an observer reports arrivals
+   * from the moment it attaches, so re-offering would fire twice for a record
+   * that arrived before a later consumer attached.
+   */
+  observe(name: string, observer: RouteHandler): { off: () => void } {
+    const state = this.#stateOrThrow(name);
+    if (state.route.delivery === "at-arrival") {
+      throw new Error(
+        `Route "${name}" is at-arrival, which cannot be observed: an observer must not decide whether a record is discarded, and cannot be offered one that already was`
+      );
+    }
+    state.observers.add(observer);
+    return {
+      off: () => {
+        state.observers.delete(observer);
+      },
+    };
+  }
+
+  /**
+   * Remove one queued record, identified by sequence.
+   *
+   * For a consumer that decided to take a record it had only observed. Returns
+   * whether it was still queued, so a caller can tell a real take from a record
+   * something else had already consumed.
+   */
+  take(name: string, seqNum: number): SessionStreamRecord | undefined {
+    const state = this.#stateOrThrow(name);
+    const index = state.queue.findIndex((record) => record.seqNum === seqNum);
+    if (index === -1) return undefined;
+    return state.queue.splice(index, 1)[0];
+  }
+
+  /**
+   * Put a taken record back, in sequence order.
+   *
+   * For a consumer that claimed a record and then could not use it. Returning
+   * it leaves the route as though the claim never happened, so the record is
+   * delivered later and goes back to holding the resume floor. Without this a
+   * failed claim-then-use is indistinguishable from a delivery, and the record
+   * is lost.
+   */
+  untake(name: string, record: SessionStreamRecord): void {
+    const state = this.#stateOrThrow(name);
+    if (state.queue.some((queued) => queued.seqNum === record.seqNum)) return;
+    const at = state.queue.findIndex((queued) => queued.seqNum > record.seqNum);
+    if (at === -1) state.queue.push(record);
+    else state.queue.splice(at, 0, record);
+  }
+
   /** Whether an `at-arrival` route currently has anywhere to deliver. */
   hasHandler(name: string): boolean {
     return this.#stateOrThrow(name).handlers.size > 0;
@@ -417,6 +490,11 @@ export class SessionChannelRouter {
    */
   clearRoute(name: string): void {
     const state = this.#stateOrThrow(name);
+    if (state.route.replayable) {
+      throw new Error(
+        `Route "${name}" is replayable, so its queue cannot be cleared: anything queued on it is still owed to a later boot, and discarding it would lose records the resume floor is holding back`
+      );
+    }
     state.queue.length = 0;
     for (const waiter of state.waiters) {
       if (waiter.timer) clearTimeout(waiter.timer);
@@ -430,6 +508,7 @@ export class SessionChannelRouter {
     for (const state of this.#routes.values()) {
       state.queue.length = 0;
       state.handlers.clear();
+      state.observers.clear();
       for (const waiter of state.waiters) {
         if (waiter.timer) clearTimeout(waiter.timer);
         waiter.resolve(undefined);