From 9a21f57b159bf3723c338ae22c3140e75990ba63 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Wed, 26 Aug 2026 20:00:03 +0100
Subject: [PATCH 01/11] fix(chat,sdk): hold the resume floor behind a message
 buffered mid-turn

A message arriving while a turn was streaming was handed to the 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 this process. A crash before the next turn lost it silently.

The handler is now attached only when there is a steering config to feed.
Without one the record stays queued on the router, which holds the floor
until a turn takes it, and both in-memory wire buffers go away. The wait
path already takes from the queue before suspending, so a message that
arrived mid-turn is still picked up as the next turn without a round trip.

The floor also doubles as the wake cursor, so an over-advanced floor parked
a waitpoint nothing would complete. It is now recorded on the wait span to
make that diagnosable from a trace.

Not addressed here: with a steering config, a declined message is still
dropped rather than left queued. That path depends on an unresolved
question about what declining should mean.
---
 .changeset/quiet-floors-hold.md               |   7 +
 packages/trigger-sdk/src/v3/ai.ts             | 234 ++++++++----------
 .../src/v3/test/test-session-handle.ts        |   4 +
 .../test/mid-turn-resume-floor.test.ts        | 120 +++++++++
 4 files changed, 237 insertions(+), 128 deletions(-)
 create mode 100644 .changeset/quiet-floors-hold.md
 create mode 100644 packages/trigger-sdk/test/mid-turn-resume-floor.test.ts

diff --git a/.changeset/quiet-floors-hold.md b/.changeset/quiet-floors-hold.md
new file mode 100644
index 00000000000..154da2ab52d
--- /dev/null
+++ b/.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.
diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts
index a44dd210cfe..c5272fc4865 100644
--- a/packages/trigger-sdk/src/v3/ai.ts
+++ b/packages/trigger-sdk/src/v3/ai.ts
@@ -1573,9 +1573,18 @@ async function waitOnChatRoute<T>(
 
       span.setAttribute("wait.resolved", "suspended");
       while (true) {
+        /**
+         * The floor doubles as the wake cursor: the server completes the
+         * waitpoint immediately if anything sits after this sequence, so a
+         * floor that has advanced past an unread record parks a waitpoint
+         * nothing will complete. Recorded on the span so a run that never woke
+         * can be diagnosed from its trace alone.
+         */
+        const wakeFrom = router.resumeFloor();
+        span.setAttribute("wait.lastSeqNum", wakeFrom ?? -1);
         const wake = await session.in.awaitWake({
           timeout: options.timeout,
-          lastSeqNum: router.resumeFloor(),
+          lastSeqNum: wakeFrom,
         });
         if (!wake.ok) {
           span.recordException(wake.error);
@@ -5802,14 +5811,6 @@ 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
@@ -6705,57 +6706,56 @@ function chatAgent<
                 const combinedSignal = AbortSignal.any([runSignal, stopController.signal]);
 
                 const pmConfig = locals.get(chatPendingMessagesKey);
-                const msgSub = messagesInput.on(async (msg) => {
-                  // If pendingMessages is configured, route to the steering queue
-                  // instead of the wire buffer. The frontend handles re-sending
-                  // non-injected messages via sendMessage on turn complete.
-                  if (pmConfig) {
-                    // Slim wire: at most one delta message per record. The
-                    // pendingMessages handler reads `msg.message` directly
-                    // instead of slicing an array — a wire record arrives
-                    // with the new user message in `.message`, or no message
-                    // at all (regenerate / preload / close / handover-prepare).
-                    const lastUIMessage = msg.message as TUIMessage | undefined;
-                    if (lastUIMessage) {
-                      if (pmConfig.onReceived) {
+                /**
+                 * Only attached when there is a steering config to feed. Without
+                 * one a mid-turn message is left queued on the router, which is
+                 * what holds the resume floor behind it: a record handed to a
+                 * handler counts as terminally decided, so buffering one here
+                 * published a cursor past a message held only in memory and a
+                 * crash before the next turn lost it.
+                 */
+                const msgSub = pmConfig
+                  ? messagesInput.on(async (msg) => {
+                      // Slim wire: at most one delta message per record. The
+                      // pendingMessages handler reads `msg.message` directly
+                      // instead of slicing an array — a wire record arrives
+                      // with the new user message in `.message`, or no message
+                      // at all (regenerate / preload / close / handover-prepare).
+                      const lastUIMessage = msg.message as TUIMessage | undefined;
+                      if (lastUIMessage) {
+                        if (pmConfig.onReceived) {
+                          try {
+                            await pmConfig.onReceived({
+                              message: lastUIMessage as TUIMessage,
+                              chatId: currentWirePayload.chatId,
+                              turn,
+                            });
+                          } catch {
+                            /* non-fatal */
+                          }
+                        }
+
                         try {
-                          await pmConfig.onReceived({
-                            message: lastUIMessage as TUIMessage,
-                            chatId: currentWirePayload.chatId,
-                            turn,
+                          const queue = locals.get(chatSteeringQueueKey) ?? [];
+                          // Deduplicate by message ID — guards against double-sends
+                          if (
+                            lastUIMessage.id &&
+                            queue.some((e) => e.uiMessage.id === lastUIMessage.id)
+                          ) {
+                            return;
+                          }
+                          const modelMsgs = await toModelMessages([lastUIMessage]);
+                          queue.push({
+                            uiMessage: lastUIMessage as UIMessage,
+                            modelMessages: modelMsgs,
                           });
+                          locals.set(chatSteeringQueueKey, queue);
                         } catch {
-                          /* non-fatal */
-                        }
-                      }
-
-                      try {
-                        const queue = locals.get(chatSteeringQueueKey) ?? [];
-                        // Deduplicate by message ID — guards against double-sends
-                        if (
-                          lastUIMessage.id &&
-                          queue.some((e) => e.uiMessage.id === lastUIMessage.id)
-                        ) {
-                          return;
+                          /* conversion failed — skip steering queue */
                         }
-                        const modelMsgs = await toModelMessages([lastUIMessage]);
-                        queue.push({
-                          uiMessage: lastUIMessage as UIMessage,
-                          modelMessages: modelMsgs,
-                        });
-                        locals.set(chatSteeringQueueKey, queue);
-                      } catch {
-                        /* conversion failed — skip steering queue */
                       }
-                    }
-                    return; // Don't add to wire buffer — frontend handles non-injected case
-                  }
-
-                  // No pendingMessages config — standard wire buffer for next turn
-                  pendingWireMessages.push(
-                    msg as ChatTaskWirePayload<TUIMessage, inferSchemaIn<TClientDataSchema>>
-                  );
-                });
+                    })
+                  : undefined;
                 turnMsgSub = msgSub;
 
                 // Track new messages for this turn (user input + assistant response).
@@ -7120,7 +7120,7 @@ function chatAgent<
                 // The turn counter is decremented so the next iteration
                 // sees the same `turn` value — actions don't count.
                 if (isAction) {
-                  msgSub.off();
+                  msgSub?.off();
 
                   if (
                     (locals.get(chatPipeCountKey) ?? 0) === 0 &&
@@ -7410,7 +7410,7 @@ function chatAgent<
                       throw error;
                     }
                   } finally {
-                    msgSub.off();
+                    msgSub?.off();
                   }
 
                   // Wait for onFinish to fire — on abort this may resolve slightly
@@ -7966,14 +7966,6 @@ function chatAgent<
                   return "continue";
                 }
 
-                // If messages arrived during streaming (without pendingMessages config),
-                // 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";
-                }
-
                 // chat.requestUpgrade() was called — exit the loop so the
                 // transport triggers a new run on the latest version.
                 // chat.endRun() — same exit, no upgrade semantics.
@@ -8282,12 +8274,6 @@ 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) ??
@@ -9775,10 +9761,6 @@ 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;
@@ -9857,29 +9839,28 @@ function createChatSession(
             }
           }
 
-          // Subsequent turns: drain buffered mid-turn messages first (they
-          // were consumed and won't be re-delivered), then wait.
+          /**
+           * Subsequent turns take the next message from the router. A record
+           * that arrived mid-turn is already queued there, so this returns it
+           * without suspending.
+           */
           if (turn > 0) {
-            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 };
-              }
+            // 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 };
-              }
-              currentPayload = next.output;
+            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;
           }
 
           // Check limits
@@ -9907,38 +9888,35 @@ function createChatSession(
             clientData: currentPayload.metadata,
           });
 
