fix(sdk,core): stop chat sessions dropping messages that arrive during a turn (#4176)

## Summary

Sending a message to a chat whose run had ended could make the message
vanish: the continuation run replayed already-answered messages, never
processed the new one, and a page refresh lost it entirely. Chasing that
report surfaced four composing message-loss bugs in the chat session
runtime; this PR fixes all of them, each with a regression test.

## The fixes

1. **Stale resume cursor.** Records delivered while a run was suspended
(the waitpoint path) advanced the SSE resume counter but not the
committed-consume cursor, so the `session-in-event-id` header stamped on
turn-completes went stale by one record per suspended turn. Continuation
boots seed from that header, which is what made them replay
already-processed messages. `session.in.wait()` now advances both
cursors.

2. **Only the first buffered message dispatched.** Messages arriving
during a turn are consumed into a buffer whose end-of-turn pickup
dispatched only the first entry; the buffer was recreated each turn, so
the rest were discarded, and since consuming a record commits the cursor
the loss was permanent. A continuation boot's replay delivers several
records back-to-back, which put the user's new message at index 1 or
later. The buffer now outlives the turn and drains one message per turn
in both `chat.agent` and `chat.createSession` (whose equivalent buffer
was never read at all).

3. **Post-stop window in `chat.createSession`.** The turn's message
listener stayed attached through the stopped turn's post-stream work, so
a message sent shortly after stopping a turn was consumed into the dead
steering queue and lost. The listener now detaches when the stream
settles, matching the `chat.agent` loop.

4. **Handler leak on errored turns.** A turn that threw outside the
streaming section (for example from an `onTurnStart` hook) leaked its
message listener. Previously that silently lost mid-turn messages; with
the loop-level buffer it would have duplicated them instead. The
subscription handle is now detached by the turn's catch/finally, and
`chat.createSession` defensively detaches its prior turn's listener when
user code exits a turn without `complete()`/`done()`.

## Verification

Reproduced end-to-end with the ai-chat reference project before the fix
(message consumed but never answered, two replayed turns, gone on
refresh) and verified after (single clean turn, survives refresh,
turn-complete cursors strictly advancing). Regression tests in
`packages/trigger-sdk/test/pending-message-drain.test.ts` cover all
four, each verified red against the unfixed behavior. A smoke sweep of
the standard chat scenarios (basic send, multi-turn, suspend/resume,
mid-stream refresh, stop, steering, cancel + continue, and the
`createSession` variant) passes on the final branch state.

Author: ericallam

.changeset/chat-errored-turn-handler-leak.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fix chat turns that throw (for example from an `onTurnStart` hook) leaking their message listener, which lost or duplicated messages sent during later turns.

.changeset/chat-pending-message-drain.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fix `chat.agent` and `chat.createSession` permanently dropping user messages when several arrived during a single turn: every buffered message is now dispatched as its own turn instead of only the first.

.changeset/chat-session-in-resume-cursor.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fix chat continuation runs replaying already-answered messages: turns delivered while the run was suspended now advance the session.in resume cursor, so a new run picks up exactly where the previous one left off.

.changeset/chat-session-stop-window.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fix `chat.createSession` swallowing a message sent shortly after stopping a turn: the turn's message listener now detaches when the stream settles, so those messages run as the next turn.

docs/ai-chat/pending-messages.mdx

@@ -8,7 +8,9 @@ description: "Inject user messages mid-execution to steer agents between tool-ca
 
 When an AI agent is executing tool calls, users may want to send a message that **steers the agent mid-execution** — adding context, correcting course, or refining the request without waiting for the response to finish.
 
-The `pendingMessages` option enables this by 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.
+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.
 
 ## How it works
 

packages/core/src/v3/test/test-session-stream-manager.ts

@@ -33,6 +33,7 @@ export class TestSessionStreamManager implements SessionStreamManager {
   private onceWaiters = new Map<string, OnceWaiter[]>();
   private buffer = new Map<string, unknown[]>();
   private seqNums = new Map<string, number>();
+  private dispatchedSeqNums = new Map<string, number>();
 
   on(sessionId: string, io: SessionChannelIO, handler: Handler): { off: () => void } {
     const key = keyFor(sessionId, io);
@@ -150,15 +151,20 @@ export class TestSessionStreamManager implements SessionStreamManager {
     this.seqNums.set(keyFor(sessionId, io), seqNum);
   }
 
-  lastDispatchedSeqNum(_sessionId: string, _io: SessionChannelIO): number | undefined {
-    // The test harness drives records via `__sendFromTest` without seq
-    // numbers, so the committed-consume cursor stays undefined. Tests
-    // that need cursor behaviour exercise it via the real manager.
-    return undefined;
+  lastDispatchedSeqNum(sessionId: string, io: SessionChannelIO): number | undefined {
+    // `__sendFromTest` carries no seq numbers, so this only reflects
+    // explicit `setLastDispatchedSeqNum` calls (e.g. the waitpoint
+    // delivery path). Full cursor behaviour is exercised via the real
+    // manager.
+    return this.dispatchedSeqNums.get(keyFor(sessionId, io));
   }
 
-  setLastDispatchedSeqNum(_sessionId: string, _io: SessionChannelIO, _seqNum: number): void {
-    // no-op — see comment on `lastDispatchedSeqNum`.
+  setLastDispatchedSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void {
+    const key = keyFor(sessionId, io);
+    const current = this.dispatchedSeqNums.get(key);
+    if (current === undefined || seqNum > current) {
+      this.dispatchedSeqNums.set(key, seqNum);
+    }
   }
 
   setMinTimestamp(
@@ -202,6 +208,7 @@ export class TestSessionStreamManager implements SessionStreamManager {
     this.handlers.clear();
     this.buffer.clear();
     this.seqNums.clear();
+    this.dispatchedSeqNums.clear();
   }
 
   disconnect(): void {

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

@@ -5569,6 +5569,14 @@ function chatAgent<
       // `messagesInput.waitWithIdleTimeout` so recovered turns fire first.
       const bootInjectedQueue: ChatTaskWirePayload<TUIMessage, inferSchemaIn<TClientDataSchema>>[] =
         [];
+      // Messages consumed by a turn's `messagesInput.on` handler, dispatched
+      // one per turn by the end-of-turn pickup. Loop-level on purpose:
+      // consuming a record advances the committed `.in` cursor, so entries
+      // dropped with a turn-local buffer are lost permanently.
+      const pendingWireMessages: ChatTaskWirePayload<
+        TUIMessage,
+        inferSchemaIn<TClientDataSchema>
+      >[] = [];
       const couldHavePriorState = payload.continuation === true || ctx.attempt.number > 1;
 
       // `.in` resume cursor, computed at most once per boot. The boot
@@ -6378,6 +6386,9 @@ function chatAgent<
         }
 
         for (let turn = 0; turn < maxTurns; turn++) {
+          // Declared here so the finally can detach it — a handler leaked past
+          // its turn duplicates every mid-stream message into the shared buffer.
+          let turnMsgSub: { off: () => void } | undefined;
           try {
             // Extract turn-level context before entering the span. Slim
             // wire: at most one delta message per record. `headStartMessages`
@@ -6479,11 +6490,6 @@ function chatAgent<
                 const cancelSignal = runSignal;
                 const combinedSignal = AbortSignal.any([runSignal, stopController.signal]);
 
-                // Buffer messages that arrive during streaming
-                const pendingMessages: ChatTaskWirePayload<
-                  TUIMessage,
-                  inferSchemaIn<TClientDataSchema>
-                >[] = [];
                 const pmConfig = locals.get(chatPendingMessagesKey);
                 const msgSub = messagesInput.on(async (msg) => {
                   // If pendingMessages is configured, route to the steering queue
@@ -6532,10 +6538,11 @@ function chatAgent<
                   }
 
                   // No pendingMessages config — standard wire buffer for next turn
-                  pendingMessages.push(
+                  pendingWireMessages.push(
                     msg as ChatTaskWirePayload<TUIMessage, inferSchemaIn<TClientDataSchema>>
                   );
                 });
+                turnMsgSub = msgSub;
 
                 // Track new messages for this turn (user input + assistant response).
                 const turnNewModelMessages: ModelMessage[] = [];
@@ -7738,9 +7745,10 @@ function chatAgent<
                 }
 
                 // If messages arrived during streaming (without pendingMessages config),
-                // use the first one immediately as the next turn.
-                if (pendingMessages.length > 0) {
-                  currentWirePayload = pendingMessages[0]!;
+                // dispatch the oldest as the next turn. The rest stay queued
+                // and drain one per turn.
+                if (pendingWireMessages.length > 0) {
+                  currentWirePayload = pendingWireMessages.shift()!;
                   return "continue";
                 }
 
@@ -7847,6 +7855,9 @@ function chatAgent<
             // Turn error handler: write an error chunk + turn-complete to the stream
             // so the client sees the error, then wait for the next message instead
             // of killing the entire run. This keeps the conversation alive.
+            // Detach the turn's message handler first — left attached it would
+            // eat the very message the wait below is waiting for.
+            turnMsgSub?.off();
             if (
               turnError instanceof Error &&
               turnError.name === "AbortError" &&
@@ -7982,6 +7993,12 @@ function chatAgent<
               continue;
             }
 
+            // Same for messages buffered during the errored turn — already consumed, idling strands them.
+            if (pendingWireMessages.length > 0) {
+              currentWirePayload = pendingWireMessages.shift()!;
+              continue;
+            }
+
             // Wait for the next message — same as after a successful turn
             const effectiveIdleTimeout =
               (metadata.get(IDLE_TIMEOUT_METADATA_KEY) as number | undefined) ??
@@ -8004,6 +8021,8 @@ function chatAgent<
               inferSchemaIn<TClientDataSchema>
             >;
             // Continue to next iteration of the for loop
+          } finally {
+            turnMsgSub?.off();
           }
         }
       } finally {
@@ -9309,9 +9328,18 @@ function createChatSession(
       const accumulator = new ChatMessageAccumulator();
       let previousTurnUsage: LanguageModelUsage | undefined;
       let cumulativeUsage: LanguageModelUsage = emptyUsage();
+      // Messages consumed mid-turn, dispatched one per next(). Iterator-level
+      // for the same reason as the agent loop's `pendingWireMessages`:
+      // consumed records never replay, so a turn-local buffer loses them.
+      const sessionPendingWire: ChatTaskWirePayload[] = [];
+      // The current turn's message subscription — detached defensively at the
+      // top of next() in case user code threw without complete()/done().
+      let activeMsgSub: { off: () => void } | undefined;
 
       return {
         async next(): Promise<IteratorResult<ChatTurn>> {
+          activeMsgSub?.off();
+          activeMsgSub = undefined;
           if (!booted) {
             booted = true;
             await seedSessionInResumeCursorForCustomLoop(currentPayload);
@@ -9380,24 +9408,29 @@ function createChatSession(
             }
           }
 
-          // Subsequent turns: wait for the next message
+          // Subsequent turns: drain buffered mid-turn messages first (they
+          // were consumed and won't be re-delivered), then wait.
           if (turn > 0) {
-            // chat.requestUpgrade() / chat.endRun() — exit before waiting
-            if (locals.get(chatUpgradeRequestedKey) || locals.get(chatEndRunRequestedKey)) {
-              stop.cleanup();
-              return { done: true, value: undefined };
-            }
+            if (sessionPendingWire.length > 0) {
+              currentPayload = sessionPendingWire.shift()!;
+            } else {
+              // chat.requestUpgrade() / chat.endRun() — exit before waiting
+              if (locals.get(chatUpgradeRequestedKey) || locals.get(chatEndRunRequestedKey)) {
+                stop.cleanup();
+                return { done: true, value: undefined };
+              }
 
-            const next = await messagesInput.waitWithIdleTimeout({
-              idleTimeoutInSeconds,
-              timeout,
-              spanName: "waiting for next message",
-            });
-            if (!next.ok || runSignal.aborted) {
-              stop.cleanup();
-              return { done: true, value: undefined };
+              const next = await messagesInput.waitWithIdleTimeout({
+                idleTimeoutInSeconds,
+                timeout,
+                spanName: "waiting for next message",
+              });
+              if (!next.ok || runSignal.aborted) {
+                stop.cleanup();
+                return { done: true, value: undefined };
+              }
+              currentPayload = next.output;
             }
-            currentPayload = next.output;
           }
 
           // Check limits
@@ -9426,11 +9459,10 @@ function createChatSession(
           });
 
           // Listen for messages during streaming (steering + next-turn buffer)
-          const sessionPendingWire: ChatTaskWirePayload[] = [];
           const sessionMsgSub = messagesInput.on(async (msg) => {
-            sessionPendingWire.push(msg);
-
             if (sessionPendingMessages) {
+              // Steering route — the frontend re-sends non-injected
+              // messages on turn complete, so don't also buffer the wire.
               // Slim wire: at most one delta message per record. Read
               // `msg.message` directly — no array slicing needed.
               const lastUIMessage = msg.message;
@@ -9453,8 +9485,12 @@ function createChatSession(
                   /* non-fatal */
                 }
               }
+              return;
             }
+
+            sessionPendingWire.push(msg);
           });
+          activeMsgSub = sessionMsgSub;
 
           // Accumulate messages. Slim wire: pass the single delta message as
           // a 0-or-1-length array. The accumulator's behavior is unchanged —
@@ -9555,6 +9591,10 @@ function createChatSession(
                 } else {
                   throw error;
                 }
+              } finally {
+                // Detach at stream end (like the agent loop): the steering queue
+                // can't inject anymore, so later arrivals must buffer for the next turn.
+                sessionMsgSub.off();
               }
 
               if (response) {
@@ -9726,6 +9766,8 @@ function createChatSession(
         },
 
         async return() {
+          activeMsgSub?.off();
+          activeMsgSub = undefined;
           // `stop` only exists once next() has booted the iterator.
           stop?.cleanup();
           return { done: true, value: undefined };

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

@@ -753,11 +753,14 @@ export class SessionInputChannel {
                 : undefined;
 
             if (waitResult.ok) {
-              // Advance the seq counter so the SSE tail doesn't replay the
-              // record that was consumed via the waitpoint.
+              // Advance both cursors past the record consumed via the
+              // waitpoint: the seq counter so the SSE tail doesn't replay
+              // it, and the consume cursor so turn-completes don't stamp a
+              // stale `session-in-event-id`.
               const prevSeq = sessionStreams.lastSeqNum(this.sessionId, "in");
               const nextSeq = (prevSeq ?? -1) + 1;
               sessionStreams.setLastSeqNum(this.sessionId, "in", nextSeq);
+              sessionStreams.setLastDispatchedSeqNum(this.sessionId, "in", nextSeq);
 
               return { ok: true as const, output: data as T };
             } else {