triggerdotdev · ericallam · Jul 23, 2026 · Jul 23, 2026 · Jul 23, 2026
diff --git a/.changeset/chat-session-caught-up-resume.md 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
@@ -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
@@ -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) },

diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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,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
@@ -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<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);
+  });
+});