-          // Listen for messages during streaming (steering + next-turn buffer)
-          const sessionMsgSub = messagesInput.on(async (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;
-              if (lastUIMessage) {
-                if (sessionPendingMessages.onReceived) {
+          /**
+           * Only attached when there is a steering config to feed. Without one a
+           * mid-turn message stays queued on the router, which is what keeps the
+           * resume floor behind it.
+           */
+          const sessionMsgSub = sessionPendingMessages
+            ? messagesInput.on(async (msg) => {
+                const lastUIMessage = msg.message;
+                if (lastUIMessage) {
+                  if (sessionPendingMessages.onReceived) {
+                    try {
+                      await sessionPendingMessages.onReceived({
+                        message: lastUIMessage,
+                        chatId: currentPayload.chatId,
+                        turn,
+                      });
+                    } catch {
+                      /* non-fatal */
+                    }
+                  }
                   try {
-                    await sessionPendingMessages.onReceived({
-                      message: lastUIMessage,
-                      chatId: currentPayload.chatId,
-                      turn,
-                    });
+                    const modelMsgs = await toModelMessages([lastUIMessage]);
+                    turnSteeringQueue.push({ uiMessage: lastUIMessage, modelMessages: modelMsgs });
                   } catch {
                     /* non-fatal */
                   }
                 }
-                try {
-                  const modelMsgs = await toModelMessages([lastUIMessage]);
-                  turnSteeringQueue.push({ uiMessage: lastUIMessage, modelMessages: modelMsgs });
-                } catch {
-                  /* non-fatal */
-                }
-              }
-              return;
-            }
-
-            sessionPendingWire.push(msg);
-          });
+              })
+            : undefined;
           activeMsgSub = sessionMsgSub;
 
           // Accumulate messages. Slim wire: pass the single delta message as
@@ -9967,7 +9945,7 @@ function createChatSession(
           // chat.requestUpgrade() called before this turn — signal transport and exit
           if (locals.get(chatUpgradeRequestedKey)) {
             await writeUpgradeRequiredChunk();
-            sessionMsgSub.off();
+            sessionMsgSub?.off();
             stop.cleanup();
             return { done: true, value: undefined };
           }
@@ -10013,7 +9991,7 @@ function createChatSession(
                 if (!response || response.role !== "assistant") {
                   throw new Error("turn.complete() could not find the spliced handover response");
                 }
-                sessionMsgSub.off();
+                sessionMsgSub?.off();
                 await chatWriteTurnComplete();
                 return response;
               }
@@ -10030,7 +10008,7 @@ function createChatSession(
                 });
                 if (runSignal.aborted) {
                   // Full cancel — don't accumulate
-                  sessionMsgSub.off();
+                  sessionMsgSub?.off();
                   await chatWriteTurnComplete();
                   return undefined;
                 }
@@ -10052,7 +10030,7 @@ function createChatSession(
               } 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();
+                sessionMsgSub?.off();
               }
 
               if (response) {
@@ -10155,7 +10133,7 @@ function createChatSession(
                 }
               }
 
-              sessionMsgSub.off();
+              sessionMsgSub?.off();
               await chatWriteTurnComplete();
               return response;
             },
@@ -10174,7 +10152,7 @@ function createChatSession(
             },
 
             async done() {
-              sessionMsgSub.off();
+              sessionMsgSub?.off();
               await chatWriteTurnComplete();
             },
 
diff --git a/packages/trigger-sdk/src/v3/test/test-session-handle.ts b/packages/trigger-sdk/src/v3/test/test-session-handle.ts
index 5150de87da3..c98bf92f8f8 100644
--- a/packages/trigger-sdk/src/v3/test/test-session-handle.ts
+++ b/packages/trigger-sdk/src/v3/test/test-session-handle.ts
@@ -259,6 +259,10 @@ class TestSessionOutputChannel extends SessionOutputChannel {
       for (const [name, value] of extraHeaders) {
         if (name === "public-access-token") {
           synthetic.publicAccessToken = value;
+        } else if (name === "session-in-event-id") {
+          synthetic.sessionInEventId = value;
+        } else if (name === "session-in-consumed-id") {
+          synthetic.sessionInConsumedId = value;
         }
       }
     }
