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.

Author: ericallam

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

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;

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);

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<void> {
     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;