From f26770be701862a41798de902bdb02c6626fdc8d Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 23 Jul 2026 15:51:49 +0100 Subject: [PATCH 1/3] feat(sdk,core): close resumed chat streams promptly when caught up Resuming a chat session stream (page reload or reconnect) held the SSE connection open for the whole long-poll window even after every buffered output record had already arrived. The client now detects when it has caught up to the latest output and closes the resumed stream right away, so reconnecting to an idle chat settles immediately. Detection reuses the stream's tail-carrying heartbeat: batch and ping frames now carry the tail, and CaughtUpTracker from @s2-dev/streamstore 0.25.0 turns "last delivered seq + 1 === tail" into a caught-up signal. When the tail is absent (older self-hosted stream backends) the client keeps its previous behavior, so nothing regresses. Also moves to the current S2 hosts that 0.25.0 defaults to. --- .changeset/chat-session-caught-up-resume.md | 7 + .../realtime/s2realtimeStreams.server.ts | 4 +- .../realtime/streamBasinProvisioner.server.ts | 4 +- apps/webapp/package.json | 2 +- packages/cli-v3/package.json | 2 +- packages/core/package.json | 2 +- packages/core/src/v3/apiClient/index.ts | 10 ++ .../core/src/v3/apiClient/runStream.test.ts | 130 ++++++++++++++++++ packages/core/src/v3/apiClient/runStream.ts | 51 +++++++ .../core/src/v3/sessionStreams/manager.ts | 5 + packages/trigger-sdk/src/v3/chat.ts | 15 +- pnpm-lock.yaml | 18 +-- 12 files changed, 233 insertions(+), 17 deletions(-) create mode 100644 .changeset/chat-session-caught-up-resume.md diff --git a/.changeset/chat-session-caught-up-resume.md b/.changeset/chat-session-caught-up-resume.md new file mode 100644 index 00000000000..f97c3f6b911 --- /dev/null +++ b/.changeset/chat-session-caught-up-resume.md @@ -0,0 +1,7 @@ +--- +"@trigger.dev/core": patch +"@trigger.dev/sdk": patch +--- + +Chat sessions now close a resumed stream as soon as it has caught up to the latest output, instead of holding the connection open for the full long-poll window. Reloading or reconnecting to an idle chat settles faster. + diff --git a/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts b/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts index 787b3f1ab44..428163d3f8d 100644 --- a/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts +++ b/apps/webapp/app/services/realtime/s2realtimeStreams.server.ts @@ -107,8 +107,8 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor { constructor(opts: S2RealtimeStreamsOptions) { this.basin = opts.basin; - this.baseUrl = opts.endpoint ?? `https://${this.basin}.b.aws.s2.dev/v1`; - this.accountUrl = opts.endpoint ?? `https://aws.s2.dev/v1`; + this.baseUrl = opts.endpoint ?? `https://${this.basin}.b.s2.dev/v1`; + this.accountUrl = opts.endpoint ?? `https://a.s2.dev/v1`; this.endpoint = opts.endpoint; this.token = opts.accessToken; this.streamPrefix = opts.streamPrefix ?? ""; diff --git a/apps/webapp/app/services/realtime/streamBasinProvisioner.server.ts b/apps/webapp/app/services/realtime/streamBasinProvisioner.server.ts index 98316b84fe8..d4a6d1738ee 100644 --- a/apps/webapp/app/services/realtime/streamBasinProvisioner.server.ts +++ b/apps/webapp/app/services/realtime/streamBasinProvisioner.server.ts @@ -185,7 +185,7 @@ type CreateBasinOptions = { }; async function s2CreateBasin(name: string, opts: CreateBasinOptions): Promise { - const url = `https://aws.s2.dev/v1/basins`; + const url = `https://a.s2.dev/v1/basins`; const body = { basin: name, config: { @@ -222,7 +222,7 @@ type ReconfigureBasinOptions = { }; async function s2ReconfigureBasin(name: string, opts: ReconfigureBasinOptions): Promise { - const url = `https://aws.s2.dev/v1/basins/${encodeURIComponent(name)}`; + const url = `https://a.s2.dev/v1/basins/${encodeURIComponent(name)}`; const body = { default_stream_config: { retention_policy: { age: parseDuration(opts.retentionPolicy) }, diff --git a/apps/webapp/package.json b/apps/webapp/package.json index 678eeb76033..e4d72f05e77 100644 --- a/apps/webapp/package.json +++ b/apps/webapp/package.json @@ -103,7 +103,7 @@ "@remix-run/react": "2.17.5", "@remix-run/router": "^1.23.3", "@remix-run/server-runtime": "2.17.5", - "@s2-dev/streamstore": "^0.22.10", + "@s2-dev/streamstore": "^0.25.0", "@sentry/remix": "9.46.0", "@slack/web-api": "7.16.0", "@socket.io/redis-adapter": "^8.3.0", diff --git a/packages/cli-v3/package.json b/packages/cli-v3/package.json index c6d8d9719dc..318ff9268d5 100644 --- a/packages/cli-v3/package.json +++ b/packages/cli-v3/package.json @@ -95,7 +95,7 @@ "@opentelemetry/resources": "2.7.1", "@opentelemetry/sdk-trace-node": "2.7.1", "@opentelemetry/semantic-conventions": "1.41.1", - "@s2-dev/streamstore": "^0.22.10", + "@s2-dev/streamstore": "^0.25.0", "@trigger.dev/build": "workspace:4.5.7", "@trigger.dev/core": "workspace:4.5.7", "@trigger.dev/schema-to-json": "workspace:4.5.7", diff --git a/packages/core/package.json b/packages/core/package.json index f0c3adfcf2f..7eb81b899ed 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -207,7 +207,7 @@ "@opentelemetry/sdk-trace-base": "2.7.1", "@opentelemetry/sdk-trace-node": "2.7.1", "@opentelemetry/semantic-conventions": "1.41.1", - "@s2-dev/streamstore": "0.22.10", + "@s2-dev/streamstore": "0.25.0", "dequal": "^2.0.3", "eventsource": "^3.0.5", "eventsource-parser": "^3.0.0", diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts index deda5d01f9a..21ca40f4bbc 100644 --- a/packages/core/src/v3/apiClient/index.ts +++ b/packages/core/src/v3/apiClient/index.ts @@ -1410,6 +1410,13 @@ export class ApiClient { * enqueued into the consumer stream — handle the event here. */ onControl?: (event: ControlEvent) => void; + /** + * Fires once when the session reaches the live tail (backlog drained), + * with the observed tail position. No-op when the backend omits the + * tail-carrying heartbeat. Mirrors the browser transport's caught-up + * signal on the worker / apiClient read path. + */ + onCaughtUp?: (tail: { seqNum: number; timestamp: Date }) => void; } ): Promise> { const url = `${options?.baseUrl ?? this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}`; @@ -1424,6 +1431,9 @@ export class ApiClient { }); const stream = await subscription.subscribe(); + if (options?.onCaughtUp) { + subscription.caughtUp().then(options.onCaughtUp).catch(() => {}); + } const onPart = options?.onPart; const onControl = options?.onControl; diff --git a/packages/core/src/v3/apiClient/runStream.test.ts b/packages/core/src/v3/apiClient/runStream.test.ts index 0dca73779af..df8b6325e37 100644 --- a/packages/core/src/v3/apiClient/runStream.test.ts +++ b/packages/core/src/v3/apiClient/runStream.test.ts @@ -600,3 +600,133 @@ describe("SSEStreamSubscription v2 batch parsing — record kinds", () => { expect((parts[1]!.chunk as any).delta).toBe("x"); }); }); + +describe("SSEStreamSubscription caught-up tracking", () => { + const originalFetch = globalThis.fetch; + + afterEach(() => { + globalThis.fetch = originalFetch; + vi.restoreAllMocks(); + }); + + type Rec = { body: string; seq_num: number; timestamp: number; headers?: Array<[string, string]> }; + type Tail = { seq_num: number; timestamp: number }; + + function dataRec(seq: number): Rec { + return { + body: JSON.stringify({ data: { type: "text-delta", delta: "x" }, id: `p${seq}` }), + seq_num: seq, + timestamp: 1, + headers: [], + }; + } + + function batchEvent(records: Rec[], tail?: Tail): string { + const data = tail ? { records, tail } : { records }; + return `event: batch\ndata: ${JSON.stringify(data)}\n\n`; + } + + function pingEvent(tail?: Tail): string { + const data = tail ? { timestamp: 1, tail } : { timestamp: 1 }; + return `event: ping\ndata: ${JSON.stringify(data)}\n\n`; + } + + function makeEventsResponse(events: string[]) { + const body = new ReadableStream({ + start(controller) { + for (const e of events) controller.enqueue(new TextEncoder().encode(e)); + controller.close(); + }, + }); + return new Response(body, { + status: 200, + headers: { "Content-Type": "text/event-stream", "X-Stream-Version": "v2" }, + }); + } + + async function drain(stream: ReadableStream<{ id: string; chunk: unknown }>) { + const reader = stream.getReader(); + const parts: Array<{ id: string; chunk: unknown }> = []; + while (true) { + const { done, value } = await reader.read(); + if (done) { + reader.releaseLock(); + return parts; + } + parts.push(value); + } + } + + it("resolves caughtUp() when a batch reaches the reported tail", async () => { + globalThis.fetch = vi + .fn() + .mockResolvedValue( + makeEventsResponse([ + batchEvent([dataRec(0), dataRec(1), dataRec(2)], { seq_num: 3, timestamp: 1 }), + ]) + ); + const sub = new SSEStreamSubscription("http://x", { maxRetries: 0 }); + const stream = await sub.subscribe(); + const cu = sub.caughtUp(); + await drain(stream); + const tail = await cu; + expect(tail.seqNum).toBe(3); + expect(sub.isCaughtUp()).toBe(true); + }); + + it("stays behind when the batch does not reach the tail", async () => { + globalThis.fetch = vi + .fn() + .mockResolvedValue(makeEventsResponse([batchEvent([dataRec(0)], { seq_num: 3, timestamp: 1 })])); + const sub = new SSEStreamSubscription("http://x", { maxRetries: 0 }); + const stream = await sub.subscribe(); + await drain(stream); + expect(sub.isCaughtUp()).toBe(false); + }); + + it("resolves caughtUp() from a ping tail with an empty backlog (open-at-tail)", async () => { + globalThis.fetch = vi + .fn() + .mockResolvedValue(makeEventsResponse([pingEvent({ seq_num: 3, timestamp: 1 })])); + const sub = new SSEStreamSubscription("http://x", { maxRetries: 0 }); + const stream = await sub.subscribe(); + const cu = sub.caughtUp(); + await drain(stream); + const tail = await cu; + expect(tail.seqNum).toBe(3); + expect(sub.isCaughtUp()).toBe(true); + }); + + it("reaches caught-up when the tail is a trim command record (raw counts include it)", async () => { + globalThis.fetch = vi.fn().mockResolvedValue( + makeEventsResponse([ + batchEvent( + [ + dataRec(0), + { body: "", seq_num: 1, timestamp: 1, headers: [["trigger-control", "turn-complete"]] }, + { body: "AAAAAAAAAAQ=", seq_num: 2, timestamp: 1, headers: [["", "trim"]] }, + ], + { seq_num: 3, timestamp: 1 } + ), + ]) + ); + const sub = new SSEStreamSubscription("http://x", { maxRetries: 0 }); + const stream = await sub.subscribe(); + const cu = sub.caughtUp(); + const parts = await drain(stream); + const tail = await cu; + expect(tail.seqNum).toBe(3); + expect(sub.isCaughtUp()).toBe(true); + expect(parts).toHaveLength(2); + }); + + it("never reaches caught-up when the wire carries no tail (feature-detect fallback)", async () => { + globalThis.fetch = vi + .fn() + .mockResolvedValue(makeEventsResponse([batchEvent([dataRec(0), dataRec(1), dataRec(2)]), pingEvent()])); + const sub = new SSEStreamSubscription("http://x", { maxRetries: 0 }); + const stream = await sub.subscribe(); + await drain(stream); + expect(sub.isCaughtUp()).toBe(false); + }); +}); diff --git a/packages/core/src/v3/apiClient/runStream.ts b/packages/core/src/v3/apiClient/runStream.ts index 5bd7986c86a..c3904aa7c00 100644 --- a/packages/core/src/v3/apiClient/runStream.ts +++ b/packages/core/src/v3/apiClient/runStream.ts @@ -14,6 +14,7 @@ import { conditionallyImportAndParsePacket, parsePacket } from "../utils/ioSeria import { ApiError, isTriggerRealtimeAuthError } from "./errors.js"; import type { ApiClient } from "./index.js"; import { zodShapeStream } from "./stream.js"; +import { CaughtUpTracker } from "@s2-dev/streamstore"; export type RunShape = TRunTypes extends AnyRunTypes ? { @@ -197,6 +198,7 @@ export class SSEStreamSubscription implements StreamSubscription { private nonRetryableStatuses: ReadonlySet; private retryNowController: AbortController | null = null; private internalAbort: AbortController | null = null; + private caughtUpTracker = new CaughtUpTracker(); constructor( private url: string, @@ -278,6 +280,26 @@ export class SSEStreamSubscription implements StreamSubscription { this.retryNowController?.abort(); } + /** + * True once this session has consumed everything up to the tail the server + * last reported (via a batch `tail` or a heartbeat `ping`). Resets to false + * on reconnect and while records remain before the reported tail. Backed by + * S2's `CaughtUpTracker`, fed from the v2 batch/ping wire signals. + */ + isCaughtUp(): boolean { + return this.caughtUpTracker.isCaughtUp(); + } + + /** + * Resolves when this session reaches the live tail (backlog drained), + * carrying the last observed tail position. Resolves immediately if already + * caught up; call again after falling behind. Rejects if the stream ends + * before catching up. Stays pending across internal reconnects. + */ + caughtUp(): ReturnType { + return this.caughtUpTracker.caughtUp(); + } + async subscribe(): Promise> { // eslint-disable-next-line no-this-alias const self = this; @@ -407,9 +429,18 @@ export class SSEStreamSubscription implements StreamSubscription { timestamp: number; headers?: Array<[string, string]>; }>; + tail?: { seq_num: number; timestamp: number }; }; if (!data || !Array.isArray(data.records)) return; + const boundary = this.caughtUpTracker.observeBatch({ + recordCount: data.records.length, + lastSeqNum: data.records.at(-1)?.seq_num, + tail: data.tail + ? { seqNum: data.tail.seq_num, timestamp: new Date(data.tail.timestamp) } + : undefined, + }); + for (const record of data.records) { // Always advance the resume cursor — even for records we // skip — so a future Last-Event-ID reconnect lands past @@ -444,6 +475,22 @@ export class SSEStreamSubscription implements StreamSubscription { headers: record.headers ?? [], }); } + + boundary?.markDelivered(); + } else if (chunk.event === "ping") { + const ping = safeParseJSON(chunk.data) as + | { tail?: { seq_num: number; timestamp: number } } + | undefined; + if (ping?.tail) { + const pingBoundary = this.caughtUpTracker.observeBatch({ + recordCount: 0, + tail: { + seqNum: ping.tail.seq_num, + timestamp: new Date(ping.tail.timestamp), + }, + }); + pingBoundary?.markDelivered(); + } } } }, @@ -458,6 +505,7 @@ export class SSEStreamSubscription implements StreamSubscription { if (done) { reader.releaseLock(); + this.caughtUpTracker.end(); controller.close(); this.options.onComplete?.(); return; @@ -490,6 +538,7 @@ export class SSEStreamSubscription implements StreamSubscription { // `onError` was already invoked in the `!response.ok` branch above // (where the auth ApiError was originally constructed and thrown). // Auth errors are non-retryable: terminate the stream cleanly. + this.caughtUpTracker.end(); controller.error(error as Error); return; } @@ -513,6 +562,7 @@ export class SSEStreamSubscription implements StreamSubscription { if (this.retryCount >= this.maxRetries) { const finalError = error || new Error("Max retries reached"); + this.caughtUpTracker.end(); controller.error(finalError); this.options.onError?.(finalError); return; @@ -554,6 +604,7 @@ export class SSEStreamSubscription implements StreamSubscription { } // Reconnect + this.caughtUpTracker.reconnect(); await this.connectStream(controller); } } diff --git a/packages/core/src/v3/sessionStreams/manager.ts b/packages/core/src/v3/sessionStreams/manager.ts index fb87b211643..6f22757e589 100644 --- a/packages/core/src/v3/sessionStreams/manager.ts +++ b/packages/core/src/v3/sessionStreams/manager.ts @@ -466,6 +466,11 @@ export class StandardSessionStreamManager implements SessionStreamManager { console.error(`[SessionStreamManager] Tail error for "${key}":`, error); } }, + onCaughtUp: (tail) => { + if (this.debug) { + console.log(`[SessionStreamManager] Caught up on "${key}" at seq ${tail.seqNum}`); + } + }, }); // Drain to keep the pipeThrough flowing. Records were already diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts index 9dddac1c0eb..ae02855aced 100644 --- a/packages/trigger-sdk/src/v3/chat.ts +++ b/packages/trigger-sdk/src/v3/chat.ts @@ -1750,7 +1750,7 @@ export class TriggerChatTransport implements ChatTransport { reader.releaseLock(); return null; } - return { reader, primed: first.value }; + return { reader, primed: first.value, subscription }; } catch (readErr) { reader.releaseLock(); throw readErr; @@ -1764,6 +1764,7 @@ export class TriggerChatTransport implements ChatTransport { timestamp: number; }>; let primed: { id: string; chunk: unknown; timestamp: number } | undefined; + let sub: SSEStreamSubscription | undefined; try { const opened = await connectSseOnce(state.publicAccessToken); @@ -1773,6 +1774,7 @@ export class TriggerChatTransport implements ChatTransport { } reader = opened.reader; primed = opened.primed; + sub = opened.subscription; } catch (e) { if (isAuthError(e)) { const fresh = await this.resolveAccessToken({ chatId }); @@ -1800,6 +1802,17 @@ export class TriggerChatTransport implements ChatTransport { }); let sawFirstChunk = false; + if (options?.peekSettled && sub) { + sub + .caughtUp() + .then(() => { + if (!sawFirstChunk && !combinedSignal.aborted) { + internalAbort.abort(); + } + }) + .catch(() => {}); + } + while (true) { let value: { id: string; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index cd2d2d93fba..75156104357 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -436,8 +436,8 @@ importers: specifier: 2.17.5 version: 2.17.5(typescript@6.0.3) '@s2-dev/streamstore': - specifier: ^0.22.10 - version: 0.22.10(supports-color@10.0.0) + specifier: ^0.25.0 + version: 0.25.0(supports-color@10.0.0) '@sentry/remix': specifier: 9.46.0 version: 9.46.0(patch_hash=146126b032581925294aaed63ab53ce3f5e0356a755f1763d7a9a76b9846943b)(@remix-run/node@2.17.5(typescript@6.0.3))(@remix-run/react@2.17.5(react-dom@18.3.1(react@18.3.1))(react@18.3.1)(typescript@6.0.3))(@remix-run/server-runtime@2.17.5(typescript@6.0.3))(encoding@0.1.13)(react@18.3.1) @@ -1490,8 +1490,8 @@ importers: specifier: 1.41.1 version: 1.41.1 '@s2-dev/streamstore': - specifier: ^0.22.10 - version: 0.22.10(supports-color@10.0.0) + specifier: ^0.25.0 + version: 0.25.0(supports-color@10.0.0) '@trigger.dev/build': specifier: workspace:4.5.7 version: link:../build @@ -1761,8 +1761,8 @@ importers: specifier: 1.41.1 version: 1.41.1 '@s2-dev/streamstore': - specifier: 0.22.10 - version: 0.22.10(supports-color@10.0.0) + specifier: 0.25.0 + version: 0.25.0(supports-color@10.0.0) dequal: specifier: ^2.0.3 version: 2.0.3 @@ -6755,8 +6755,8 @@ packages: cpu: [x64] os: [win32] - '@s2-dev/streamstore@0.22.10': - resolution: {integrity: sha512-dtm+oFHVE8szINwOUoNQdx9xpGSJOrcAEvsxspPFvomjYKGnmhIRmU4OX8o6kxcPoiK76S1tPeU0smjZdmOngA==} + '@s2-dev/streamstore@0.25.0': + resolution: {integrity: sha512-oB8OJObT/s2Q68Tr01NLfDnQPFJp/Kn2pP3seWybgP7VrjQ+j1efXReOsyVCDsUBVMxkyPlzt75o/+GuL5Q3vg==} '@sec-ant/readable-stream@0.4.1': resolution: {integrity: sha512-831qok9r2t8AlxLko40y2ebgSDhenenCatLVeW/uBtnHPyhHOvG0C7TvfgecV+wHzIm5KUICgzmVpWS+IMEAeg==} @@ -21177,7 +21177,7 @@ snapshots: '@rollup/rollup-win32-x64-msvc@4.60.1': optional: true - '@s2-dev/streamstore@0.22.10(supports-color@10.0.0)': + '@s2-dev/streamstore@0.25.0(supports-color@10.0.0)': dependencies: '@protobuf-ts/runtime': 2.11.1 debug: 4.4.3(supports-color@10.0.0) From 940d60f57d01f6cb74443c14bb4311308e17a6b0 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 23 Jul 2026 15:52:17 +0100 Subject: [PATCH 2/3] fix(webapp): stop the dev server sending duplicate CORS headers In local development, Vite's dev-server CORS middleware reflected the request Origin on every Express response, on top of the app's own CORS handling. The two layers produced a duplicated Access-Control-Allow-Origin header, which browsers reject, breaking cross-origin API calls from a separate frontend in dev. Disabling Vite's dev CORS lets the app be the single source of CORS headers. Dev-only; production is unaffected. --- apps/webapp/vite.config.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/webapp/vite.config.ts b/apps/webapp/vite.config.ts index b008ae04ed0..0b33a4edb6f 100644 --- a/apps/webapp/vite.config.ts +++ b/apps/webapp/vite.config.ts @@ -30,6 +30,7 @@ export default defineConfig({ }, }, server: { + cors: false, warmup: { clientFiles: ["./app/entry.client.tsx", "./app/root.tsx", "./app/components/**/*.tsx"], ssrFiles: ["./app/entry.server.tsx", "./app/root.tsx"], From 87258154b7c607e02d3869caec8341245baaf8e5 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 23 Jul 2026 16:05:46 +0100 Subject: [PATCH 3/3] fix(core): end caught-up tracking on all terminal stream paths caughtUp() is documented to reject when the stream ends before reaching the tail, but cancel(), non-retryable HTTP responses, and the user-abort branches closed or errored the stream without ending the tracker, so a caller awaiting caughtUp() could stay pending forever. End the tracker on every terminal path. Also removes a stray closing tag from the changeset and applies oxfmt formatting. --- .changeset/chat-session-caught-up-resume.md | 1 - packages/core/src/v3/apiClient/index.ts | 5 ++++- .../core/src/v3/apiClient/runStream.test.ts | 22 +++++++++++++++---- packages/core/src/v3/apiClient/runStream.ts | 6 +++++ 4 files changed, 28 insertions(+), 6 deletions(-) diff --git a/.changeset/chat-session-caught-up-resume.md b/.changeset/chat-session-caught-up-resume.md index f97c3f6b911..600ae7c9aed 100644 --- a/.changeset/chat-session-caught-up-resume.md +++ b/.changeset/chat-session-caught-up-resume.md @@ -4,4 +4,3 @@ --- Chat sessions now close a resumed stream as soon as it has caught up to the latest output, instead of holding the connection open for the full long-poll window. Reloading or reconnecting to an idle chat settles faster. - diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts index 21ca40f4bbc..58e8e60b7a7 100644 --- a/packages/core/src/v3/apiClient/index.ts +++ b/packages/core/src/v3/apiClient/index.ts @@ -1432,7 +1432,10 @@ export class ApiClient { const stream = await subscription.subscribe(); if (options?.onCaughtUp) { - subscription.caughtUp().then(options.onCaughtUp).catch(() => {}); + subscription + .caughtUp() + .then(options.onCaughtUp) + .catch(() => {}); } const onPart = options?.onPart; const onControl = options?.onControl; diff --git a/packages/core/src/v3/apiClient/runStream.test.ts b/packages/core/src/v3/apiClient/runStream.test.ts index df8b6325e37..680ea927e80 100644 --- a/packages/core/src/v3/apiClient/runStream.test.ts +++ b/packages/core/src/v3/apiClient/runStream.test.ts @@ -609,7 +609,12 @@ describe("SSEStreamSubscription caught-up tracking", () => { vi.restoreAllMocks(); }); - type Rec = { body: string; seq_num: number; timestamp: number; headers?: Array<[string, string]> }; + type Rec = { + body: string; + seq_num: number; + timestamp: number; + headers?: Array<[string, string]>; + }; type Tail = { seq_num: number; timestamp: number }; function dataRec(seq: number): Rec { @@ -677,7 +682,9 @@ describe("SSEStreamSubscription caught-up tracking", () => { it("stays behind when the batch does not reach the tail", async () => { globalThis.fetch = vi .fn() - .mockResolvedValue(makeEventsResponse([batchEvent([dataRec(0)], { seq_num: 3, timestamp: 1 })])); + .mockResolvedValue( + makeEventsResponse([batchEvent([dataRec(0)], { seq_num: 3, timestamp: 1 })]) + ); const sub = new SSEStreamSubscription("http://x", { maxRetries: 0 }); const stream = await sub.subscribe(); await drain(stream); @@ -703,7 +710,12 @@ describe("SSEStreamSubscription caught-up tracking", () => { batchEvent( [ dataRec(0), - { body: "", seq_num: 1, timestamp: 1, headers: [["trigger-control", "turn-complete"]] }, + { + body: "", + seq_num: 1, + timestamp: 1, + headers: [["trigger-control", "turn-complete"]], + }, { body: "AAAAAAAAAAQ=", seq_num: 2, timestamp: 1, headers: [["", "trim"]] }, ], { seq_num: 3, timestamp: 1 } @@ -723,7 +735,9 @@ describe("SSEStreamSubscription caught-up tracking", () => { it("never reaches caught-up when the wire carries no tail (feature-detect fallback)", async () => { globalThis.fetch = vi .fn() - .mockResolvedValue(makeEventsResponse([batchEvent([dataRec(0), dataRec(1), dataRec(2)]), pingEvent()])); + .mockResolvedValue( + makeEventsResponse([batchEvent([dataRec(0), dataRec(1), dataRec(2)]), pingEvent()]) + ); const sub = new SSEStreamSubscription("http://x", { maxRetries: 0 }); const stream = await sub.subscribe(); await drain(stream); diff --git a/packages/core/src/v3/apiClient/runStream.ts b/packages/core/src/v3/apiClient/runStream.ts index c3904aa7c00..bdf7200074c 100644 --- a/packages/core/src/v3/apiClient/runStream.ts +++ b/packages/core/src/v3/apiClient/runStream.ts @@ -309,6 +309,7 @@ export class SSEStreamSubscription implements StreamSubscription { await self.connectStream(controller); }, cancel() { + self.caughtUpTracker.end(); self.options.onComplete?.(); }, }); @@ -373,6 +374,7 @@ export class SSEStreamSubscription implements StreamSubscription { ); this.options.onError?.(error); if (this.nonRetryableStatuses.has(response.status)) { + this.caughtUpTracker.end(); controller.error(error); return; } @@ -514,6 +516,7 @@ export class SSEStreamSubscription implements StreamSubscription { if (this.options.signal?.aborted) { reader.cancel(); reader.releaseLock(); + this.caughtUpTracker.end(); controller.close(); this.options.onComplete?.(); return; @@ -529,6 +532,7 @@ export class SSEStreamSubscription implements StreamSubscription { } catch (error) { if (this.options.signal?.aborted) { // User cancel — exit cleanly, don't retry. + this.caughtUpTracker.end(); controller.close(); this.options.onComplete?.(); return; @@ -555,6 +559,7 @@ export class SSEStreamSubscription implements StreamSubscription { error?: Error ): Promise { if (this.options.signal?.aborted) { + this.caughtUpTracker.end(); controller.close(); this.options.onComplete?.(); return; @@ -598,6 +603,7 @@ export class SSEStreamSubscription implements StreamSubscription { this.retryNowController = null; if (this.options.signal?.aborted) { + this.caughtUpTracker.end(); controller.close(); this.options.onComplete?.(); return;