fix(sdk): detach the turn message handler when a turn throws

The per-turn handler leaked past turns that threw outside the streaming
section (e.g. an onTurnStart hook). With the buffer now loop-level, a
leaked handler pushed alongside the next turn's handler, duplicating
every mid-stream message; pre-existing behavior lost them instead. The
subscription handle is hoisted so the turn's catch/finally always
detaches it, and createSession defensively detaches its prior turn's
handler when user code exits a turn without complete()/done().

Author: ericallam

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

@@ -6386,6 +6386,9 @@ function chatAgent<
         }
 
         for (let turn = 0; turn < maxTurns; turn++) {
+          // Declared here so the finally can detach it — a handler leaked past
+          // its turn duplicates every mid-stream message into the shared buffer.
+          let turnMsgSub: { off: () => void } | undefined;
           try {
             // Extract turn-level context before entering the span. Slim
             // wire: at most one delta message per record. `headStartMessages`
@@ -6539,6 +6542,7 @@ function chatAgent<
                     msg as ChatTaskWirePayload<TUIMessage, inferSchemaIn<TClientDataSchema>>
                   );
                 });
+                turnMsgSub = msgSub;
 
                 // Track new messages for this turn (user input + assistant response).
                 const turnNewModelMessages: ModelMessage[] = [];
@@ -7851,6 +7855,9 @@ function chatAgent<
             // Turn error handler: write an error chunk + turn-complete to the stream
             // so the client sees the error, then wait for the next message instead
             // of killing the entire run. This keeps the conversation alive.