diff --git a/packages/trigger-sdk/test/mid-turn-resume-floor.test.ts b/packages/trigger-sdk/test/mid-turn-resume-floor.test.ts
new file mode 100644
index 00000000000..2f2f12e9c88
--- /dev/null
+++ b/packages/trigger-sdk/test/mid-turn-resume-floor.test.ts
@@ -0,0 +1,120 @@
+import { mockChatAgent } from "../src/v3/test/index.js";
+
+import { sessionStreams } from "@trigger.dev/core/v3";
+import { simulateReadableStream, streamText } from "ai";
+import { MockLanguageModelV3 } from "ai/test";
+import type { LanguageModelV3StreamPart } from "@ai-sdk/provider";
+import { describe, expect, it } from "vitest";
+import { chat } from "../src/v3/ai.js";
+
+function userMessage(text: string, id: string) {
+  return { id, role: "user" as const, parts: [{ type: "text" as const, text }] };
+}
+
+function textStreamChunks(text: string): LanguageModelV3StreamPart[] {
+  return [
+    { type: "text-start", id: "t1" },
+    { type: "text-delta", id: "t1", delta: text },
+    { type: "text-end", id: "t1" },
+    {
+      type: "finish",
+      finishReason: { unified: "stop", raw: "stop" },
+      usage: {
+        inputTokens: { total: 1, noCache: 1, cacheRead: undefined, cacheWrite: undefined },
+        outputTokens: { total: 1, text: 1, reasoning: undefined },
+      },
+    },
+  ];
+}
+
+/** Answers `ANSWER(<last user text>)` slowly enough for a record sent after the
+ * turn starts to arrive mid-stream. */
+function echoModel() {
+  return new MockLanguageModelV3({
+    doStream: async ({ prompt }) => {
+      const users = prompt.filter((m) => m.role === "user");
+      const last = users[users.length - 1];
+      const text = Array.isArray(last?.content)
+        ? last.content
+            .filter((p): p is { type: "text"; text: string } => p.type === "text")
+            .map((p) => p.text)
+            .join("")
+        : "";
+      return {
+        stream: simulateReadableStream({
+          chunks: textStreamChunks(`ANSWER(${text})`),
+          initialDelayInMs: 150,
+          chunkDelayInMs: 10,
+        }),
+      };
+    },
+  });
+}
+
+function streamedText(harness: { allChunks: unknown[] }): string {
+  return (harness.allChunks as { type?: string; delta?: string }[])
+    .filter((c) => c.type === "text-delta")
+    .map((c) => c.delta ?? "")
+    .join("");
+}
+
+function turnCompletes(harness: { allRawChunks: unknown[] }) {
+  return (harness.allRawChunks as { type?: string; sessionInEventId?: string }[]).filter(
+    (c) => c.type === "trigger:turn-complete"
+  );
+}
+
+async function waitFor(check: () => boolean, timeoutMs = 10_000) {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (check()) return;
+    await new Promise((r) => setTimeout(r, 20));
+  }
+  throw new Error("waitFor timed out");
+}
+
+/**
+ * A message that arrives while a turn is streaming, on an agent with no
+ * `pendingMessages` config, is delivered to the turn's push handler and parked
+ * in the in-memory wire buffer to become the next turn. The router treats a
+ * record handed to a push handler as terminally decided, so it stops holding
+ * the resume floor behind it, and the turn boundary then publishes a cursor
+ * past a message that exists only in this process's memory.
+ *
+ * The cursor is the contract for what a later boot may skip. Publishing one
+ * past an unanswered message means a crash between that boundary and the next
+ * turn loses the message with nothing raised.
+ */
+describe("chat.agent resume floor with a message buffered mid-turn", () => {
+  it("does not publish a resume cursor past a message that has not been answered", async () => {
+    const chatId = "mid-turn-resume-floor";
+    const agent = chat.agent({
+      id: "mid-turn-resume-floor.agent",
+      run: async ({ messages, signal }) =>
+        streamText({ model: echoModel(), messages, abortSignal: signal }),
+    });
+
+    const harness = mockChatAgent(agent, { chatId });
+    try {
+      const firstTurn = harness.sendMessage(userMessage("m1", "u-1"));
+      await waitFor(() => streamedText(harness).includes("ANSWER(m1)"));
+
+      void harness.sendMessage(userMessage("m2", "u-2"));
+      await waitFor(
+        () => (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in") !== undefined
+      );
+      const m2Seq = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in")!;
+      await firstTurn;
+
+      await waitFor(() => turnCompletes(harness).length >= 1);
+      const firstBoundary = turnCompletes(harness)[0]!;
+      expect(firstBoundary.sessionInEventId).toBeDefined();
+
+      expect(Number(firstBoundary.sessionInEventId)).toBeLessThan(m2Seq);
+    } finally {
+      await harness.close();
+    }
+  });
+});
+
+type SeqReader = { lastSeqNum(sessionId: string, io: "in" | "out"): number | undefined };

From 383d169e78982c2f402948c18576a6a3d5fc820e Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Wed, 26 Aug 2026 20:15:21 +0100
Subject: [PATCH 02/11] fix(chat,core): defer a declined steering message
 instead of dropping it

A message arriving mid-turn with a `pendingMessages` config was routed into a
turn-local steering queue. If the batch was declined for injection it was
discarded with the turn: never injected, never written to the wire buffer,
never answered, and nothing raised at either end. Declining is also the
default, since a config without `shouldInject` declines every batch, so the
documented default behaviour was the losing one.

Notification and consumption are now separate. `observe` on the router tells a
consumer a record arrived without taking it, so the record stays queued and
keeps holding the resume floor, and injection is the point of consumption:
`take` removes exactly the records that were injected. A declined batch never
reaches that line, so its records stay queued and become later turns, which is
what the docs have always promised.

`observe` 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 the wedged mailbox, or watch records it cannot
affect.
---
 .changeset/spry-steers-defer.md               | 20 +++++
 .../core/src/v3/sessionStreams/router.test.ts | 90 +++++++++++++++++++
 packages/core/src/v3/sessionStreams/router.ts | 58 ++++++++++++
 packages/trigger-sdk/src/v3/ai.ts             | 39 ++++++--
 .../test/pending-message-drain.test.ts        | 42 +++++++++
 5 files changed, 244 insertions(+), 5 deletions(-)
 create mode 100644 .changeset/spry-steers-defer.md

diff --git a/.changeset/spry-steers-defer.md b/.changeset/spry-steers-defer.md
new file mode 100644
index 00000000000..f6603b84b5c
--- /dev/null
+++ b/.changeset/spry-steers-defer.md
@@ -0,0 +1,20 @@
+---
+"@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({ ... }),
+});
+```
+
+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.
diff --git a/packages/core/src/v3/sessionStreams/router.test.ts b/packages/core/src/v3/sessionStreams/router.test.ts
index 6f654269d1c..4a74bf7c1bc 100644
--- a/packages/core/src/v3/sessionStreams/router.test.ts
+++ b/packages/core/src/v3/sessionStreams/router.test.ts
@@ -407,3 +407,93 @@ 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)).toBe(true);
+    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)).toBe(true);
+    expect(r.take("messages", 0)).toBe(false);
+    expect(r.take("messages", 99)).toBe(false);
+  });
+
+  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);
+  });
+});
diff --git a/packages/core/src/v3/sessionStreams/router.ts b/packages/core/src/v3/sessionStreams/router.ts
index 17b4cdaddb2..1840fb1fb86 100644
--- a/packages/core/src/v3/sessionStreams/router.ts
+++ b/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,54 @@ 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): boolean {
+    const state = this.#stateOrThrow(name);
+    const index = state.queue.findIndex((record) => record.seqNum === seqNum);
+    if (index === -1) return false;
+    state.queue.splice(index, 1);
+    return true;
+  }
+
   /** Whether an `at-arrival` route currently has anywhere to deliver. */
   hasHandler(name: string): boolean {
     return this.#stateOrThrow(name).handlers.size > 0;
@@ -430,6 +487,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);
diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts
index c5272fc4865..6840b50ef29 100644
--- a/packages/trigger-sdk/src/v3/ai.ts
+++ b/packages/trigger-sdk/src/v3/ai.ts
@@ -3007,7 +3007,17 @@ export { PENDING_MESSAGE_INJECTED_TYPE, upsertIncomingMessage };
 export type { InferChatClientData, InferChatUIMessage, InferChatUIMessageFromTools };
 
 /** @internal */
-type SteeringQueueEntry = { uiMessage: UIMessage; modelMessages: ModelMessage[] };
+/**
+ * A message observed mid-turn and offered for injection. `seqNum` is the record
+ * it came from, so injecting can take it off the router; a batch that is
+ * declined leaves its records queued and they become later turns instead.
+ * Absent for entries that did not come from a channel record.
+ */
+type SteeringQueueEntry = {
+  uiMessage: UIMessage;
+  modelMessages: ModelMessage[];
+  seqNum?: number;
+};
 /** @internal */
 const chatPendingMessagesKey = locals.create<PendingMessagesOptions>("chat.pendingMessages");
 /** @internal */
@@ -3650,7 +3660,18 @@ async function drainSteeringQueue(
         ? await config.prepare(batchEvent)
         : queue.flatMap((e) => e.modelMessages);
 
-      // Clear the queue and record injected IDs
+      /**
+       * Injection is the point of consumption. The records were only observed
+       * on arrival, so they are still queued on the router and still holding
+       * the resume floor; taking them here is what stops the same message also
+       * being answered as a later turn. A batch that was declined never reaches
+       * this line, so its records stay queued and become later turns, which is
+       * what "messages queue for the next turn" means.
+       */
+      const router = chatInputRouter();
+      for (const entry of queue) {
+        if (entry.seqNum !== undefined) router.take(CHAT_ROUTE_MESSAGES, entry.seqNum);
+      }
       queue.length = 0;
       const injectedIds = locals.get(chatInjectedMessageIdsKey);
       if (injectedIds) {
@@ -6715,7 +6736,9 @@ function chatAgent<
                  * crash before the next turn lost it.
                  */
                 const msgSub = pmConfig
-                  ? messagesInput.on(async (msg) => {
+                  ? chatInputRouter().observe(CHAT_ROUTE_MESSAGES, async (record) => {
+                      const msg = (record.data as Extract<ChatInputChunk, { kind: "message" }>)
+                        .payload;
                       // Slim wire: at most one delta message per record. The
                       // pendingMessages handler reads `msg.message` directly
                       // instead of slicing an array — a wire record arrives
@@ -6748,6 +6771,7 @@ function chatAgent<
                           queue.push({
                             uiMessage: lastUIMessage as UIMessage,
                             modelMessages: modelMsgs,
+                            seqNum: record.seqNum,
                           });
                           locals.set(chatSteeringQueueKey, queue);
                         } catch {
@@ -9894,7 +9918,8 @@ function createChatSession(
            * resume floor behind it.
            */
           const sessionMsgSub = sessionPendingMessages
-            ? messagesInput.on(async (msg) => {
+            ? chatInputRouter().observe(CHAT_ROUTE_MESSAGES, async (record) => {
+                const msg = (record.data as Extract<ChatInputChunk, { kind: "message" }>).payload;
                 const lastUIMessage = msg.message;
                 if (lastUIMessage) {
                   if (sessionPendingMessages.onReceived) {
@@ -9910,7 +9935,11 @@ function createChatSession(
                   }
                   try {
                     const modelMsgs = await toModelMessages([lastUIMessage]);
-                    turnSteeringQueue.push({ uiMessage: lastUIMessage, modelMessages: modelMsgs });
+                    turnSteeringQueue.push({
+                      uiMessage: lastUIMessage,
+                      modelMessages: modelMsgs,
+                      seqNum: record.seqNum,
+                    });
                   } catch {
                     /* non-fatal */
                   }
diff --git a/packages/trigger-sdk/test/pending-message-drain.test.ts b/packages/trigger-sdk/test/pending-message-drain.test.ts
index 44a2054a64e..1bd286c54d4 100644
--- a/packages/trigger-sdk/test/pending-message-drain.test.ts
+++ b/packages/trigger-sdk/test/pending-message-drain.test.ts
@@ -182,6 +182,48 @@ describe("chat.agent steering config", () => {
   });
 });
 
+/**
+ * A mid-turn message the agent declines to inject is documented to "queue for
+ * the next turn". It used to be diverted into a turn-local steering queue and
+ * discarded with the turn, so it was never injected, never written to the wire
+ * buffer, and never answered, with nothing raised at either end. Declining is
+ * also the default: with a `pendingMessages` config and no `shouldInject`, the
+ * callback is treated as returning false for every batch.
+ */
+describe("chat.agent declined steering message", () => {
+  it("answers a declined mid-turn message as its own turn", async () => {
+    const received: string[] = [];
+
+    const agent = chat.agent({
+      id: "pending-drain.declined",
+      pendingMessages: {
+        onReceived: ({ message }) => {
+          received.push(message.id);
+        },
+        shouldInject: () => false,
+      },
+      run: async ({ messages, signal }) => {
+        return streamText({ model: echoModel(), messages, abortSignal: signal });
+      },
+    });
+
+    const harness = mockChatAgent(agent, { chatId: "pending-drain-declined" });
+    try {
+      const first = harness.sendMessage(userMessage("m1", "u-1"));
+      await waitFor(() => streamedText(harness).includes("ANSWER(m1)"));
+      void harness.sendMessage(userMessage("m2", "u-2"));
+      await first;
+
+      await waitFor(() => turnCompleteCount(harness) >= 2);
+
+      expect(received).toContain("u-2");
+      expect(streamedText(harness)).toContain("ANSWER(m2)");
+    } finally {
+      await harness.close();
+    }
+  });
+});
+
 describe("chat.agent errored turn", () => {
   it(
     "does not duplicate messages buffered after a turn that threw",

From 3dd60c20106ae07ec16db2053162a8fade5302e6 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Wed, 26 Aug 2026 20:20:21 +0100
Subject: [PATCH 03/11] docs(ai-chat): say what happens to a message that is
 not injected

The docs described a mid-turn message becoming the next turn only when there
were no more step boundaries, and the client-side lifecycle credited the
frontend with auto-sending it. Neither matched the behaviour: a message the
agent declines to inject is now held on the backend and answered as the next
turn, with no client re-send involved, and that covers an explicit
`shouldInject: false` as well as a turn that never reaches a boundary.

Also spells out that a declined message keeps its place in the queue, so it
survives a crash rather than living only in the worker that received it.
---
 docs/ai-chat/client-protocol.mdx  | 5 ++++-
 docs/ai-chat/pending-messages.mdx | 6 +++---
 docs/ai-chat/reference.mdx        | 2 +-
 3 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/docs/ai-chat/client-protocol.mdx b/docs/ai-chat/client-protocol.mdx
index d039b39366a..a9acb1614f0 100644
--- a/docs/ai-chat/client-protocol.mdx
+++ b/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>
diff --git a/docs/ai-chat/pending-messages.mdx b/docs/ai-chat/pending-messages.mdx
index c77600f470a..6a7c89cd580 100644
--- a/docs/ai-chat/pending-messages.mdx
+++ b/docs/ai-chat/pending-messages.mdx
@@ -10,7 +10,7 @@ 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). Nothing is lost either way, and the backend handles it, so no client-side re-send is involved.
 
 ## How it works
 
@@ -20,7 +20,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 +310,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".
 
diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx
index dd444961af9..05b509215ef 100644
--- a/docs/ai-chat/reference.mdx
+++ b/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. |
 | `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.                                         |

