refactor(core,sdk): drop the client-side caught-up close, rely on the server fast-close

Author: ericallam

.changeset/chat-session-caught-up-resume.md

@@ -3,4 +3,4 @@
 "@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. This applies to the server-to-server `AgentChat` client's `reconnect()` too, not just the browser transport.
+Reconnecting to an idle chat now closes promptly instead of holding the connection open for the full long-poll window. When a resumed `.out` stream is already settled (the agent finished its turn), the server drains any residual records and closes right away, so a reload or reconnect settles in about a second instead of tens of seconds. `AgentChat.reconnect()` opts into the same fast-close.

apps/webapp/test/helpers/sessionStream.ts

@@ -279,53 +279,3 @@ export async function collectSessionOut(
   return { parts, durationMs: performance.now() - started, subscription };
 }
 
-/**
- * Subscribe + drain while awaiting the client's `caughtUp()`. Resolves once the
- * client reports it has drained to the live tail (or `maxMs` elapses). Used by
- * the GREEN legs: the drain is what lets the wrapper mark the tail boundary, so
- * `parts` holds everything delivered up to the moment caught-up fires — the
- * data-loss guard.
- */
-export async function collectUntilCaughtUp(opts: SubscribeOptions & { maxMs?: number }): Promise<{
-  parts: CollectedPart[];
-  caughtUp: boolean;
-  tailSeqNum?: number;
-  settleMs: number;
-}> {
-  const subscription = subscribeSessionOut(opts);
-  const stream = await subscription.subscribe();
-  const reader = stream.getReader();
-  const parts: CollectedPart[] = [];
-  const started = performance.now();
-
-  let caughtUp = false;
-  let tailSeqNum: number | undefined;
-  let settleMs = 0;
-  subscription
-    .caughtUp()
-    .then((tail) => {
-      caughtUp = true;
-      tailSeqNum = tail.seqNum;
-      settleMs = performance.now() - started;
-    })
-    .catch(() => {});
-
-  const deadline = started + (opts.maxMs ?? 30_000);
-  try {
-    while (!caughtUp) {
-      const remaining = deadline - performance.now();
-      if (remaining <= 0) break;
-      const next = await Promise.race([
-        reader.read(),
-        new Promise<"tick">((r) => setTimeout(() => r("tick"), Math.min(remaining, 250))),
-      ]);
-      if (next === "tick") continue;
-      if (next.done) break;
-      parts.push(next.value as CollectedPart);
-    }
-  } finally {
-    await reader.cancel().catch(() => {});
-  }
-
-  return { parts, caughtUp, tailSeqNum, settleMs: settleMs || performance.now() - started };
-}

apps/webapp/test/session-stream.e2e.test.ts

