diff --git a/.changeset/chat-session-caught-up-resume.md b/.changeset/chat-session-caught-up-resume.md new file mode 100644 index 00000000000..600ae7c9aed --- /dev/null +++ b/.changeset/chat-session-caught-up-resume.md @@ -0,0 +1,6 @@ +--- +"@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/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"], 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..58e8e60b7a7 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,12 @@ 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..680ea927e80 100644 --- a/packages/core/src/v3/apiClient/runStream.test.ts +++ b/packages/core/src/v3/apiClient/runStream.test.ts @@ -600,3 +600,147 @@ 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..bdf7200074c 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; @@ -287,6 +309,7 @@ export class SSEStreamSubscription implements StreamSubscription { await self.connectStream(controller); }, cancel() { + self.caughtUpTracker.end(); self.options.onComplete?.(); }, }); @@ -351,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; } @@ -407,9 +431,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 +477,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 +507,7 @@ export class SSEStreamSubscription implements StreamSubscription { if (done) { reader.releaseLock(); + this.caughtUpTracker.end(); controller.close(); this.options.onComplete?.(); return; @@ -466,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; @@ -481,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; @@ -490,6 +542,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; } @@ -506,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; @@ -513,6 +567,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; @@ -548,12 +603,14 @@ export class SSEStreamSubscription implements StreamSubscription { this.retryNowController = null; if (this.options.signal?.aborted) { + this.caughtUpTracker.end(); controller.close(); this.options.onComplete?.(); return; } // 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)