From 87e45776361c9a9863a60d105c5a87f29dbe15e5 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Thu, 27 Aug 2026 09:23:14 +0100
Subject: [PATCH 04/11] fix(chat,sdk): consume only the steering batch the
 callbacks were given

`shouldInject` and `prepare` can await, so a record could arrive between the
batch being assembled and the injection being applied. That record was not in
the batch and the callbacks never saw it, but it was taken and cleared along
with them, so a message that should have become a later turn was discarded.

The batch is now snapshotted before the callbacks run, and only its entries
are taken from the router and removed from the queue.

Also scopes the deferral guarantee in the docs and the changeset: it holds
only when `pendingMessages` actually reaches `streamText`, via
`chat.toStreamTextOptions()` or an explicit `prepareStep`. A config without
that wiring has nothing to drain the queue and still loses messages.

The resume-floor test now waits for the channel sequence to advance rather
than to merely exist, since the first message had already advanced it and the
old predicate could capture that sequence instead of the second message's.
---
 .changeset/spry-steers-defer.md                 |  9 ++++++++-
 docs/ai-chat/pending-messages.mdx               |  4 +++-
 docs/ai-chat/reference.mdx                      |  2 +-
 packages/trigger-sdk/src/v3/ai.ts               | 16 ++++++++++++----
 .../test/mid-turn-resume-floor.test.ts          | 17 +++++++++++++----
 5 files changed, 37 insertions(+), 11 deletions(-)

diff --git a/.changeset/spry-steers-defer.md b/.changeset/spry-steers-defer.md
index f6603b84b5c..49968d3f9ef 100644
--- a/.changeset/spry-steers-defer.md
+++ b/.changeset/spry-steers-defer.md
@@ -13,7 +13,14 @@ chat.agent({
     // Only interrupt once the agent has started calling tools.
     shouldInject: ({ steps }) => steps.length > 0,
   },
-  run: async ({ messages, signal }) => streamText({ ... }),
+  run: async ({ messages, signal }) =>
+    streamText({
+      model,
+      messages,
+      abortSignal: signal,
+      // Required for pendingMessages to be wired up at all.
+      ...chat.toStreamTextOptions(),
+    }),
 });
 ```
 
diff --git a/docs/ai-chat/pending-messages.mdx b/docs/ai-chat/pending-messages.mdx
index 6a7c89cd580..2bbbb83d7a9 100644
--- a/docs/ai-chat/pending-messages.mdx
+++ b/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. 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). Nothing is lost either way, and the backend handles it, so no client-side re-send is involved.
+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.
+
+This requires the `pendingMessages` options to actually reach `streamText`, by spreading `chat.toStreamTextOptions()` (or passing `prepareStep`). Configuring `pendingMessages` without that wiring leaves nothing to drain the queue, and mid-turn messages are lost.
 
 ## How it works
 
diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx
index 05b509215ef..518566f4a1d 100644
--- a/docs/ai-chat/reference.mdx
+++ b/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, nothing is injected and the messages are answered as the next turn. |
+| `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. Requires `chat.toStreamTextOptions()` (or `prepareStep`) to be passed to `streamText`. |
 | `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.                                         |
diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts
index 6840b50ef29..4bf6e8fd9a9 100644
--- a/packages/trigger-sdk/src/v3/ai.ts
+++ b/packages/trigger-sdk/src/v3/ai.ts
@@ -3624,7 +3624,14 @@ async function drainSteeringQueue(
 
   const ctx = locals.get(chatTurnContextKey);
   const stepNumber = steps.length - 1;
-  const uiMessages = queue.map((e) => e.uiMessage);
+  /**
+   * Snapshot, because `shouldInject` and `prepare` can await. A record arriving
+   * during either is not in this batch, so it must not be consumed by it: the
+   * callbacks never saw it and could not have injected it, and taking it would
+   * lose a message that should become a later turn instead.
+   */
+  const batch = [...queue];
+  const uiMessages = batch.map((e) => e.uiMessage);
 
   const batchEvent: PendingMessagesBatchEvent = {
     messages: uiMessages,
@@ -3658,7 +3665,7 @@ async function drainSteeringQueue(
       // Transform the batch — default: concatenate all pre-converted model messages
       const injected = config.prepare
         ? await config.prepare(batchEvent)
-        : queue.flatMap((e) => e.modelMessages);
+        : batch.flatMap((e) => e.modelMessages);
 
       /**
        * Injection is the point of consumption. The records were only observed
@@ -3669,10 +3676,11 @@ async function drainSteeringQueue(
        * what "messages queue for the next turn" means.
        */
       const router = chatInputRouter();
-      for (const entry of queue) {
+      for (const entry of batch) {
         if (entry.seqNum !== undefined) router.take(CHAT_ROUTE_MESSAGES, entry.seqNum);
+        const at = queue.indexOf(entry);
+        if (at !== -1) queue.splice(at, 1);
       }
-      queue.length = 0;
       const injectedIds = locals.get(chatInjectedMessageIdsKey);
       if (injectedIds) {
         for (const m of uiMessages) injectedIds.add(m.id);
diff --git a/packages/trigger-sdk/test/mid-turn-resume-floor.test.ts b/packages/trigger-sdk/test/mid-turn-resume-floor.test.ts
index 2f2f12e9c88..742c6c7f436 100644
--- a/packages/trigger-sdk/test/mid-turn-resume-floor.test.ts
+++ b/packages/trigger-sdk/test/mid-turn-resume-floor.test.ts
@@ -99,11 +99,20 @@ describe("chat.agent resume floor with a message buffered mid-turn", () => {
       const firstTurn = harness.sendMessage(userMessage("m1", "u-1"));
       await waitFor(() => streamedText(harness).includes("ANSWER(m1)"));
 
+      /**
+       * m1's own record has already advanced the channel, so waiting for a
+       * sequence to merely exist would pass on the first check and capture m1's
+       * sequence instead of m2's. Wait for it to move.
+       */
+      const seqs = sessionStreams as unknown as SeqReader;
+      const seqBeforeM2 = seqs.lastSeqNum(chatId, "in");
       void harness.sendMessage(userMessage("m2", "u-2"));
-      await waitFor(
-        () => (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in") !== undefined
-      );
-      const m2Seq = (sessionStreams as unknown as SeqReader).lastSeqNum(chatId, "in")!;
+      await waitFor(() => {
+        const now = seqs.lastSeqNum(chatId, "in");
+        return now !== undefined && (seqBeforeM2 === undefined || now > seqBeforeM2);
+      });
+      const m2Seq = seqs.lastSeqNum(chatId, "in")!;
+      expect(m2Seq).toBeGreaterThan(seqBeforeM2 ?? -1);
       await firstTurn;
 
       await waitFor(() => turnCompletes(harness).length >= 1);

From 66c83c8c8828f2df7897811b3f2c4debd5c6549b Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Thu, 27 Aug 2026 09:26:14 +0100
Subject: [PATCH 05/11] fix(core): refuse to clear a replayable route's queue

`clearRoute` discards a queue outright. That is right for the handover route,
whose window closes at a turn boundary, and wrong for any replayable route,
where anything queued is still owed to a later boot and the resume floor is
deliberately held behind it.

Nothing calls it that way today, but the messages queue now holds real
unanswered user input rather than being drained into an in-memory buffer, so a
future caller would silently discard messages instead of merely losing a
window. Enforced rather than left as a comment.
---
 .../core/src/v3/sessionStreams/router.test.ts | 19 +++++++++++++++++++
 packages/core/src/v3/sessionStreams/router.ts |  5 +++++
 2 files changed, 24 insertions(+)

diff --git a/packages/core/src/v3/sessionStreams/router.test.ts b/packages/core/src/v3/sessionStreams/router.test.ts
index 4a74bf7c1bc..eef45e913b2 100644
--- a/packages/core/src/v3/sessionStreams/router.test.ts
+++ b/packages/core/src/v3/sessionStreams/router.test.ts
@@ -497,3 +497,22 @@ describe("SessionChannelRouter: take", () => {
     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);
+  });
+});
diff --git a/packages/core/src/v3/sessionStreams/router.ts b/packages/core/src/v3/sessionStreams/router.ts
index 1840fb1fb86..fadf144cb93 100644
--- a/packages/core/src/v3/sessionStreams/router.ts
+++ b/packages/core/src/v3/sessionStreams/router.ts
@@ -474,6 +474,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);

From 02aa30d5a686b92fc15afc65503c8a983fadf927 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Thu, 27 Aug 2026 09:52:20 +0100
Subject: [PATCH 06/11] fix(chat): correct the wiring caveat, deferral does not
 need the spread

The previous commit said a `pendingMessages` config without
`chat.toStreamTextOptions()` still loses mid-turn messages. That was true
before this branch and is not true on it: the arrival path only observes now,
so the record stays queued on the channel whatever happens to the steering
queue, and the next turn takes it.

Injection is the part that needs the spread. Without it nothing injects, so
every mid-turn message is answered as the next turn, which is the documented
default rather than a loss.

Covers the shape a developer reaches by following the docs for `onReceived`
alone, with no `shouldInject` and no spread, so nothing can drain the queue.
---
 .changeset/spry-steers-defer.md               |  3 +-
 docs/ai-chat/pending-messages.mdx             |  2 +-
 docs/ai-chat/reference.mdx                    |  2 +-
 .../test/pending-message-drain.test.ts        | 40 +++++++++++++++++++
 4 files changed, 44 insertions(+), 3 deletions(-)

diff --git a/.changeset/spry-steers-defer.md b/.changeset/spry-steers-defer.md
index 49968d3f9ef..01a85c8bb15 100644
--- a/.changeset/spry-steers-defer.md
+++ b/.changeset/spry-steers-defer.md
@@ -18,7 +18,8 @@ chat.agent({
       model,
       messages,
       abortSignal: signal,
-      // Required for pendingMessages to be wired up at all.
+      // Required for injection. Without it nothing injects, and every
+      // mid-turn message is answered as the next turn instead.
       ...chat.toStreamTextOptions(),
     }),
 });
diff --git a/docs/ai-chat/pending-messages.mdx b/docs/ai-chat/pending-messages.mdx
index 2bbbb83d7a9..a4dec357a47 100644
--- a/docs/ai-chat/pending-messages.mdx
+++ b/docs/ai-chat/pending-messages.mdx
@@ -12,7 +12,7 @@ By default (without `pendingMessages`), a message sent while the agent is respon
 
 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.
 
-This requires the `pendingMessages` options to actually reach `streamText`, by spreading `chat.toStreamTextOptions()` (or passing `prepareStep`). Configuring `pendingMessages` without that wiring leaves nothing to drain the queue, and mid-turn messages are lost.
+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
 
diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx
index 518566f4a1d..1b9a67d572a 100644
--- a/docs/ai-chat/reference.mdx
+++ b/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, nothing is injected and the messages are answered as the next turn. Requires `chat.toStreamTextOptions()` (or `prepareStep`) to be passed to `streamText`. |
+| `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.                                         |
diff --git a/packages/trigger-sdk/test/pending-message-drain.test.ts b/packages/trigger-sdk/test/pending-message-drain.test.ts
index 1bd286c54d4..c15a0903acb 100644
--- a/packages/trigger-sdk/test/pending-message-drain.test.ts
+++ b/packages/trigger-sdk/test/pending-message-drain.test.ts
@@ -224,6 +224,46 @@ describe("chat.agent declined steering message", () => {
   });
 });
 
