diff --git a/.changeset/chat-stop-successor-boundary.md b/.changeset/chat-stop-successor-boundary.md new file mode 100644 index 00000000000..47fbc8e1e4f --- /dev/null +++ b/.changeset/chat-stop-successor-boundary.md @@ -0,0 +1,6 @@ +--- +"@trigger.dev/sdk": patch +--- + +Keep new chat responses intact after Stop, including slow Stop acknowledgments and page reloads. +Sequence-free replies after Stop require a transcript reload before further messages. diff --git a/packages/trigger-sdk/src/v3/chat-stop.test.ts b/packages/trigger-sdk/src/v3/chat-stop.test.ts new file mode 100644 index 00000000000..861de1f09f5 --- /dev/null +++ b/packages/trigger-sdk/src/v3/chat-stop.test.ts @@ -0,0 +1,402 @@ +import { createServer, type Server, type ServerResponse } from "node:http"; +import { readUIMessageStream, type UIMessageChunk } from "ai"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + TriggerChatTransport, + type ChatSessionPersistedState, + type TriggerChatTransportOptions, +} from "./chat.js"; + +type OutputRecord = { + seq_num: number; + timestamp: number; + body: string; + headers: string[][]; +}; + +function chunk(seq: number, data: UIMessageChunk): OutputRecord { + return { + seq_num: seq, + timestamp: seq, + body: JSON.stringify({ id: `part-${seq}`, data }), + headers: [], + }; +} + +function complete(seq: number, input: number): OutputRecord { + return { + seq_num: seq, + timestamp: seq, + body: "", + headers: [ + ["trigger-control", "turn-complete"], + ["session-in-event-id", String(input)], + ], + }; +} + +function reply(start: number): OutputRecord[] { + return [ + chunk(start, { type: "start", messageId: "new" }), + chunk(start + 1, { type: "text-start", id: "text" }), + chunk(start + 2, { type: "text-delta", id: "text", delta: "New response" }), + chunk(start + 3, { type: "text-end", id: "text" }), + chunk(start + 4, { type: "finish" }), + ]; +} + +async function readText(stream: ReadableStream): Promise { + let text = ""; + for await (const message of readUIMessageStream({ stream, terminateOnError: true })) { + text = message.parts + .filter((part) => part.type === "text") + .map((part) => part.text) + .join(""); + } + return text; +} + +describe("Stop with a successor response", () => { + let server: Server; + let baseURL: string; + let transport: TriggerChatTransport; + let outputs: ServerResponse[]; + let inputSeq: number; + let holdStop: boolean; + let stopStatus: number; + let settled: boolean; + let includeSequence: boolean; + let pendingStop: { response: ServerResponse; seq: number } | undefined; + let saved: ChatSessionPersistedState | null; + + function createTransport( + session: ChatSessionPersistedState, + options: Partial = {} + ) { + return new TriggerChatTransport({ + task: "test-chat", + baseURL, + accessToken: () => "test-token", + sessions: { chat: session }, + onSessionChange: (_chatId, session) => { + saved = session; + }, + ...options, + }); + } + + function appendResponse(response: ServerResponse, seq: number, status = 200) { + response + .writeHead(status, { "Content-Type": "application/json" }) + .end(JSON.stringify(includeSequence ? { seq } : {})); + } + + beforeEach(async () => { + outputs = []; + inputSeq = 10; + holdStop = false; + stopStatus = 200; + settled = false; + includeSequence = true; + pendingStop = undefined; + saved = null; + server = createServer(async (request, response) => { + if (request.method === "POST") { + let body = ""; + for await (const data of request) body += data; + const input: unknown = JSON.parse(body); + const isStop = + typeof input === "object" && input !== null && "kind" in input && input.kind === "stop"; + const seq = inputSeq++; + if (isStop && holdStop) { + pendingStop = { response, seq }; + } else { + appendResponse(response, seq, isStop ? stopStatus : 200); + } + return; + } + response.writeHead(200, { + "Content-Type": "text/event-stream", + "X-Stream-Version": "v2", + "X-Session-Settled": String(settled), + }); + response.flushHeaders(); + outputs.push(response); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Expected a TCP address"); + baseURL = `http://127.0.0.1:${address.port}`; + transport = createTransport({ publicAccessToken: "test-token" }); + }); + + afterEach(async () => { + transport.dispose(); + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + }); + + async function send(abortSignal?: AbortSignal) { + const before = outputs.length; + const stream = await transport.sendMessages({ + chatId: "chat", + trigger: "submit-message", + messageId: "user", + messages: [{ id: "user", role: "user", parts: [{ type: "text", text: "Continue" }] }], + abortSignal, + }); + await vi.waitFor(() => expect(outputs.length).toBeGreaterThan(before)); + return stream; + } + + function emit(records: OutputRecord[]) { + const response = outputs.at(-1); + if (!response || response.destroyed) throw new Error("The output subscription is closed"); + response.write(`event: batch\ndata: ${JSON.stringify({ records })}\n\n`); + } + + function oldTailAndReply(oldInput = 11, newInput = 12): OutputRecord[] { + return [ + chunk(4, { type: "tool-output-available", toolCallId: "old-tool", output: "Late output" }), + complete(5, oldInput), + ...reply(6), + complete(11, newInput), + ]; + } + + it.each([false, true])( + "keeps a successor before the Stop acknowledgment (resumed: %s)", + async (resumed) => { + const abort = new AbortController(); + let first: ReadableStream; + if (resumed) { + transport.setSession("chat", { publicAccessToken: "test-token", lastEventId: "1" }); + inputSeq = 11; + const stream = await transport.reconnectToStream({ + chatId: "chat", + abortSignal: abort.signal, + }); + if (!stream) throw new Error("Expected a resumed stream"); + first = stream; + await vi.waitFor(() => expect(outputs).toHaveLength(1)); + } else { + first = await send(abort.signal); + } + const reader = first.getReader(); + emit([ + chunk(2, { type: "start", messageId: "old" }), + chunk(3, { + type: "tool-input-available", + toolCallId: "old-tool", + toolName: "bash", + input: {}, + }), + ]); + await reader.read(); + await reader.read(); + // Resumed streams do not send Stop on abort. This matches useChat.stop(). + if (resumed) { + abort.abort(); + await reader.read(); + } + holdStop = true; + const stopped = transport.stopGeneration("chat"); + await vi.waitFor(() => expect(pendingStop).toBeDefined()); + const next = await send(); + const pending = pendingStop!; + appendResponse(pending.response, pending.seq); + expect(await stopped).toBe(true); + expect(transport.getSession("chat")?.isStreaming).toBe(true); + emit(oldTailAndReply()); + await expect(readText(next)).resolves.toBe("New response"); + } + ); + + it.each(["constructor", "setSession"] as const)( + "retains the stopped boundary through %s hydration", + async (hydrate) => { + await send(); + await transport.stopGeneration("chat"); + expect(saved).toMatchObject({ + skipToTurnComplete: true, + supersededInputSeq: 10, + isStreaming: false, + }); + const session = transport.getSession("chat"); + if (!session) throw new Error("Expected persisted state"); + transport.dispose(); + transport = createTransport( + hydrate === "constructor" ? session : { publicAccessToken: "test-token" } + ); + if (hydrate === "setSession") transport.setSession("chat", session); + const next = await send(); + emit(oldTailAndReply()); + await expect(readText(next)).resolves.toBe("New response"); + expect(saved).toMatchObject({ skipToTurnComplete: false, supersededInputSeq: undefined }); + } + ); + + it("retains unread output after a failed Stop request", async () => { + await send(); + stopStatus = 400; + expect(await transport.stopGeneration("chat")).toBe(false); + expect(transport.getSession("chat")?.isStreaming).toBe(false); + await vi.waitFor(() => expect(outputs[0]?.destroyed).toBe(true)); + const next = await send(); + emit(oldTailAndReply(10, 12)); + await expect(readText(next)).resolves.toBe("New response"); + }); + + it("does not resume a stopped owning consumer after hydration", async () => { + const abort = new AbortController(); + await send(abort.signal); + abort.abort(); + expect(saved).toMatchObject({ + skipToTurnComplete: true, + supersededInputSeq: 10, + isStreaming: false, + }); + if (!saved) throw new Error("Expected persisted state"); + transport.dispose(); + transport = createTransport(saved); + expect(await transport.reconnectToStream({ chatId: "chat" })).toBeNull(); + }); + + it("closes the SSE connection when the stopped boundary is missing", async () => { + await send(); + await transport.stopGeneration("chat"); + const next = await send(); + emit([...reply(1), complete(6, 12)]); + await expect(readText(next)).rejects.toThrow("The previous turn's output was lost"); + await vi.waitFor(() => expect(outputs.at(-1)?.destroyed).toBe(true)); + expect(saved).toMatchObject({ + skipToTurnComplete: false, + isStreaming: false, + activeInputSeq: undefined, + }); + }); + + it("does not retain an old stopped input after session recreation", async () => { + transport.dispose(); + transport = createTransport( + { publicAccessToken: "test-token" }, + { + startSession: async () => { + stopStatus = 200; + return { publicAccessToken: "replacement-token" }; + }, + } + ); + await send(); + stopStatus = 404; + expect(await transport.stopGeneration("chat")).toBe(true); + expect(saved).toMatchObject({ + skipToTurnComplete: false, + supersededInputSeq: undefined, + activeInputSeq: undefined, + }); + await transport.stopGeneration("chat"); + const next = await send(); + emit([...reply(1), complete(6, 14)]); + await expect(readText(next)).resolves.toBe("New response"); + }); + + it("does not gate a response after a completed turn", async () => { + const first = await send(); + emit([...reply(1), complete(6, 10)]); + await expect(readText(first)).resolves.toBe("New response"); + await transport.stopGeneration("chat"); + const next = await send(); + emit([...reply(7), complete(12, 12)]); + await expect(readText(next)).resolves.toBe("New response"); + }); + + it("retains the first stopped boundary after repeated Stop calls", async () => { + await send(); + await transport.stopGeneration("chat"); + await transport.stopGeneration("chat"); + const next = await send(); + emit(oldTailAndReply(12, 13)); + await expect(readText(next)).resolves.toBe("New response"); + }); + + it("does not stop a successor when an old consumer aborts after settled EOF", async () => { + const abort = new AbortController(); + settled = true; + const first = await send(abort.signal); + emit(reply(1)); + outputs[0]!.end(); + await expect(readText(first)).resolves.toBe("New response"); + settled = false; + const next = await send(); + abort.abort(); + expect(transport.getSession("chat")?.isStreaming).toBe(true); + emit([...reply(6), complete(11, 11)]); + await expect(readText(next)).resolves.toBe("New response"); + expect(inputSeq).toBe(12); + }); + + it.each(["message", "action"] as const)( + "requires transcript reload when a stopped successor %s has no sequence", + async (kind) => { + await send(); + await transport.stopGeneration("chat"); + includeSequence = false; + const reloadError = + "Stopped chat response cannot be matched. Reload the chat before sending another message."; + const next = kind === "message" ? send() : transport.sendAction("chat", { type: "undo" }); + await expect(next).rejects.toThrow(reloadError); + expect(saved).toMatchObject({ requiresTranscriptReload: true, isStreaming: false }); + expect(inputSeq).toBe(13); + await expect(send()).rejects.toThrow(reloadError); + await expect(transport.sendAction("chat", { type: "undo" })).rejects.toThrow(reloadError); + expect(inputSeq).toBe(13); + + if (!saved) throw new Error("Expected persisted state"); + transport.dispose(); + transport = createTransport(saved, { watch: true }); + await expect(send()).rejects.toThrow(reloadError); + expect(inputSeq).toBe(13); + expect(await transport.reconnectToStream({ chatId: "chat" })).toBeNull(); + expect(outputs).toHaveLength(1); + + // A fresh transcript supplies a cursor beyond the accepted response. + transport.dispose(); + transport = createTransport(saved); + transport.setSession("chat", { + publicAccessToken: "test-token", + lastEventId: "11", + isStreaming: false, + }); + includeSequence = true; + const afterReload = await send(); + emit([...reply(12), complete(17, 13)]); + await expect(readText(afterReload)).resolves.toBe("New response"); + } + ); + + it("accepts a sequence-free response without a stopped boundary", async () => { + includeSequence = false; + const stream = await send(); + emit([...reply(1), complete(6, 10)]); + await expect(readText(stream)).resolves.toBe("New response"); + }); + + it("persists a cleared boundary without rearming the abandoned turn after hydration", async () => { + await send(); + transport.clearSupersedeGate("chat"); + expect(saved).toMatchObject({ + skipToTurnComplete: false, + supersededInputSeq: undefined, + activeInputSeq: undefined, + isStreaming: false, + }); + if (!saved) throw new Error("Expected persisted state"); + transport.dispose(); + transport = createTransport(saved); + await transport.stopGeneration("chat"); + const next = await send(); + emit([...reply(1), complete(6, 12)]); + await expect(readText(next)).resolves.toBe("New response"); + }); +}); diff --git a/packages/trigger-sdk/src/v3/chat.test.ts b/packages/trigger-sdk/src/v3/chat.test.ts index c3157a5bbff..d75d8d7a547 100644 --- a/packages/trigger-sdk/src/v3/chat.test.ts +++ b/packages/trigger-sdk/src/v3/chat.test.ts @@ -1577,13 +1577,13 @@ describe("TriggerChatTransport", () => { ]); }); - it("keeps the gate out of the persisted session", async () => { + it("persists the stopped boundary", async () => { mockFetch([() => defaultSseResponse()]); const sessions: Record = {}; const transport = await armedGate( "chat-persist", - { publicAccessToken: "p" }, + { publicAccessToken: "p", isStreaming: true, activeInputSeq: 5 }, { onSessionChange: (chatId, session) => { sessions[chatId] = session; @@ -1591,8 +1591,14 @@ describe("TriggerChatTransport", () => { } ); - expect(transport.getSession("chat-persist")).not.toHaveProperty("skipToTurnComplete"); - expect(sessions["chat-persist"]).not.toHaveProperty("skipToTurnComplete"); + expect(transport.getSession("chat-persist")).toMatchObject({ + skipToTurnComplete: true, + supersededInputSeq: 5, + }); + expect(sessions["chat-persist"]).toMatchObject({ + skipToTurnComplete: true, + supersededInputSeq: 5, + }); }); it("clears on the first turn-complete after two consecutive stops", async () => { diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts index b3c10316df9..3c16665d530 100644 --- a/packages/trigger-sdk/src/v3/chat.ts +++ b/packages/trigger-sdk/src/v3/chat.ts @@ -496,6 +496,12 @@ export type ChatSessionPersistedState = { /** The `.in` append sequence of the last send this client owned; reused as `sinceInSeq` on reconnect. */ activeInputSeq?: number; isStreaming?: boolean; + /** Discard unread output from a stopped turn before the next response. */ + skipToTurnComplete?: boolean; + /** The stopped input sequence excludes older completion records from the boundary. */ + supersededInputSeq?: number; + /** A send lacks an input sequence while stopped output remains unread. Reload the transcript before another send. */ + requiresTranscriptReload?: boolean; /** Set once the session is closed. Persisted so a reload doesn't retry a dead session. */ closed?: boolean; /** The reason the session was closed, when one was given. */ @@ -713,6 +719,7 @@ type ChatSessionState = { skipToTurnComplete?: boolean; /** `.in` seq of the turn the gate supersedes; only its boundary (or a later one) clears the gate. */ supersededInputSeq?: number; + requiresTranscriptReload?: boolean; /** Whether the agent is currently streaming a response. Set on first chunk, cleared on turn-complete. */ isStreaming?: boolean; /** Set once the outstanding turn is declared dead: a later stop must not gate the next turn on it. */ @@ -807,6 +814,9 @@ export class TriggerChatTransport implements ChatTransport { lastEventId: session.lastEventId, activeInputSeq: session.activeInputSeq, isStreaming: session.isStreaming, + skipToTurnComplete: session.skipToTurnComplete, + supersededInputSeq: session.supersededInputSeq, + requiresTranscriptReload: session.requiresTranscriptReload, closed: session.closed, closedReason: session.closedReason, }); @@ -951,6 +961,7 @@ export class TriggerChatTransport implements ChatTransport { // Generated outside the closure so auth-retries reuse the same part id // and the server-side dedupe sees one logical append. + this.assertTranscriptReady(chatId, state); const partId = crypto.randomUUID(); const serializedBody = this.serializeInputChunk({ kind: "message", payload: wirePayload }); const sendChatMessage = (token: string) => @@ -975,6 +986,7 @@ export class TriggerChatTransport implements ChatTransport { } state.activeInputSeq = inSeq; + this.requireStoppedTurnCorrelation(chatId, state, inSeq); state.isStreaming = true; state.outstandingTurnAbandoned = false; this.notifySessionChange(chatId, state); @@ -1280,6 +1292,7 @@ export class TriggerChatTransport implements ChatTransport { if (!state) return null; // A closed session has no further turns to resume. if (state.closed) return null; + if (state.requiresTranscriptReload) return null; // Watch is a standing subscription: a settled session is exactly the // state it waits in, so a completed last turn must not block the resume. @@ -1315,24 +1328,8 @@ export class TriggerChatTransport implements ChatTransport { stopGeneration = async (chatId: string): Promise => { const state = this.sessions.get(chatId); if (!state) return false; - - const partId = crypto.randomUUID(); - const serializedBody = this.serializeInputChunk({ kind: "stop" }); - const send = async (token: string) => { - await this.appendInputChunk(chatId, token, serializedBody, partId); - }; - - try { - await this.sendWithEvents( - chatId, - "stop", - { partId, bodyBytes: byteLength(serializedBody) }, - () => this.callWithAuthRetry(chatId, state, send) - ); - } catch { - return false; - } - + // Close the captured turn before the request awaits. A delayed acknowledgment + // must not change a successor's reader or stopped-output boundary. // Only gate when a sent turn is still outstanding. A stop at a boundary has // nothing to supersede, and gating it would swallow the next turn. if ( @@ -1360,7 +1357,24 @@ export class TriggerChatTransport implements ChatTransport { // explicitly stopped. state.isStreaming = false; this.notifySessionChange(chatId, state); - return true; + + const partId = crypto.randomUUID(); + const serializedBody = this.serializeInputChunk({ kind: "stop" }); + const send = async (token: string) => { + await this.appendInputChunk(chatId, token, serializedBody, partId); + }; + try { + await this.sendWithEvents( + chatId, + "stop", + { partId, bodyBytes: byteLength(serializedBody) }, + () => this.callWithAuthRetry(chatId, state, send) + ); + return true; + } catch { + // The reader already closed. Retain its unread boundary for the next send. + return false; + } }; /** @@ -1373,7 +1387,10 @@ export class TriggerChatTransport implements ChatTransport { if (!state) return; state.skipToTurnComplete = false; state.supersededInputSeq = undefined; + state.activeInputSeq = undefined; state.outstandingTurnAbandoned = true; + state.isStreaming = false; + this.notifySessionChange(chatId, state); }; /** @@ -1408,6 +1425,7 @@ export class TriggerChatTransport implements ChatTransport { : undefined, }; + this.assertTranscriptReady(chatId, state); const body = this.serializeInputChunk({ kind: "message", payload: wirePayload }); const partId = crypto.randomUUID(); const send = (token: string) => this.appendInputChunk(chatId, token, body, partId); @@ -1432,6 +1450,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; + this.requireStoppedTurnCorrelation(chatId, state, inSeq); state.isStreaming = true; state.outstandingTurnAbandoned = false; this.notifySessionChange(chatId, state); @@ -1461,6 +1480,9 @@ export class TriggerChatTransport implements ChatTransport { lastEventId: session.lastEventId, activeInputSeq: session.activeInputSeq, isStreaming: session.isStreaming, + skipToTurnComplete: session.skipToTurnComplete, + supersededInputSeq: session.supersededInputSeq, + requiresTranscriptReload: session.requiresTranscriptReload, }) ); this.notifySessionChange(chatId, this.toPersisted(this.sessions.get(chatId)!)); @@ -1645,11 +1667,35 @@ export class TriggerChatTransport implements ChatTransport { return JSON.stringify(chunk); } + private assertTranscriptReady(chatId: string, state: ChatSessionState): void { + if (!state.requiresTranscriptReload) return; + this.coordinator?.release(chatId); + throw new Error( + "Stopped chat response cannot be matched. Reload the chat before sending another message." + ); + } + + private requireStoppedTurnCorrelation( + chatId: string, + state: ChatSessionState, + inSeq: number | undefined + ): void { + if (!state.skipToTurnComplete || inSeq !== undefined) return; + // The server accepted the prompt. A retry can create a duplicate turn. + state.requiresTranscriptReload = true; + state.isStreaming = false; + this.notifySessionChange(chatId, state); + this.assertTranscriptReady(chatId, state); + } + private toPersisted = (state: ChatSessionState): ChatSessionPersistedState => ({ publicAccessToken: state.publicAccessToken, lastEventId: state.lastEventId, activeInputSeq: state.activeInputSeq, isStreaming: state.isStreaming, + skipToTurnComplete: state.skipToTurnComplete, + supersededInputSeq: state.supersededInputSeq, + requiresTranscriptReload: state.requiresTranscriptReload, closed: state.closed, closedReason: state.closedReason, }); @@ -1961,7 +2007,11 @@ export class TriggerChatTransport implements ChatTransport { } state.publicAccessToken = publicAccessToken; state.lastEventId = undefined; + state.activeInputSeq = undefined; state.isStreaming = false; + state.skipToTurnComplete = false; + state.supersededInputSeq = undefined; + state.requiresTranscriptReload = false; this.sessions.set(chatId, state); this.notifySessionChange(chatId, state); } @@ -2002,9 +2052,16 @@ export class TriggerChatTransport implements ChatTransport { const outstanding = !state.outstandingTurnAbandoned && (state.isStreaming || state.activeInputSeq !== undefined); - if (options?.sendStopOnAbort !== false && outstanding && !internalAbort.signal.aborted) { + if ( + options?.sendStopOnAbort !== false && + outstanding && + !internalAbort.signal.aborted && + this.activeStreams.get(chatId) === internalAbort + ) { state.skipToTurnComplete = true; state.supersededInputSeq = state.activeInputSeq; + state.isStreaming = false; + this.notifySessionChange(chatId, state); this.appendInputChunk( chatId, state.publicAccessToken, @@ -2146,7 +2203,7 @@ export class TriggerChatTransport implements ChatTransport { if (opened) return opened; } - // A settled session or an abort ends the turn cleanly. Exhausting the + // A settled session or an abort ends the subscription cleanly. Exhausting the // resubscribe budget while the turn is still streaming means it was cut // off — surface an error so the UI doesn't read a truncated reply as // complete. The caller's catch emits stream-error and errors the stream. @@ -2160,9 +2217,13 @@ export class TriggerChatTransport implements ChatTransport { ); } - // Settled close, or the turn is gone — tell the UI instead of - // leaving it spinning on a stream nobody will finish. - if (state.isStreaming && this.activeStreams.get(chatId) === internalAbort) { + // A passive abort closes this view, not the remote turn. Only a + // settled subscription changes the turn's persisted streaming state. + if ( + state.isStreaming && + !combinedSignal.aborted && + this.activeStreams.get(chatId) === internalAbort + ) { state.isStreaming = false; this.notifySessionChange(chatId, state); } @@ -2297,6 +2358,7 @@ export class TriggerChatTransport implements ChatTransport { } state.skipToTurnComplete = false; state.supersededInputSeq = undefined; + this.notifySessionChange(chatId, state); // This boundary is the new turn's own, so the gate swallowed its // output: fail the turn instead of completing an empty answer, and // leave nothing armed for the retry. @@ -2416,6 +2478,12 @@ export class TriggerChatTransport implements ChatTransport { // unwrapped from the S2 record envelope (the parser does the // JSON unwrap). Drop empty/malformed payloads defensively. if (value.chunk == null) continue; + // A resumed session can contain only a token and output cursor. + // Its first data record establishes an active turn for Stop. + if (!state.outstandingTurnAbandoned && state.isStreaming !== true) { + state.isStreaming = true; + this.notifySessionChange(chatId, state); + } if (!sawFirstChunk) { sawFirstChunk = true; this.emitEvent({ @@ -2430,6 +2498,7 @@ export class TriggerChatTransport implements ChatTransport { controller.enqueue(value.chunk as UIMessageChunk); } } catch (error) { + internalAbort.abort(); if (error instanceof Error && error.name === "AbortError") { try { controller.close(); diff --git a/packages/trigger-sdk/test/chat-transport-events.test.ts b/packages/trigger-sdk/test/chat-transport-events.test.ts index 2c4d24b85a0..99c29d3bb68 100644 --- a/packages/trigger-sdk/test/chat-transport-events.test.ts +++ b/packages/trigger-sdk/test/chat-transport-events.test.ts @@ -175,68 +175,30 @@ describe("transport send events", () => { }); describe("stopped turn followed by a new turn", () => { - /** - * `.out` stub that honours the `Last-Event-ID` cursor like the server does, so - * a resubscribe cannot replay records the reader already consumed. Legacy v1 - * frames carry no `session-in-event-id`, so the stopped turn's boundary is - * indistinguishable from this turn's: the tail is dropped and the turn closes. - */ - function cursoredTwoTurnTransport() { - const frames = [ - { id: "1", data: `{"type":"text-delta","id":"t1","delta":"stale"}` }, - { id: "2", data: `{"type":"trigger:turn-complete"}` }, - ]; - - return makeTransport({ - sessions: { c1: { publicAccessToken: "tok_test", isStreaming: true } }, - fetch: async (_url, init, ctx) => { - if (ctx.endpoint === "in") return jsonOk(); - - const cursor = new Headers(init.headers).get("Last-Event-ID"); - const from = cursor ? frames.findIndex((f) => f.id === cursor) + 1 : 0; - const remaining = frames.slice(from); - const response = sseResponse( - remaining.map((f) => `id: ${f.id}\ndata: ${f.data}\n\n`).join("") - ); - // Nothing left to send: the session is settled, so the reader stops - // instead of resubscribing. - if (remaining.length === 0) response.headers.set("X-Session-Settled", "true"); - return response; - }, - }); - } - - it("drops the stopped turn's tail and closes the sendMessages turn", async () => { - const { transport, events } = cursoredTwoTurnTransport(); - - expect(await transport.stopGeneration("c1")).toBe(true); - events.length = 0; - - const stream = await transport.sendMessages({ - trigger: "submit-message", - chatId: "c1", - messageId: undefined, - messages: [user("after stop", "u-2")], - abortSignal: undefined, - }); - const chunks = await readAll(stream); - - expect(chunks).toEqual([]); - expect(events.some((e) => e.type === "turn-completed")).toBe(true); - }); - - it("drops the stopped turn's tail and closes the sendAction turn", async () => { - const { transport, events } = cursoredTwoTurnTransport(); - - expect(await transport.stopGeneration("c1")).toBe(true); - events.length = 0; - - const stream = await transport.sendAction("c1", { type: "undo" }); - const chunks = await readAll(stream); - - expect(chunks).toEqual([]); - expect(events.some((e) => e.type === "turn-completed")).toBe(true); - }); + it.each(["message", "action"] as const)( + "does not report completion for an uncorrelated %s", + async (kind) => { + const { transport, events } = makeTransport({ + sessions: { c1: { publicAccessToken: "tok_test", isStreaming: true } }, + }); + expect(await transport.stopGeneration("c1")).toBe(true); + events.length = 0; + const sent = + kind === "action" + ? transport.sendAction("c1", { type: "undo" }) + : transport.sendMessages({ + trigger: "submit-message", + chatId: "c1", + messageId: undefined, + messages: [user("after stop", "u-2")], + abortSignal: undefined, + }); + await expect(sent).rejects.toThrow("Reload the chat before sending another message"); + expect(events.some((e) => e.type === "message-sent")).toBe(true); + expect(events.some((e) => e.type === "stream-connected")).toBe(false); + expect(events.some((e) => e.type === "turn-completed")).toBe(false); + } + ); }); describe("transport stream events", () => {