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

.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 {