+/**
+ * The shape a developer reaches by following the docs for `onReceived` alone:
+ * a `pendingMessages` config, no `shouldInject`, and no
+ * `chat.toStreamTextOptions()` spread, so nothing can ever drain the steering
+ * queue. The message must still be answered. It used to be taken off the
+ * channel by the arrival handler and then stranded in a queue with no consumer.
+ */
+describe("chat.agent pendingMessages with nothing wired to drain it", () => {
+  it("still answers a mid-turn message as its own turn", async () => {
+    const received: string[] = [];
+
+    const agent = chat.agent({
+      id: "pending-drain.unwired",
+      pendingMessages: {
+        onReceived: ({ message }) => {
+          received.push(message.id);
+        },
+      },
+      run: async ({ messages, signal }) => {
+        return streamText({ model: echoModel(), messages, abortSignal: signal });
+      },
+    });
+
+    const harness = mockChatAgent(agent, { chatId: "pending-drain-unwired" });
+    try {
+      const first = harness.sendMessage(userMessage("m1", "u-1"));
+      await waitFor(() => streamedText(harness).includes("ANSWER(m1)"));
+      void harness.sendMessage(userMessage("m2", "u-2"));
+      await first;
+
+      await waitFor(() => turnCompleteCount(harness) >= 2);
+
+      expect(received).toContain("u-2");
+      expect(streamedText(harness)).toContain("ANSWER(m2)");
+    } finally {
+      await harness.close();
+    }
+  });
+});
+
 describe("chat.agent errored turn", () => {
   it(
     "does not duplicate messages buffered after a turn that threw",

From e3e6ad7b20b6f2b86c7f7e521cdded84ae6623fb Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Thu, 27 Aug 2026 10:54:11 +0100
Subject: [PATCH 07/11] fix(chat,sdk): inject only the steering entries this
 drain claimed

`observe` notifies before the router decides where a record goes, so a record
can be seen by an observer and then consumed by a waiting puller rather than
queued. If that happened while `shouldInject` was awaiting, the entry was
still injected even though its `take` failed, and the same message would be
both injected and answered as a turn of its own.

Claiming now happens before the transform, and only claimed entries are
injected. A failed claim means something else is already answering that
message, so dropping it here is what keeps it processed once. The injection
chunk and `onInjected` report the claimed set rather than the offered one, and
an entry with no `seqNum` (the accumulator's own queue) is kept, since there
is no record to claim.

Both mid-turn tests now assert no turn has completed before the second
message is sent. A text delta having arrived does not prove the turn is still
open, so a message landing late would take the ordinary next-turn path and
pass without exercising the mid-turn one.
---
 packages/trigger-sdk/src/v3/ai.ts             | 66 +++++++++++--------
 .../test/pending-message-drain.test.ts        |  8 +++
 2 files changed, 48 insertions(+), 26 deletions(-)

diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts
index 4bf6e8fd9a9..591738b626a 100644
--- a/packages/trigger-sdk/src/v3/ai.ts
+++ b/packages/trigger-sdk/src/v3/ai.ts
@@ -3648,42 +3648,56 @@ async function drainSteeringQueue(
 
   if (!shouldInject) return [];
 
-  // Extract message texts for span attributes
-  const messageTexts = uiMessages.map(
-    (m) =>
-      (m.parts ?? [])
-        .filter((p: any) => p.type === "text")
-        .map((p: any) => p.text)
-        .join("") || ""
-  );
+  const textOfUIMessage = (m: UIMessage) =>
+    (m.parts ?? [])
+      .filter((p: any) => p.type === "text")
+      .map((p: any) => p.text)
+      .join("") || "";
+  // Span attributes describe the offered batch; the chunk and callback below
+  // describe what was actually claimed and injected.
+  const messageTexts = uiMessages.map(textOfUIMessage);
   const _previewText =
     messageTexts.length === 1 ? messageTexts[0]!.slice(0, 80) : `${queue.length} messages`;
 
   return tracer.startActiveSpan(
     "pending message injected",
     async () => {
-      // Transform the batch — default: concatenate all pre-converted model messages
-      const injected = config.prepare
-        ? await config.prepare(batchEvent)
-        : batch.flatMap((e) => e.modelMessages);
-
       /**
-       * Injection is the point of consumption. The records were only observed
+       * Claim before transforming, and inject only what was claimed.
+       *
+       * Injection is the point of consumption: the records were only observed
        * on arrival, so they are still queued on the router and still holding
-       * the resume floor; taking them here is what stops the same message also
-       * being answered as a later turn. A batch that was declined never reaches
+       * the resume floor, and taking them here is what stops the same message
+       * also being answered as a later turn. A declined batch never reaches
        * this line, so its records stay queued and become later turns, which is
        * what "messages queue for the next turn" means.
+       *
+       * A failed claim means something else already consumed that record while
+       * `shouldInject` was awaiting, so it is already being answered as a turn
+       * of its own. Injecting it as well would process the same message twice.
+       * An entry with no `seqNum` did not come from a channel record (the
+       * accumulator's own queue), so there is nothing to claim and it is kept.
        */
       const router = chatInputRouter();
-      for (const entry of batch) {
-        if (entry.seqNum !== undefined) router.take(CHAT_ROUTE_MESSAGES, entry.seqNum);
-        const at = queue.indexOf(entry);
-        if (at !== -1) queue.splice(at, 1);
-      }
+      const claimed = batch.filter((entry) => {
+        const mine = entry.seqNum === undefined || router.take(CHAT_ROUTE_MESSAGES, entry.seqNum);
+        if (mine) {
+          const at = queue.indexOf(entry);
+          if (at !== -1) queue.splice(at, 1);
+        }
+        return mine;
+      });
+
+      if (claimed.length === 0) return [];
+
+      const claimedUIMessages = claimed.map((e) => e.uiMessage);
+      const injected = config.prepare
+        ? await config.prepare({ ...batchEvent, messages: claimedUIMessages })
+        : claimed.flatMap((e) => e.modelMessages);
+
       const injectedIds = locals.get(chatInjectedMessageIdsKey);
       if (injectedIds) {
-        for (const m of uiMessages) injectedIds.add(m.id);
+        for (const m of claimedUIMessages) injectedIds.add(m.id);
       }
 
       // Write injection confirmation chunk to the stream so the frontend
@@ -3697,10 +3711,10 @@ async function drainSteeringQueue(
                 type: PENDING_MESSAGE_INJECTED_TYPE,
                 id: generateMessageId(),
                 data: {
-                  messageIds: uiMessages.map((m) => m.id),
-                  messages: uiMessages.map((m, idx) => ({
+                  messageIds: claimedUIMessages.map((m) => m.id),
+                  messages: claimedUIMessages.map((m) => ({
                     id: m.id,
-                    text: messageTexts[idx] ?? "",
+                    text: textOfUIMessage(m),
                   })),
                 },
               });
@@ -3716,7 +3730,7 @@ async function drainSteeringQueue(
       if (config.onInjected && injected.length > 0) {
         try {
           await config.onInjected({
-            messages: uiMessages,
+            messages: claimedUIMessages,
             injectedModelMessages: injected,
             chatId: ctx?.chatId ?? "",
             turn: ctx?.turn ?? 0,
diff --git a/packages/trigger-sdk/test/pending-message-drain.test.ts b/packages/trigger-sdk/test/pending-message-drain.test.ts
index c15a0903acb..b4f073c544d 100644
--- a/packages/trigger-sdk/test/pending-message-drain.test.ts
+++ b/packages/trigger-sdk/test/pending-message-drain.test.ts
@@ -211,6 +211,10 @@ describe("chat.agent declined steering message", () => {
     try {
       const first = harness.sendMessage(userMessage("m1", "u-1"));
       await waitFor(() => streamedText(harness).includes("ANSWER(m1)"));
+      // A delta having arrived does not prove the turn is still open, and a
+      // message landing after it would take the ordinary next-turn path and
+      // pass this test without exercising the mid-turn one.
+      expect(turnCompleteCount(harness)).toBe(0);
       void harness.sendMessage(userMessage("m2", "u-2"));
       await first;
 
@@ -251,6 +255,10 @@ describe("chat.agent pendingMessages with nothing wired to drain it", () => {
     try {
       const first = harness.sendMessage(userMessage("m1", "u-1"));
       await waitFor(() => streamedText(harness).includes("ANSWER(m1)"));
+      // A delta having arrived does not prove the turn is still open, and a
+      // message landing after it would take the ordinary next-turn path and
+      // pass this test without exercising the mid-turn one.
+      expect(turnCompleteCount(harness)).toBe(0);
       void harness.sendMessage(userMessage("m2", "u-2"));
       await first;
 

From f581a5097347afc3f8e52f375343e1c2e2545ab0 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Thu, 27 Aug 2026 11:02:32 +0100
Subject: [PATCH 08/11] test(core): pin the observe-versus-puller ordering that
 take() guards

An observer is notified before the router decides where a record goes, so a
parked puller can consume a record an observer has just been told about, and a
later take() for it correctly reports false. That ordering is the precondition
for the steering drain injecting an entry it does not own, so it is worth
holding in place rather than leaving it to be rediscovered by review.

Testable here even though the SDK-side drain is not: it needs no model and no
step boundary, only a waiter and an ingest.
---
 .../core/src/v3/sessionStreams/router.test.ts | 29 +++++++++++++++++++
 1 file changed, 29 insertions(+)

diff --git a/packages/core/src/v3/sessionStreams/router.test.ts b/packages/core/src/v3/sessionStreams/router.test.ts
index eef45e913b2..996b9f258a9 100644
--- a/packages/core/src/v3/sessionStreams/router.test.ts
+++ b/packages/core/src/v3/sessionStreams/router.test.ts
@@ -516,3 +516,32 @@ describe("SessionChannelRouter: clearRoute guard", () => {
     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)).toBe(false);
+  });
+});

From 41e132d67038d138980569cf712446ccb1b0d954 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Thu, 27 Aug 2026 11:24:22 +0100
Subject: [PATCH 09/11] fix(chat,core): give the steering claim back when the
 transform throws

Claiming before the transform fixed injecting an entry this drain does not
own, but introduced a worse failure on the way: `prepare` is caller code and
can throw, and by then the records have left the router, so a failing
transform consumed the messages and they were never answered at all. Before
the reorder a throw left them queued.

`take` now returns the record it removed rather than a boolean, `untake` puts
one back in sequence order, and the transform runs inside a try that returns
every claim before rethrowing. `untake` ignores a record already queued, so a
double return cannot duplicate one.

The re-raise is deliberate: the turn should still fail, since the caller's
transform failed. What changes is that the messages survive it and are
answered by a later turn.
---
 .../core/src/v3/sessionStreams/router.test.ts | 38 +++++++++++++--
 packages/core/src/v3/sessionStreams/router.ts | 24 ++++++++--
 packages/trigger-sdk/src/v3/ai.ts             | 46 +++++++++++++++----
 3 files changed, 89 insertions(+), 19 deletions(-)

diff --git a/packages/core/src/v3/sessionStreams/router.test.ts b/packages/core/src/v3/sessionStreams/router.test.ts
index 996b9f258a9..7fd57465cae 100644
--- a/packages/core/src/v3/sessionStreams/router.test.ts
+++ b/packages/core/src/v3/sessionStreams/router.test.ts
@@ -471,7 +471,7 @@ describe("SessionChannelRouter: take", () => {
     r.ingest(rec(0, "message"));
     r.ingest(rec(1, "message"));
 
-    expect(r.take("messages", 0)).toBe(true);
+    expect(r.take("messages", 0)?.seqNum).toBe(0);
     expect(r.pendingCount("messages")).toBe(1);
     expect(r.peek("messages")?.seqNum).toBe(1);
   });
@@ -480,9 +480,9 @@ describe("SessionChannelRouter: take", () => {
     const r = router();
     r.ingest(rec(0, "message"));
 
-    expect(r.take("messages", 0)).toBe(true);
-    expect(r.take("messages", 0)).toBe(false);
-    expect(r.take("messages", 99)).toBe(false);
+    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 () => {
@@ -542,6 +542,34 @@ describe("SessionChannelRouter: observe versus a waiting consumer", () => {
     await pull;
 
     expect(seen).toEqual([0]);
-    expect(r.take("messages", 0)).toBe(false);
+    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);
   });
 });
diff --git a/packages/core/src/v3/sessionStreams/router.ts b/packages/core/src/v3/sessionStreams/router.ts
index fadf144cb93..53669443a20 100644
--- a/packages/core/src/v3/sessionStreams/router.ts
+++ b/packages/core/src/v3/sessionStreams/router.ts
@@ -405,12 +405,28 @@ export class SessionChannelRouter {
    * 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): boolean {
+  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 false;
-    state.queue.splice(index, 1);
-    return true;
+    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. */
diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts
index 591738b626a..c23436bb0b5 100644
--- a/packages/trigger-sdk/src/v3/ai.ts
+++ b/packages/trigger-sdk/src/v3/ai.ts
@@ -40,6 +40,7 @@ import {
   type StreamWriteResult,
   type RouterCheckpoint,
   type SessionRouteTable,
+  type SessionStreamRecord,
 } from "@trigger.dev/core/v3";
 import type {
   FinishReason,
@@ -3679,21 +3680,46 @@ async function drainSteeringQueue(
        * accumulator's own queue), so there is nothing to claim and it is kept.
        */
       const router = chatInputRouter();
-      const claimed = batch.filter((entry) => {
-        const mine = entry.seqNum === undefined || router.take(CHAT_ROUTE_MESSAGES, entry.seqNum);
-        if (mine) {
-          const at = queue.indexOf(entry);
-          if (at !== -1) queue.splice(at, 1);
+      const claimed: SteeringQueueEntry[] = [];
+      const takenRecords: SessionStreamRecord[] = [];
+      for (const entry of batch) {
+        if (entry.seqNum === undefined) {
+          claimed.push(entry);
+          continue;
         }
-        return mine;
-      });
+        const record = router.take(CHAT_ROUTE_MESSAGES, entry.seqNum);
+        if (!record) continue;
+        takenRecords.push(record);
+        claimed.push(entry);
+      }
+      for (const entry of claimed) {
+        const at = queue.indexOf(entry);
+        if (at !== -1) queue.splice(at, 1);
+      }
 
       if (claimed.length === 0) return [];
 
+      /**
+       * Give the claim back if the transform fails. `prepare` is caller code and
+       * can throw; the records have already left the router by this point, so
+       * without returning them a failed transform would consume the messages and
+       * they would never be answered at all.
+       */
+      const releaseClaim = () => {
+        for (const record of takenRecords) router.untake(CHAT_ROUTE_MESSAGES, record);
+        for (const entry of claimed) if (!queue.includes(entry)) queue.push(entry);
+      };
+
       const claimedUIMessages = claimed.map((e) => e.uiMessage);
-      const injected = config.prepare
-        ? await config.prepare({ ...batchEvent, messages: claimedUIMessages })
-        : claimed.flatMap((e) => e.modelMessages);
+      let injected: ModelMessage[];
+      try {
+        injected = config.prepare
+          ? await config.prepare({ ...batchEvent, messages: claimedUIMessages })
+          : claimed.flatMap((e) => e.modelMessages);
+      } catch (err) {
+        releaseClaim();
+        throw err;
+      }
 
       const injectedIds = locals.get(chatInjectedMessageIdsKey);
       if (injectedIds) {

From f6bfe941f55e01197bc970fb56ed21bf3d78d112 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Thu, 27 Aug 2026 11:35:36 +0100
Subject: [PATCH 10/11] test(sdk): cover injection at a prepareStep boundary

Injection had no unit coverage. A prepareStep boundary only exists on a turn
that takes more than one step, and nothing in this package produced one, so
every steering test asserted arrival and none could reach the drain. Both bugs
found in that code during review were found by reading it, not by running it.

`twoStepModel` gives a turn a real boundary: step one calls a tool, the tool
blocks on a gate the test holds, and step two answers. Holding the tool open is
what makes it deterministic, since a message appended while the gate is shut is
queued before `prepareStep` runs with no reliance on stream timing.

Three cases, each checked against the commit that introduced the bug it guards
rather than only observed to pass:

- a mid-turn message is injected and not also answered as its own turn
- a message arriving after the batch was assembled is left for a later turn,
  which fails at 3dd60c20
- a claim is returned when `prepare` throws, which fails at e3e6ad7b

The middle one needed two attempts. Asserting the late message eventually gets
answered passes on the bug, because without the snapshot its model messages
are injected into the first turn, so the text appears either way. A second
turn-complete is the real discriminator.
---
 .../test/steering-injection.test.ts           | 353 ++++++++++++++++++
 1 file changed, 353 insertions(+)
 create mode 100644 packages/trigger-sdk/test/steering-injection.test.ts

diff --git a/packages/trigger-sdk/test/steering-injection.test.ts b/packages/trigger-sdk/test/steering-injection.test.ts
new file mode 100644
index 00000000000..3e0e6336f7f
--- /dev/null
+++ b/packages/trigger-sdk/test/steering-injection.test.ts
@@ -0,0 +1,353 @@
+import { mockChatAgent } from "../src/v3/test/index.js";
+
+import { sessionStreams } from "@trigger.dev/core/v3";
+import type { LanguageModelV3StreamPart } from "@ai-sdk/provider";
+import { simulateReadableStream, stepCountIs, streamText, tool } from "ai";
+import { MockLanguageModelV3 } from "ai/test";
+import { describe, expect, it } from "vitest";
+import { z } from "zod";
+import { chat } from "../src/v3/ai.js";
+
+/**
+ * A `prepareStep` boundary, which is where pending messages are injected, only
+ * exists on a turn that takes more than one step. Nothing else in this package
+ * produced one, so injection had no unit coverage at all: every steering test
+ * asserted arrival and none could reach the drain.
+ *
+ * `twoStepModel` gives a turn a real boundary. Step one calls a tool, the tool
+ * blocks on a gate the test controls, and step two answers. Holding the tool
+ * open is what makes the boundary deterministic: a message appended while the
+ * gate is shut is guaranteed to be queued before `prepareStep` runs, with no
+ * reliance on stream timing.
+ */
+
+const USAGE = {
+  inputTokens: { total: 1, noCache: 1, cacheRead: undefined, cacheWrite: undefined },
+  outputTokens: { total: 1, text: 1, reasoning: undefined },
+};
+
+function userMessage(text: string, id: string) {
+  return { id, role: "user" as const, parts: [{ type: "text" as const, text }] };
+}
+
+function textChunks(text: string): LanguageModelV3StreamPart[] {
+  return [
+    { type: "text-start", id: "t1" },
+    { type: "text-delta", id: "t1", delta: text },
+    { type: "text-end", id: "t1" },
+    { type: "finish", finishReason: { unified: "stop", raw: "stop" }, usage: USAGE },
+  ];
+}
+
+function toolCallChunks(callId: string): LanguageModelV3StreamPart[] {
+  return [
+    { type: "tool-input-start", id: callId, toolName: "gate" },
+    { type: "tool-input-delta", id: callId, delta: JSON.stringify({ q: "x" }) },
+    { type: "tool-input-end", id: callId },
+    { type: "tool-call", toolCallId: callId, toolName: "gate", input: JSON.stringify({ q: "x" }) },
+    { type: "finish", finishReason: { unified: "tool-calls", raw: "tool_calls" }, usage: USAGE },
+  ];
+}
+
+function lastUserText(prompt: { role: string; content: unknown }[]): string {
+  const users = prompt.filter((m) => m.role === "user");
+  const last = users[users.length - 1];
+  return Array.isArray(last?.content)
+    ? (last.content as { type: string; text?: string }[])
+        .filter((p) => p.type === "text")
+        .map((p) => p.text ?? "")
+        .join("")
+    : "";
+}
+
+/** Emits a tool call on each turn's first step, then answers on the second. */
+function twoStepModel() {
+  let step = 0;
+  return new MockLanguageModelV3({
+    doStream: async ({ prompt }) => {
+      const isToolStep = step++ % 2 === 0;
+      return {
+        stream: simulateReadableStream({
+          chunks: isToolStep
+            ? toolCallChunks(`tc-${step}`)
+            : textChunks(`ANSWER(${lastUserText(prompt)})`),
+          initialDelayInMs: 10,
+          chunkDelayInMs: 2,
+        }),
+      };
+    },
+  });
+}
+
+function makeGate() {
+  let open: () => void = () => {};
+  const promise = new Promise<void>((resolve) => {
+    open = resolve;
+  });
+  return { promise, open };
+}
+
+function streamedText(harness: { allChunks: unknown[] }): string {
+  return (harness.allChunks as { type?: string; delta?: string }[])
+    .filter((c) => c.type === "text-delta")
+    .map((c) => c.delta ?? "")
+    .join("");
+}
+
+function injectedChunks(harness: { allRawChunks: unknown[] }) {
+  return (harness.allRawChunks as { type?: string; data?: { messageIds?: string[] } }[]).filter(
+    (c) => c.type === "data-pending-message-injected"
+  );
+}
+
+function injectedIds(harness: { allRawChunks: unknown[] }): string[] {
+  return injectedChunks(harness).flatMap((c) => c.data?.messageIds ?? []);
+}
+
+function turnCompleteCount(harness: { allRawChunks: unknown[] }): number {
+  return (harness.allRawChunks as { type?: string }[]).filter(
+    (c) => c.type === "trigger:turn-complete"
+  ).length;
+}
+
+async function waitFor(check: () => boolean, label = "condition", timeoutMs = 8_000) {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (check()) return;
+    await new Promise((r) => setTimeout(r, 10));
+  }
+  throw new Error(`waitFor timed out: ${label}`);
+}
+
+type SeqReader = { lastSeqNum(sessionId: string, io: "in" | "out"): number | undefined };
+
+/** Appends a message and resolves once the channel has actually taken it. */
+async function sendAndLand(
+  harness: { sendMessage: (m: ReturnType<typeof userMessage>) => Promise<unknown> },
+  chatId: string,
+  text: string,
+  id: string
+) {
+  const seqs = sessionStreams as unknown as SeqReader;
+  const before = seqs.lastSeqNum(chatId, "in") ?? -1;
+  void harness.sendMessage(userMessage(text, id));
+  await waitFor(() => (seqs.lastSeqNum(chatId, "in") ?? -1) > before, `append ${id}`);
+}
+
+describe("chat.agent injection at a prepareStep boundary", () => {
+  it(
+    "injects a message that arrived mid-turn and does not answer it again",
+    { timeout: 30_000 },
+    async () => {
+      const chatId = "inject-basic";
+      const gate = makeGate();
+      let toolEntered = false;
+      const injectedBatches: string[][] = [];
+
+      const gateTool = tool({
+        description: "blocks until the test opens it",
+        inputSchema: z.object({ q: z.string() }),
+        execute: async () => {
+          toolEntered = true;
+          await gate.promise;
+          return "ok";
+        },
+      });
+
+      const agent = chat.agent({
+        id: "steering-injection.basic",
+        pendingMessages: {
+          shouldInject: () => true,
+          onInjected: ({ messages }) => {
+            injectedBatches.push(messages.map((m) => m.id));
+          },
+        },
+        run: async ({ messages, signal }) =>
+          streamText({
+            model: twoStepModel(),
+            messages,
+            abortSignal: signal,
+            // Spread first so the prepareStep it supplies survives and nothing
+            // below is clobbered by it.
+            ...chat.toStreamTextOptions(),
+            tools: { gate: gateTool },
+            stopWhen: stepCountIs(5),
+          }),
+      });
+
+      const harness = mockChatAgent(agent, { chatId });
+      try {
+        const first = harness.sendMessage(userMessage("m1", "u-1"));
+        await waitFor(() => toolEntered, "tool entered");
+
+        await sendAndLand(harness, chatId, "m2", "u-2");
+        gate.open();
+        await first;
+
+        await waitFor(() => injectedBatches.length > 0, "onInjected fired");
+
+        expect(injectedBatches[0]).toEqual(["u-2"]);
+        expect(injectedIds(harness)).toContain("u-2");
+        // Answered inside turn 1, which is what injection means.
+        expect(streamedText(harness)).toContain("ANSWER(m2)");
+
+        // And consumed by it, so it must not also get a turn of its own. A
+        // second turn-complete would mean the record was left on the channel.
+        await new Promise((r) => setTimeout(r, 400));
+        expect(turnCompleteCount(harness)).toBe(1);
+      } finally {
+        gate.open();
+        await harness.close();
+      }
+    }
+  );
+});
+
+describe("chat.agent injection claims only its own batch", () => {
+  /**
+   * `shouldInject` and `prepare` can await, so a record can arrive after the
+   * batch was assembled. The callbacks never saw it and could not have injected
+   * it, so consuming it with the batch would lose a message that should have
+   * become a later turn.
+   */
+  it(
+    "leaves a message that arrived after the batch was assembled",
+    { timeout: 30_000 },
+    async () => {
+      const chatId = "inject-late-arrival";
+      const toolGate = makeGate();
+      const injectGate = makeGate();
+      let toolEntered = false;
+      let injectAsked = false;
+      const injectedBatches: string[][] = [];
+
+      const gateTool = tool({
+        description: "blocks until the test opens it",
+        inputSchema: z.object({ q: z.string() }),
+        execute: async () => {
+          toolEntered = true;
+          await toolGate.promise;
+          return "ok";
+        },
+      });
+
+      const agent = chat.agent({
+        id: "steering-injection.late-arrival",
+        pendingMessages: {
+          shouldInject: async () => {
+            injectAsked = true;
+            await injectGate.promise;
+            return true;
+          },
+          onInjected: ({ messages }) => {
+            injectedBatches.push(messages.map((m) => m.id));
+          },
+        },
+        run: async ({ messages, signal }) =>
+          streamText({
+            model: twoStepModel(),
+            messages,
+            abortSignal: signal,
+            ...chat.toStreamTextOptions(),
+            tools: { gate: gateTool },
+            stopWhen: stepCountIs(5),
+          }),
+      });
+
+      const harness = mockChatAgent(agent, { chatId });
+      try {
+        const first = harness.sendMessage(userMessage("m1", "u-1"));
+        await waitFor(() => toolEntered, "tool entered");
+
+        // m2 is the batch: queued before the boundary, so the callback sees it.
+        await sendAndLand(harness, chatId, "m2", "u-2");
+        toolGate.open();
+        await waitFor(() => injectAsked, "shouldInject called");
+
+        // m3 lands while the callback is parked, so it is not in the batch.
+        await sendAndLand(harness, chatId, "m3", "u-3");
+        injectGate.open();
+        await first;
+
+        await waitFor(() => injectedBatches.length > 0, "onInjected fired");
+        expect(injectedBatches.flat()).toEqual(["u-2"]);
+        expect(injectedIds(harness)).not.toContain("u-3");
+
+        /**
+         * The discriminator. Without the snapshot, m3 is swept into the same
+         * drain: its model messages reach the model, so `ANSWER(m3)` still
+         * appears, but inside turn 1 and without being reported as injected.
+         * Asserting on the text alone therefore passes on the bug. A second
+         * turn-complete is what distinguishes "m3 got its own turn" from "m3
+         * was silently consumed by turn 1".
+         */
+        await waitFor(() => turnCompleteCount(harness) >= 2, "m3 got its own turn");
+        expect(streamedText(harness)).toContain("ANSWER(m3)");
+      } finally {
+        toolGate.open();
+        injectGate.open();
+        await harness.close();
+      }
+    }
+  );
+
+  /**
+   * `prepare` is caller code. Claiming happens before it runs, so a throw would
+   * consume the messages and leave them unanswered unless the claim is returned.
+   */
+  it("gives the claim back when prepare throws", { timeout: 30_000 }, async () => {
+    const chatId = "inject-prepare-throws";
+    const toolGate = makeGate();
+    let toolEntered = false;
+    let prepareCalls = 0;
+
+    const gateTool = tool({
+      description: "blocks until the test opens it",
+      inputSchema: z.object({ q: z.string() }),
+      execute: async () => {
+        toolEntered = true;
+        await toolGate.promise;
+        return "ok";
+      },
+    });
+
+    const agent = chat.agent({
+      id: "steering-injection.prepare-throws",
+      pendingMessages: {
+        shouldInject: () => true,
+        prepare: () => {
+          prepareCalls++;
+          throw new Error("synthetic prepare failure");
+        },
+      },
+      run: async ({ messages, signal }) =>
+        streamText({
+          model: twoStepModel(),
+          messages,
+          abortSignal: signal,
+          ...chat.toStreamTextOptions(),
+          tools: { gate: gateTool },
+          stopWhen: stepCountIs(5),
+        }),
+    });
+
+    const harness = mockChatAgent(agent, { chatId });
+    try {
+      const first = harness.sendMessage(userMessage("m1", "u-1")).catch(() => undefined);
+      await waitFor(() => toolEntered, "tool entered");
+
+      await sendAndLand(harness, chatId, "m2", "u-2");
+      toolGate.open();
+      await first;
+
+      await waitFor(() => prepareCalls > 0, "prepare called");
+      // Nothing was injected, and the message must not have been eaten by the
+      // failed transform: it is still owed and gets answered by a later turn.
+      expect(injectedIds(harness)).not.toContain("u-2");
+      await waitFor(() => streamedText(harness).includes("ANSWER(m2)"), "m2 answered later");
+    } finally {
+      toolGate.open();
+      await harness.close();
+    }
+  });
+});

From b0ecb9114cf0c6a2dfbcc67fdf2759caf6a1d0fe Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Thu, 27 Aug 2026 11:37:27 +0100
Subject: [PATCH 11/11] test(sdk): repro the managed path dropping an injected
 message from history

The deployed QA lane finds the two surfaces disagree: a `chat.createSession()`
recap in the same run recalls a mid-turn steer, while the managed `chat.agent`
loop denies it. This pins the managed half at unit level, which the tracked
gap has never had.

Turn 2's prompt comes back as the original message and the following one, with
the injected one absent, so the message reaches the model inside turn 1 and
then leaves no trace in history. Verified as a real assertion failure rather
than trusting `it.fails`, which would also pass on a timeout.

Not fixed here, and not caused by this branch: nothing in it touches the
accumulator. Recorded so the day the managed path starts carrying it is
noticed, and so the difference between the surfaces has a repro that needs no
deployed environment.
---
 .../test/steering-injection.test.ts           | 98 +++++++++++++++++++
 1 file changed, 98 insertions(+)

diff --git a/packages/trigger-sdk/test/steering-injection.test.ts b/packages/trigger-sdk/test/steering-injection.test.ts
index 3e0e6336f7f..12f3dfa0c87 100644
--- a/packages/trigger-sdk/test/steering-injection.test.ts
+++ b/packages/trigger-sdk/test/steering-injection.test.ts
@@ -351,3 +351,101 @@ describe("chat.agent injection claims only its own batch", () => {
     }
   });
 });
+
+/**
+ * Whether an injected message survives into the next turn's model context.
+ *
+ * Recorded here because the deployed QA lane finds the two surfaces disagree:
+ * a `chat.createSession()` recap in the same run recalls a mid-turn steer,
+ * while the managed `chat.agent` loop denies it. That difference is
+ * pre-existing and is the surface-specific half of the accumulator gap.
+ *
+ * `it.fails` because the managed path does not carry it: turn 2's prompt comes
+ * back as the original and the following message only, with the injected one
+ * absent. Held here so the day that changes is noticed, and so the gap has a
+ * repro that does not need a deployed environment.
+ */
+describe("chat.agent injected message in the next turn's context", () => {
+  it.fails(
+    "carries an injected message into the following turn's prompt",
+    { timeout: 30_000 },
+    async () => {
+      const chatId = "inject-next-turn";
+      const toolGate = makeGate();
+      let toolEntered = false;
+      const prompts: string[][] = [];
+
+      const gateTool = tool({
+        description: "blocks until the test opens it",
+        inputSchema: z.object({ q: z.string() }),
+        execute: async () => {
+          toolEntered = true;
+          await toolGate.promise;
+          return "ok";
+        },
+      });
+
+      let step = 0;
+      const recordingModel = new MockLanguageModelV3({
+        doStream: async ({ prompt }) => {
+          prompts.push(
+            prompt
+              .filter((m) => m.role === "user")
+              .flatMap((m) =>
+                Array.isArray(m.content)
+                  ? (m.content as { type: string; text?: string }[])
+                      .filter((c) => c.type === "text")
+                      .map((c) => c.text ?? "")
+                  : []
+              )
+          );
+          const isToolStep = step++ % 2 === 0;
+          return {
+            stream: simulateReadableStream({
+              chunks: isToolStep ? toolCallChunks(`tc-${step}`) : textChunks("done"),
+              initialDelayInMs: 10,
+              chunkDelayInMs: 2,
+            }),
+          };
+        },
+      });
+
+      const agent = chat.agent({
+        id: "steering-injection.next-turn",
+        pendingMessages: { shouldInject: () => true },
+        run: async ({ messages, signal }) =>
+          streamText({
+            model: recordingModel,
+            messages,
+            abortSignal: signal,
+            ...chat.toStreamTextOptions(),
+            tools: { gate: gateTool },
+            stopWhen: stepCountIs(5),
+          }),
+      });
+
+      const harness = mockChatAgent(agent, { chatId });
+      try {
+        const first = harness.sendMessage(userMessage("m1", "u-1"));
+        await waitFor(() => toolEntered, "tool entered");
+        await sendAndLand(harness, chatId, "steer-me", "u-2");
+        toolGate.open();
+        await first;
+
+        await waitFor(() => turnCompleteCount(harness) >= 1, "turn 1 complete");
+        const promptsAfterTurn1 = prompts.length;
+
+        // A fresh turn. Its prompt is built from accumulated history, so it
+        // should still contain the message that was injected into turn 1.
+        await harness.sendMessage(userMessage("m3", "u-3"));
+        await waitFor(() => prompts.length > promptsAfterTurn1, "turn 2 prompt built");
+
+        const turn2Prompt = prompts[promptsAfterTurn1]!;
+        expect(turn2Prompt).toContain("steer-me");
+      } finally {
+        toolGate.open();
+        await harness.close();
+      }
+    }
+  );
+});