+            // Detach the turn's message handler first — left attached it would
+            // eat the very message the wait below is waiting for.
+            turnMsgSub?.off();
             if (
               turnError instanceof Error &&
               turnError.name === "AbortError" &&
@@ -8014,6 +8021,8 @@ function chatAgent<
               inferSchemaIn<TClientDataSchema>
             >;
             // Continue to next iteration of the for loop
+          } finally {
+            turnMsgSub?.off();
           }
         }
       } finally {
@@ -9323,9 +9332,14 @@ function createChatSession(
       // for the same reason as the agent loop's `pendingWireMessages`:
       // consumed records never replay, so a turn-local buffer loses them.
       const sessionPendingWire: ChatTaskWirePayload[] = [];
+      // The current turn's message subscription — detached defensively at the
+      // top of next() in case user code threw without complete()/done().
+      let activeMsgSub: { off: () => void } | undefined;
 
       return {
         async next(): Promise<IteratorResult<ChatTurn>> {
+          activeMsgSub?.off();
+          activeMsgSub = undefined;
           if (!booted) {
             booted = true;
             await seedSessionInResumeCursorForCustomLoop(currentPayload);
@@ -9476,6 +9490,7 @@ function createChatSession(
 
             sessionPendingWire.push(msg);
           });
+          activeMsgSub = sessionMsgSub;
 
           // Accumulate messages. Slim wire: pass the single delta message as
           // a 0-or-1-length array. The accumulator's behavior is unchanged —

packages/trigger-sdk/test/pending-message-drain.test.ts

@@ -118,6 +118,49 @@ describe("chat.agent pending wire buffer", () => {
   });
 });
 
+describe("chat.agent errored turn", () => {
+  it(
+    "does not duplicate messages buffered after a turn that threw",
+    { timeout: 20000 },
+    async () => {
+      // Throw from a pre-stream hook: throws inside the streaming section are
+      // already covered by its finally, but a hook throw used to leak the
+      // turn's message handler into the loop-level buffer.
+      let turnStarts = 0;
+      const agent = chat.agent({
+        id: "pending-drain.errored-turn",
+        onTurnStart: async () => {
+          turnStarts++;
+          if (turnStarts === 1) {
+            throw new Error("synthetic turn failure");
+          }
+        },
+        run: async ({ messages, signal }) => {
+          return streamText({ model: echoModel(), messages, abortSignal: signal });
+        },
+      });
+
+      const harness = mockChatAgent(agent, { chatId: "pending-drain-4" });
+      try {
+        // Turn 1 throws — pre-fix its message handler leaked past the turn.
+        await harness.sendMessage(userMessage("boom", "u-1"));
+        const second = harness.sendMessage(userMessage("m2", "u-2"));
+        // m3 lands mid-turn; a leaked handler would push it twice.
+        await waitFor(() => streamedText(harness).includes("ANSWER(m2)"));
+        void harness.sendMessage(userMessage("m3", "u-3"));
+        await second;
+
+        await waitFor(() => streamedText(harness).includes("ANSWER(m3)"));
+        await new Promise((r) => setTimeout(r, 500));
+        const text = streamedText(harness);
+        expect(text.match(/ANSWER\(m3\)/g)).toHaveLength(1);
+      } finally {
+        await harness.close();
+      }
+    }
+  );
+});
+
 describe("chat.createSession pending wire buffer", () => {
   it("dispatches messages buffered during a turn as subsequent turns", async () => {
     const agent = chat.customAgent({
@@ -158,46 +201,50 @@ describe("chat.createSession pending wire buffer", () => {
 });
 
 describe("chat.createSession stop + immediate send", () => {
-  it("dispatches a message that arrives right after a stopped turn", { timeout: 20000 }, async () => {
-    const agent = chat.customAgent({
-      id: "pending-drain.session-stop",
-      run: async (payload) => {
-        const session = chat.createSession(payload, {
-          signal: new AbortController().signal,
-          idleTimeoutInSeconds: 2,
-          // Steering config active — the failure mode routed post-stream
-          // arrivals into the dead steering queue instead of the next turn.
-          pendingMessages: {},
-        });
-        for await (const turn of session) {
-          const result = streamText({
-            model: echoModel(),
-            messages: turn.messages,
-            abortSignal: turn.signal,
+  it(
+    "dispatches a message that arrives right after a stopped turn",
+    { timeout: 20000 },
+    async () => {
+      const agent = chat.customAgent({
+        id: "pending-drain.session-stop",
+        run: async (payload) => {
+          const session = chat.createSession(payload, {
+            signal: new AbortController().signal,
+            idleTimeoutInSeconds: 2,
+            // Steering config active — the failure mode routed post-stream
+            // arrivals into the dead steering queue instead of the next turn.
+            pendingMessages: {},
           });
-          await turn.complete(result);
-        }
-      },
-    });
+          for await (const turn of session) {
+            const result = streamText({
+              model: echoModel(),
+              messages: turn.messages,
+              abortSignal: turn.signal,
+            });
+            await turn.complete(result);
+          }
+        },
+      });
 
-    const harness = mockChatAgent(agent, { chatId: "pending-drain-3" });
-    try {
-      const first = harness.sendMessage(userMessage("write a long essay", "u-1"));
-      await waitFor(() => streamedText(harness).length > 0);
-      await harness.sendStop();
-      // Land the next message inside the stopped turn's post-stream window
-      // (the ~2s totalUsage race), after the abort has settled — previously
-      // the still-attached handler steering-routed it into the dead queue.
-      await new Promise((r) => setTimeout(r, 150));
-      void harness.sendMessage(userMessage("m2", "u-2"));
-      await first;
+      const harness = mockChatAgent(agent, { chatId: "pending-drain-3" });
+      try {
+        const first = harness.sendMessage(userMessage("write a long essay", "u-1"));
+        await waitFor(() => streamedText(harness).length > 0);
+        await harness.sendStop();
+        // Land the next message inside the stopped turn's post-stream window
+        // (the ~2s totalUsage race), after the abort has settled — previously
+        // the still-attached handler steering-routed it into the dead queue.
+        await new Promise((r) => setTimeout(r, 150));
+        void harness.sendMessage(userMessage("m2", "u-2"));
+        await first;
 
-      await waitFor(() => turnCompleteCount(harness) >= 2);
-      await waitFor(() => streamedText(harness).includes("ANSWER(m2)"));
-    } finally {
-      await harness.close();
+        await waitFor(() => turnCompleteCount(harness) >= 2);
+        await waitFor(() => streamedText(harness).includes("ANSWER(m2)"));
+      } finally {
+        await harness.close();
+      }
     }
-  });
+  );
 });
 
 describe("session.in.wait() consume cursor", () => {