@@ -18,7 +18,6 @@ import { seedTestEnvironment } from "./helpers/seedTestEnvironment";
 import {
   appendInput,
   collectSessionOut,
-  collectUntilCaughtUp,
   isTurnComplete,
   mintSessionToken,
   openChannelRaw,
@@ -163,72 +162,27 @@ describe("session stream e2e", () => {
     expect(allData).toEqual([0, 1, 2, 3, 4]);
   });
 
-  it("E7 trim-at-tail: trim command record is filtered; caught-up still fires", async () => {
+  it("E7 trim-at-tail: trim command record is filtered from the stream", async () => {
     const { addressingKey, token, producer, baseUrl } = await setupSession();
 
     await producer.appendData({ n: 0 }, "p0");
     const keep = await producer.appendData({ n: 1 }, "p1");
     await producer.appendTurnComplete();
     await producer.trim(keep);
 
-    const { parts, caughtUp } = await collectUntilCaughtUp({
+    const { parts } = await collectSessionOut({
       baseUrl,
       addressingKey,
       token,
+      until: (p) => p.some(isTurnComplete),
       maxMs: 15_000,
     });
 
-    expect(caughtUp).toBe(true);
     const commandRecords = parts.filter((p) => (p.headers ?? []).some(([k]) => k === ""));
     expect(commandRecords).toHaveLength(0);
     expect(parts.filter((p) => p.chunk != null).length).toBeGreaterThanOrEqual(1);
   });
 
-  it("E3 quiescent reconnect (GREEN): client caught-up close settles at the tail", async () => {
-    const { addressingKey, token, producer, baseUrl } = await setupSession();
-
-    await producer.appendData({ n: 0 }, "p0");
-    await producer.appendData({ n: 1 }, "p1");
-    const tc = await producer.appendTurnComplete();
-
-    const { parts, caughtUp, settleMs } = await collectUntilCaughtUp({
-      baseUrl,
-      addressingKey,
-      token,
-      lastEventId: String(tc),
-      timeoutInSeconds: 30,
-      maxMs: 15_000,
-    });
-
-    expect(caughtUp).toBe(true);
-    expect(settleMs).toBeLessThan(5_000);
-    expect(parts.filter((p) => p.chunk != null)).toHaveLength(0);
-  });
-
-  it("E4 backlog reaches tail (GREEN): every record is delivered before caught-up", async () => {
-    const { addressingKey, token, producer, baseUrl } = await setupSession();
-
-    await producer.appendData({ n: 0 }, "p0");
-    await producer.appendData({ n: 1 }, "p1");
-    await producer.appendData({ n: 2 }, "p2");
-    const tc = await producer.appendTurnComplete();
-
-    const { parts, caughtUp, tailSeqNum } = await collectUntilCaughtUp({
-      baseUrl,
-      addressingKey,
-      token,
-      maxMs: 15_000,
-    });
-
-    expect(caughtUp).toBe(true);
-    expect(tailSeqNum).toBe(tc + 1);
-    const dataChunks = parts
-      .filter((p) => !isTurnComplete(p) && p.chunk != null)
-      .map((p) => (p.chunk as { n: number }).n);
-    expect(dataChunks).toEqual([0, 1, 2]);
-    expect(parts.some(isTurnComplete)).toBe(true);
-  });
-
   it("E8 in/append 413 for an oversized body still carries CORS headers", async () => {
     const { addressingKey, token, baseUrl } = await setupSession();
 

docs/ai-chat/client-protocol.mdx

@@ -652,7 +652,7 @@ On **reconnect-on-reload** paths (resuming a chat where nothing may be streaming
 
 **Do not send `X-Peek-Settled` on the active-send response-stream path.** The peek would race the newly-triggered turn's first chunk — if the agent hasn't written the new turn's first record yet, the peek sees the prior turn's `turn-complete` and closes the SSE before the response lands on S2. The built-in `TriggerChatTransport.reconnectToStream` sets the header; `sendMessages → subscribeToStream` does not.
 
-If you use `TriggerChatTransport` (or `useChat`, which builds on it), the transport settles the stream for you: on the reconnect path it watches the `tail` on batch and ping events via `SSEStreamSubscription.caughtUp()` and closes the resumed stream as soon as it reaches the live edge, so a settled idle reconnect closes promptly without waiting out the long poll. `SSEStreamSubscription` on its own only exposes the caught-up signal, it does not close itself: consuming it directly, call `caughtUp()` (or compare your last received `seq_num` to the ping/batch `tail`) and close your own stream. On older self-hosted backends whose `ping` carries no `tail`, this falls back to the `X-Peek-Settled` behavior above.
+If you use `TriggerChatTransport` (or `useChat`, which builds on it), this is handled for you: the reconnect path sends `X-Peek-Settled`, so a settled idle reconnect closes promptly on the server's `wait=0` fast-close instead of waiting out the long poll. Consuming `.out` directly, send the same header on your reconnect reads and treat an `X-Session-Settled: true` response as a terminal close.
 
 ```ts
 // Reconnect path (page reload)

packages/core/src/v3/apiClient/index.ts

@@ -1453,13 +1453,6 @@ 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}`;
@@ -1474,12 +1467,6 @@ 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

@@ -640,206 +640,3 @@ 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);
-  });
-
-  it("does not resolve caughtUp() until the consumer drains the 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();
-    let resolved = false;
-    cu.then(() => {
-      resolved = true;
-    }).catch(() => {});
-
-    await new Promise((r) => setTimeout(r, 20));
-    expect(resolved).toBe(false);
-    expect(sub.isCaughtUp()).toBe(false);
-
-    const parts = await drain(stream);
-    const tail = await cu;
-    expect(tail.seqNum).toBe(3);
-    expect(parts).toHaveLength(3);
-    expect(sub.isCaughtUp()).toBe(true);
-  });
-
-  it("holds caughtUp() until the final record before the tail is consumed", 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();
-    let resolved = false;
-    cu.then(() => {
-      resolved = true;
-    }).catch(() => {});
-    const reader = stream.getReader();
-
-    const first = await reader.read();
-    expect(first.value?.id).toBe("0");
-    await reader.read();
-    await reader.read();
-    await new Promise((r) => setTimeout(r, 20));
-    expect(resolved).toBe(false);
-
-    const done = await reader.read();
-    expect(done.done).toBe(true);
-    await new Promise((r) => setTimeout(r, 20));
-    expect(resolved).toBe(true);
-    expect(sub.isCaughtUp()).toBe(true);
-    reader.releaseLock();
-  });
-});