diff --git a/.changeset/calm-chat-reconnects.md b/.changeset/calm-chat-reconnects.md new file mode 100644 index 00000000000..a1af4aa43d5 --- /dev/null +++ b/.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. diff --git a/packages/trigger-sdk/src/v3/chat.test.ts b/packages/trigger-sdk/src/v3/chat.test.ts index 452c8090b70..e3fb3ad4b9a 100644 --- a/packages/trigger-sdk/src/v3/chat.test.ts +++ b/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); }); diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts index a7c7125575e..798c4b6e034 100644 --- a/packages/trigger-sdk/src/v3/chat.ts +++ b/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 { 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 { this.activeStreams.delete(chatId); } + state.activeInputSeq = inSeq; state.isStreaming = true; this.notifySessionChange(chatId, state); @@ -1177,13 +1183,14 @@ export class TriggerChatTransport implements ChatTransport { 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 { // 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 { 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 { 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 { }) 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 { 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 { 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); diff --git a/packages/trigger-sdk/test/chat-turn-correlation.test.ts b/packages/trigger-sdk/test/chat-turn-correlation.test.ts index 7295ecba797..f81ca755a27 100644 --- a/packages/trigger-sdk/test/chat-turn-correlation.test.ts +++ b/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 = { + "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 { } 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(); + }); });