fix(core): stop reconnecting a session stream after the consumer cancels

Cancelling the returned stream aborted the internal connection, but the retry path only bailed on the caller-provided abort signal, so a consumer cancel was treated as a transient drop and kept reconnecting (up to maxRetries), issuing SSE requests after the consumer was gone. Track consumer cancellation separately, wake any pending backoff, and short-circuit the retry. Adds a regression test.

Also from review: drop the unused s2 networkEndpoint field, order the webapp test-server env so the worker-disable vars win over extraEnv as documented, abort the cross-origin browser read on a timer instead of a between-reads deadline, and correct the client-protocol docs (the caught-up check counts the highest raw seq_num including skipped command records; the chat transport closes on caught-up, SSEStreamSubscription only exposes the signal).

Author: ericallam

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

@@ -91,6 +91,7 @@ describe("session stream browser e2e", () => {
       const result = await page.evaluate(
         async ({ url, token }) => {
           const ac = new AbortController();
+          const deadlineTimer = setTimeout(() => ac.abort(), 8000);
           try {
             const res = await fetch(url, {
               headers: { Authorization: `Bearer ${token}`, Accept: "text/event-stream" },
@@ -99,8 +100,7 @@ describe("session stream browser e2e", () => {
             const reader = (res.body as ReadableStream<Uint8Array>).getReader();
             const decoder = new TextDecoder();
             let text = "";
-            const deadline = Date.now() + 8000;
-            while (Date.now() < deadline) {
+            while (true) {
               const { done, value } = await reader.read();
               if (done) break;
               text += decoder.decode(value, { stream: true });
@@ -115,6 +115,8 @@ describe("session stream browser e2e", () => {
             };
           } catch (e) {
             return { error: String(e) };
+          } finally {
+            clearTimeout(deadlineTimer);
           }
         },
         { url: sseUrl, token }

docs/ai-chat/client-protocol.mdx

@@ -359,7 +359,7 @@ Decoded `data` payload:
 | `records[].timestamp` | Unix ms when the record was written to S2. |
 | `records[].body` | For data records: a JSON-encoded **string** wrapping `{ data: UIMessageChunk, id: string }`. For control records: an empty string (semantics live in `headers`). For S2 command records: opaque bytes. See [Records on session.out](#records-on-session-out). |
 | `records[].headers` | Optional `[name, value]` pairs. Empty for data records; a `trigger-control` entry for control records; a single empty-name `["", "<op>"]` entry for S2 command records. |
-| `tail.seq_num` | Latest known tail of the S2 stream, useful for detecting how far behind the live edge you are. When `last delivered seq_num + 1 === tail.seq_num` you have drained the backlog and are caught up to the live edge. The same `tail` also rides on `ping` events. Skip if you don't need it. |
+| `tail.seq_num` | Latest known tail of the S2 stream, useful for detecting how far behind the live edge you are. Track the **highest `seq_num` you have received**, counting every record in the batch (including the command records you skip, see below), not just the last application-visible one. When `highest received seq_num + 1 === tail.seq_num` you have drained the backlog and are caught up to the live edge. The same `tail` also rides on `ping` events. Skip if you don't need it. |
 | `tail.timestamp` | Timestamp of `tail.seq_num`. |
 
 ### Records on `session.out`
@@ -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 the TypeScript `SSEStreamSubscription` (or `useChat`, which builds on it), the client settles on its own too: on the reconnect path it watches the `tail` on batch and ping events and closes a resumed stream as soon as it reaches the live edge, so a settled idle reconnect closes promptly without waiting out the long poll. Hand-rolled clients can do the same by comparing their last processed `seq_num` to the ping/batch `tail`. 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), 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.
 
 ```ts
 // Reconnect path (page reload)

internal-packages/testcontainers/src/s2.ts

@@ -28,8 +28,6 @@ export interface StartedS2Container {
   container: StartedTestContainer;
   /** Base URL of the s2-lite HTTP API, e.g. `http://localhost:49xxx`. */
   endpoint: string;
-  /** In-network URL (for the spawned webapp on the same docker network). */
-  networkEndpoint: string;
   /** The single basin s2-lite is initialised with. */
   basin: string;
 }
@@ -64,7 +62,6 @@ export async function createS2Container(
   return {
     container,
     endpoint: `http://${container.getHost()}:${mappedPort}`,
-    networkEndpoint: "http://s2:80",
     basin,
   };
 }

internal-packages/testcontainers/src/webapp.ts

@@ -124,6 +124,7 @@ export async function startWebapp(
       REDIS_HOST: redis.host,
       REDIS_PORT: String(redis.port),
       REDIS_TLS_DISABLED: "true", // all *_REDIS_TLS_DISABLED vars default to this; test Redis has no TLS
+      ...(options.extraEnv ?? {}),
       // Disable all background workers. Each worker has its own env var and its own
       // check idiom ("0" vs "false" vs boolean), so we set all of them explicitly.
       WORKER_ENABLED: "false", // disables workerQueue.initialize() (checked === "true")
@@ -143,7 +144,6 @@ export async function startWebapp(
       // to "0" so a local apps/webapp/.env that sets it to "1" doesn't
       // short-circuit the loader past the REQUIRE_PLUGINS check.
       ...(requirePlugins ? { REQUIRE_PLUGINS: "1", RBAC_FORCE_FALLBACK: "0" } : {}),
-      ...(options.extraEnv ?? {}),
       NODE_PATH: nodePath,
     },
     stdio: ["ignore", "pipe", "pipe"],

packages/core/src/v3/apiClient/runStream.test.ts

@@ -27,6 +27,22 @@ describe("SSEStreamSubscription retry behavior", () => {
     });
   }
 
+  /**
+   * A response.body that emits one SSE event then stays open (never closes),
+   * so the connection sits mid-read until the fetch signal aborts.
+   */
+  function makeOpenSSEResponse() {
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(`id: 1\ndata: {"hello":1}\n\n`));
+      },
+    });
+    return new Response(body, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream", "X-Stream-Version": "v1" },
+    });
+  }
+
   // Drain a ReadableStream<SSEStreamPart> until it closes or errors.
   // Returns received chunks plus terminal state.
   async function drain(stream: ReadableStream<{ id: string; chunk: unknown }>) {
@@ -74,6 +90,30 @@ describe("SSEStreamSubscription retry behavior", () => {
     expect(result.chunks).toHaveLength(1);
   });
 
+  it("does not reconnect after the consumer cancels the returned stream", async () => {
+    let attempts = 0;
+    globalThis.fetch = vi.fn().mockImplementation(async () => {
+      attempts++;
+      return makeOpenSSEResponse();
+    });
+
+    const sub = new SSEStreamSubscription("http://example.test/sse", {
+      retryDelayMs: 1,
+      maxRetryDelayMs: 5,
+      maxRetries: 10,
+    });
+
+    const stream = await sub.subscribe();
+    const reader = stream.getReader();
+    const first = await reader.read();
+    expect(first.done).toBe(false);
+
+    await reader.cancel();
+    await new Promise((r) => setTimeout(r, 50));
+
+    expect(attempts).toBe(1);
+  });
+
   it("caps the exponential backoff at maxRetryDelayMs", async () => {
     let attempts = 0;
     const callTimes: number[] = [];

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

@@ -209,6 +209,7 @@ export class SSEStreamSubscription implements StreamSubscription {
   private nonRetryableStatuses: ReadonlySet<number>;
   private retryNowController: AbortController | null = null;
   private internalAbort: AbortController | null = null;
+  private cancelledByConsumer = false;
   private caughtUpTracker = new CaughtUpTracker();
 
   constructor(
@@ -329,7 +330,9 @@ export class SSEStreamSubscription implements StreamSubscription {
         await self.connectStream(controller);
       },
       cancel() {
+        self.cancelledByConsumer = true;
         self.internalAbort?.abort();
+        self.retryNowController?.abort();
       },
     });
     const internalReader = internal.getReader();
@@ -362,6 +365,7 @@ export class SSEStreamSubscription implements StreamSubscription {
           }
         },
         cancel(reason) {
+          self.cancelledByConsumer = true;
           self.caughtUpTracker.end();
           self.options.onComplete?.();
           internalReader.cancel(reason).catch(() => {});
@@ -591,7 +595,7 @@ export class SSEStreamSubscription implements StreamSubscription {
         throw error;
       }
     } catch (error) {
-      if (this.options.signal?.aborted) {
+      if (this.options.signal?.aborted || this.cancelledByConsumer) {
         // User cancel — exit cleanly, don't retry.
         controller.close();
         return;
@@ -616,7 +620,7 @@ export class SSEStreamSubscription implements StreamSubscription {
     controller: ReadableStreamDefaultController,
     error?: Error
   ): Promise<void> {
-    if (this.options.signal?.aborted) {
+    if (this.options.signal?.aborted || this.cancelledByConsumer) {
       controller.close();
       return;
     }
@@ -657,7 +661,7 @@ export class SSEStreamSubscription implements StreamSubscription {
     });
     this.retryNowController = null;
 
-    if (this.options.signal?.aborted) {
+    if (this.options.signal?.aborted || this.cancelledByConsumer) {
       controller.close();
       return;
     }