fix(chat): persist active input sequence across reloads

Author: gtremper

.changeset/calm-chat-reconnects.md

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

packages/trigger-sdk/src/v3/chat.test.ts

@@ -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);
     });

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

@@ -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,
         });
       }
@@ -870,6 +875,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       this.activeStreams.delete(chatId);
     }
 
+    state.activeInputSeq = inSeq;
     state.isStreaming = true;
     this.notifySessionChange(chatId, state);
 
@@ -1177,13 +1183,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,
     });
   };
 
@@ -1283,6 +1290,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);
 
@@ -1307,6 +1315,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)!));
@@ -1441,6 +1450,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
   private toPersisted = (state: ChatSessionState): ChatSessionPersistedState => ({
     publicAccessToken: state.publicAccessToken,
     lastEventId: state.lastEventId,
+    activeInputSeq: state.activeInputSeq,
     isStreaming: state.isStreaming,
   });
 
@@ -1750,6 +1760,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, {
@@ -1983,10 +1994,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;
                 }
               }
@@ -2004,6 +2015,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);

packages/trigger-sdk/test/chat-turn-correlation.test.ts

@@ -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();
+  });
 });