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.

Author: ericallam

.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.
+</content>

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 ?? "";

apps/webapp/app/services/realtime/streamBasinProvisioner.server.ts

@@ -185,7 +185,7 @@ type CreateBasinOptions = {
 };
 
 async function s2CreateBasin(name: string, opts: CreateBasinOptions): Promise<void> {
-  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<void> {
-  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) },

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",

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",

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",

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<AsyncIterableStream<T>> {
     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;
 

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

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> = TRunTypes extends AnyRunTypes
   ? {
@@ -197,6 +198,7 @@ export class SSEStreamSubscription implements StreamSubscription {
   private nonRetryableStatuses: ReadonlySet<number>;
   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<CaughtUpTracker["caughtUp"]> {
+    return this.caughtUpTracker.caughtUp();
+  }
+
   async subscribe(): Promise<ReadableStream<SSEStreamPart>> {
     // 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);
   }
 }

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