fix(chat): ignore stale turn completions after reconnect (#4643)

## Summary Reloading a browser chat mid-turn can replay a completion event for an older input and close the active turn too early. This persists the last browser-owned input sequence and reuses it on reconnect, so older completion events are ignored. The sequence is cleared after the matching boundary, and reconnect avoids the settled-peek shortcut while that sequence is active. The persisted field is optional, so sessions without it keep their existing behavior. ## Testing - `pnpm --dir packages/trigger-sdk run test ./src/v3/chat.test.ts ./test/chat-turn-correlation.test.ts --run` — 67 passed - `pnpm --dir packages/trigger-sdk run test --run` — 32 files, 379 tests passed - `pnpm run build --filter @trigger.dev/sdk` - `pnpm run format` - `pnpm run lint` ## Changelog Browser chats now keep the active turn open across page reloads when older completion records are replayed. ## ✅ Checklist - [x] I have followed every step in the [contributing guide](https://github.com/triggerdotdev/trigger.dev/blob/main/CONTRIBUTING.md) - [x] The PR title follows the convention. - [x] I ran and tested the code works 💯 Co-authored-by: Matt Aitken <matt@mattaitken.com>
2026-08-27 02:07:32 -07:00
parent c115f440bc
commit 1065251ca7
4 changed files with 196 additions and 15 deletions
@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Browser chats now keep the active turn open across page reloads when older completion records are replayed.
@@ -228,6 +228,7 @@ describe("TriggerChatTransport", () => {
          "chat-1": {
            publicAccessToken: "hydrated-pat",
            lastEventId: "42",
+            activeInputSeq: 41,
            isStreaming: false,
          },
        },
@@ -237,6 +238,7 @@ describe("TriggerChatTransport", () => {
      expect(session).toEqual({
        publicAccessToken: "hydrated-pat",
        lastEventId: "42",
+        activeInputSeq: 41,
        isStreaming: false,
      });
    });
@@ -262,15 +264,21 @@ describe("TriggerChatTransport", () => {
      transport.setSession("chat-x", {
        publicAccessToken: "tok",
        lastEventId: "10",
+        activeInputSeq: 9,
      });

      expect(transport.getSession("chat-x")).toMatchObject({
        publicAccessToken: "tok",
        lastEventId: "10",
+        activeInputSeq: 9,
      });
      expect(onSessionChange).toHaveBeenCalledWith(
        "chat-x",
-        expect.objectContaining({ publicAccessToken: "tok", lastEventId: "10" })
+        expect.objectContaining({
+          publicAccessToken: "tok",
+          lastEventId: "10",
+          activeInputSeq: 9,
+        })
      );
    });

@@ -977,7 +985,9 @@ describe("TriggerChatTransport", () => {
    it("marks the session streaming and notifies before subscribing", async () => {
      global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
        const urlStr = typeof url === "string" ? url : url.toString();
-        if (isSessionStreamAppendUrl(urlStr)) return defaultAppendResponse();
+        if (isSessionStreamAppendUrl(urlStr)) {
+          return new Response(JSON.stringify({ ok: true, seq: 7 }), { status: 200 });
+        }
        if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse();
        throw new Error(`Unexpected URL: ${urlStr}`);
      });
@@ -994,7 +1004,9 @@ describe("TriggerChatTransport", () => {
      // isStreaming:true must be observed during the action — otherwise a reload
      // mid-action sees a persisted isStreaming:false and never resumes.
      expect(
-        onSessionChange.mock.calls.some(([, session]) => session && session.isStreaming === true)
+        onSessionChange.mock.calls.some(
+          ([, session]) => session && session.isStreaming === true && session.activeInputSeq === 7
+        )
      ).toBe(true);
      await drainChunks(stream);
    });
@@ -422,11 +422,13 @@ export type StartSessionResult = {
 * Public surface of {@link TriggerChatTransport}'s session state. Everything
 * the customer should persist for resumption across page reloads. The
 * transport addresses by `chatId` everywhere, so this is light: just a PAT,
- * the last SSE event id, and a couple of UX-state flags.
+ * resume cursors, and a couple of UX-state flags.
 */
 export type ChatSessionPersistedState = {
  publicAccessToken: string;
  lastEventId?: string;
+  /** The `.in` append sequence of the last send this client owned; reused as `sinceInSeq` on reconnect. */
+  activeInputSeq?: number;
  isStreaming?: boolean;
 };

@@ -631,6 +633,8 @@ type ChatSessionState = {
  publicAccessToken: string;
  /** Last SSE event ID — used to resume the stream without replaying old events. */
  lastEventId?: string;
+  /** `.in` append sequence used to filter stale turn boundaries after reconnecting. */
+  activeInputSeq?: number;
  /** Set when the stream was aborted mid-turn (stop). On reconnect, skip chunks until trigger:turn-complete. */
  skipToTurnComplete?: boolean;
  /** Whether the agent is currently streaming a response. Set on first chunk, cleared on turn-complete. */
@@ -718,6 +722,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
        this.sessions.set(chatId, {
          publicAccessToken: session.publicAccessToken,
          lastEventId: session.lastEventId,
+          activeInputSeq: session.activeInputSeq,
          isStreaming: session.isStreaming,
        });
      }
@@ -874,6 +879,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
    // turn would be skipped record by record.
    state.skipToTurnComplete = false;

+    state.activeInputSeq = inSeq;
    state.isStreaming = true;
    this.notifySessionChange(chatId, state);

@@ -1181,13 +1187,14 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
    return this.subscribeToSessionStream(state, abortSignal, options.chatId, {
      resumed: true,
      sendStopOnAbort: options.stopOnAbort ?? false,
+      sinceInSeq: state.activeInputSeq,
      // Reconnect-on-reload opts into the server's settled-peek shortcut
-      // so the SSE doesn't hang for 60s when no turn is in flight. Active
-      // send-a-message paths must keep wait=60 to avoid racing the
-      // freshly-triggered turn's first chunk. Watch mode must NOT peek: a
-      // settled peek between turns sets sessionSettled and closes the
+      // so the SSE doesn't hang for 60s when no turn is in flight. A known
+      // active input must not peek because the previous turn's completion
+      // can remain at the tail until the current turn writes its first chunk.
+      // Watch mode must NOT peek: a settled peek between turns closes the
      // standing subscription, so the viewer never sees the next turn.
-      peekSettled: !this.watchMode,
+      peekSettled: !this.watchMode && state.activeInputSeq === undefined,
    });
  };

@@ -1291,6 +1298,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {

    // Mark streaming + persist so a reload mid-action resumes (reconnectToStream
    // no-ops when the persisted session says isStreaming: false).
+    state.activeInputSeq = inSeq;
    state.isStreaming = true;
    this.notifySessionChange(chatId, state);

@@ -1315,6 +1323,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
    this.sessions.set(chatId, {
      publicAccessToken: session.publicAccessToken,
      lastEventId: session.lastEventId,
+      activeInputSeq: session.activeInputSeq,
      isStreaming: session.isStreaming,
    });
    this.notifySessionChange(chatId, this.toPersisted(this.sessions.get(chatId)!));
@@ -1449,6 +1458,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
  private toPersisted = (state: ChatSessionState): ChatSessionPersistedState => ({
    publicAccessToken: state.publicAccessToken,
    lastEventId: state.lastEventId,
+    activeInputSeq: state.activeInputSeq,
    isStreaming: state.isStreaming,
  });

@@ -1758,6 +1768,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
            }) as typeof fetch)
          : undefined;
        let sawFirstChunk = false;
+        let sinceInSeq = options?.sinceInSeq;

        const connectSseOnce = async (token: string) => {
          const subscription = new SSEStreamSubscription(streamUrl, {
@@ -1991,10 +2002,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
            if (controlValue === TRIGGER_CONTROL_SUBTYPE.TURN_COMPLETE) {
              // Skip a turn-complete from an earlier turn (committed `.in` cursor
              // below this send's seq), e.g. an undo action that raced this send.
-              if (options?.sinceInSeq !== undefined) {
+              if (sinceInSeq !== undefined) {
                const cursorRaw = headerValue(value.headers, SESSION_IN_EVENT_ID_HEADER);
                const cursor = cursorRaw !== undefined ? Number.parseInt(cursorRaw, 10) : NaN;
-                if (!Number.isNaN(cursor) && cursor < options.sinceInSeq) {
+                if (!Number.isNaN(cursor) && cursor < sinceInSeq) {
                  continue;
                }
              }
@@ -2012,6 +2023,8 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
                sessionInEventId: headerValue(value.headers, SESSION_IN_EVENT_ID_HEADER),
                ...this.turnAttribution(chatId),
              });
+              state.activeInputSeq = undefined;
+              sinceInSeq = undefined;
              state.isStreaming = false;
              this.notifySessionChange(chatId, state);
              this.coordinator?.release(chatId);
@@ -1,4 +1,4 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import type { UIMessage } from "ai";
 import { TriggerChatTransport, type TriggerChatTransportOptions } from "../src/v3/chat.js";

@@ -17,13 +17,18 @@ type BatchRecord = {
  headers?: Array<[string, string]>;
 };

-function batchResponse(records: BatchRecord[]): Response {
+function batchResponse(records: BatchRecord[], settled = false): Response {
  const frames = records
    .map((r) => `event: batch\ndata: ${JSON.stringify({ records: [r] })}\n\n`)
    .join("");
+  const headers: Record<string, string> = {
+    "Content-Type": "text/event-stream",
+    "X-Stream-Version": "v2",
+  };
+  if (settled) headers["X-Session-Settled"] = "true";
  return new Response(frames, {
    status: 200,
-    headers: { "Content-Type": "text/event-stream", "X-Stream-Version": "v2" },
+    headers,
  });
 }

@@ -42,7 +47,10 @@ function turnComplete(seqNum: number, inCursor: number): BatchRecord {

 function textDelta(seqNum: number, text: string): BatchRecord {
  return {
-    body: JSON.stringify({ data: { type: "text-delta", id: "t1", delta: text }, id: "m1" }),
+    body: JSON.stringify({
+      data: { type: "text-delta", id: "t1", delta: text },
+      id: `m${seqNum}`,
+    }),
    seq_num: seqNum,
    timestamp: seqNum,
    headers: [],
@@ -88,6 +96,36 @@ async function submit(transport: TriggerChatTransport): Promise<string[]> {
 }

 describe("transport turn correlation", () => {
+  it("persists the owned send's input sequence before subscribing", async () => {
+    const onSessionChange = vi.fn();
+    const transport = new TriggerChatTransport({
+      task: "test-task",
+      accessToken: async () => "tok_test",
+      sessions: { c1: { publicAccessToken: "tok_test", isStreaming: false } },
+      onSessionChange,
+      fetch: async (_url, _init, ctx) =>
+        ctx.endpoint === "in" ? inResponse(5) : batchResponse([turnComplete(10, 5)]),
+    });
+
+    const stream = await transport.sendMessages({
+      trigger: "submit-message",
+      chatId: "c1",
+      messageId: undefined,
+      messages: [user("hi", "u-1")],
+      abortSignal: undefined,
+    });
+
+    expect(onSessionChange).toHaveBeenCalledWith("c1", {
+      publicAccessToken: "tok_test",
+      lastEventId: undefined,
+      activeInputSeq: 5,
+      isStreaming: true,
+    });
+    expect(transport.getSession("c1")?.activeInputSeq).toBe(5);
+    await readDeltas(stream);
+    expect(transport.getSession("c1")?.activeInputSeq).toBeUndefined();
+  });
+
  it("skips an earlier turn's turn-complete and closes on its own", async () => {
    // Append seq 5; the undo turn's complete (cursor 4) must be skipped.
    const out = batchResponse([turnComplete(10, 4), textDelta(11, "56"), turnComplete(12, 5)]);
@@ -107,4 +145,117 @@ describe("transport turn correlation", () => {
    const deltas = await submit(makeTransport(out, undefined));
    expect(deltas).toEqual([]);
  });
+
+  it("reuses a hydrated input sequence to skip stale turn-completes after reconnecting", async () => {
+    const transport = new TriggerChatTransport({
+      task: "test-task",
+      accessToken: async () => "tok_test",
+      sessions: {
+        c1: { publicAccessToken: "tok_test", isStreaming: true, activeInputSeq: 5 },
+      },
+      fetch: async () =>
+        batchResponse([turnComplete(10, 4), textDelta(11, "current"), turnComplete(12, 5)]),
+    });
+
+    const stream = await transport.reconnectToStream({ chatId: "c1" });
+
+    expect(stream).not.toBeNull();
+    await expect(readDeltas(stream!)).resolves.toEqual(["current"]);
+    expect(transport.getSession("c1")?.isStreaming).toBe(false);
+    expect(transport.getSession("c1")?.activeInputSeq).toBeUndefined();
+  });
+
+  it("does not request a settled peek while reconnecting a known active input", async () => {
+    vi.useFakeTimers();
+    try {
+      const subscribeHeaders: Headers[] = [];
+      const transport = new TriggerChatTransport({
+        task: "test-task",
+        accessToken: async () => "tok_test",
+        sessions: {
+          c1: { publicAccessToken: "tok_test", isStreaming: true, activeInputSeq: 5 },
+        },
+        fetch: async (_url, init) => {
+          const headers = new Headers(init?.headers);
+          subscribeHeaders.push(headers);
+
+          if (subscribeHeaders.length === 1) {
+            // Match the server shortcut: a peek sees the previous turn's
+            // boundary at the tail and marks this otherwise-normal EOF settled.
+            return batchResponse([turnComplete(10, 4)], headers.has("X-Peek-Settled"));
+          }
+
+          return batchResponse([textDelta(11, "current"), turnComplete(12, 5)]);
+        },
+      });
+
+      const stream = await transport.reconnectToStream({ chatId: "c1" });
+
+      expect(stream).not.toBeNull();
+      const deltas = readDeltas(stream!);
+      await vi.advanceTimersByTimeAsync(1_000);
+      await expect(deltas).resolves.toEqual(["current"]);
+      expect(subscribeHeaders).toHaveLength(2);
+      expect(subscribeHeaders[0]?.get("X-Peek-Settled")).toBeNull();
+      expect(transport.getSession("c1")?.isStreaming).toBe(false);
+      expect(transport.getSession("c1")?.activeInputSeq).toBeUndefined();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+  it.each([5, 6])(
+    "accepts a reconnected turn-complete at or after the active input sequence (%i)",
+    async (inCursor) => {
+      const transport = new TriggerChatTransport({
+        task: "test-task",
+        accessToken: async () => "tok_test",
+        sessions: {
+          c1: { publicAccessToken: "tok_test", isStreaming: true, activeInputSeq: 5 },
+        },
+        fetch: async () => batchResponse([turnComplete(10, inCursor), textDelta(11, "late")]),
+      });
+
+      const stream = await transport.reconnectToStream({ chatId: "c1" });
+
+      expect(stream).not.toBeNull();
+      await expect(readDeltas(stream!)).resolves.toEqual([]);
+      expect(transport.getSession("c1")?.isStreaming).toBe(false);
+      expect(transport.getSession("c1")?.activeInputSeq).toBeUndefined();
+    }
+  );
+
+  it("uses the input sequence for one accepted watch turn only", async () => {
+    let outCalls = 0;
+    const turnCompleted: number[] = [];
+    const transport = new TriggerChatTransport({
+      task: "test-task",
+      accessToken: async () => "tok_test",
+      watch: true,
+      sessions: {
+        c1: { publicAccessToken: "tok_test", isStreaming: true, activeInputSeq: 5 },
+      },
+      onEvent: (event) => {
+        if (event.type === "turn-completed") turnCompleted.push(Number(event.sessionInEventId));
+      },
+      fetch: async () => {
+        outCalls++;
+        return outCalls === 1
+          ? batchResponse([
+              turnComplete(10, 4),
+              textDelta(11, "first"),
+              turnComplete(12, 5),
+              textDelta(13, "second"),
+              turnComplete(14, 4),
+            ])
+          : batchResponse([], true);
+      },
+    });
+
+    const stream = await transport.reconnectToStream({ chatId: "c1" });
+
+    expect(stream).not.toBeNull();
+    await expect(readDeltas(stream!)).resolves.toEqual(["first", "second"]);
+    expect(turnCompleted).toEqual([5, 4]);
+    expect(transport.getSession("c1")?.activeInputSeq).toBeUndefined();
+  });
 });