fix(eve): client streams - reconnect stalled open responses

.changeset/calm-streams-reconnect.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Client session streams now cancel and reconnect open connections that stop delivering bytes, resuming from the last durable cursor so buffered terminal events can still reach callers.

docs/guides/client/streaming.mdx

@@ -103,7 +103,20 @@ If you support refresh while an authorization prompt is pending, keep the sessio
 
 ## Reconnection
 
-HTTP connections can end before a run does. The client reconnects from the number of events already consumed, so long turns continue without replaying events. It stops at a turn boundary, when aborted, or when the stream can no longer make progress.
+HTTP connections can end—or remain open without delivering more bytes—before a run does. The client reconnects from the number of events already consumed, so long turns continue without replaying events. By default, an open stream that produces no bytes for 30 seconds is cancelled and reopened from the durable cursor. It stops at a turn boundary, when aborted, or when the stream can no longer make progress.
+
+Tune the idle-read deadline with `streamIdleTimeoutMs`:
+
+```ts
+const response = await session.send({
+  message: "Run the long operation.",
+  streamReconnectPolicy: { streamIdleTimeoutMs: 15_000 },
+});
+```
+
+A step can legitimately run for minutes without emitting an event, so reopening a stream that went quiet does not count against `streamIdleReconnectPolicy`. That budget stops the follow only when reconnects keep _ending_ without delivering events, which means the session has nothing left to send.
+
+Set `streamIdleTimeoutMs: 0` to disable idle-read reconnects. Tail-relative streams opened with a negative `startIndex` remain single-connection streams because their absolute cursor cannot be advanced safely.
 
 Set `streamReconnectPolicy: { reconnect: false }` when a relay or proxy owns the cursor and reconnection policy. This makes a single stream GET attempt and returns when that connection ends; it does not stop the server-side turn:
 

packages/eve/src/client/ndjson.ts

@@ -5,6 +5,10 @@ import type { HandleMessageStreamEvent } from "#protocol/message.js";
  * can be recovered via reconnection.
  */
 export function isStreamDisconnectError(error: unknown): boolean {
+  if (error instanceof StreamIdleTimeoutError) {
+    return true;
+  }
+
   if (error instanceof DOMException) {
     return error.name === "AbortError";
   }
@@ -24,6 +28,34 @@ export function isStreamDisconnectError(error: unknown): boolean {
   );
 }
 
+class StreamIdleTimeoutError extends Error {
+  readonly timeoutMs: number;
+
+  constructor(timeoutMs: number) {
+    super(`Message stream produced no bytes for ${timeoutMs}ms.`);
+    this.name = "StreamIdleTimeoutError";
+    this.timeoutMs = timeoutMs;
+  }
+}
+
+/**
+ * Returns true when a stream ended because it stopped delivering bytes rather
+ * than because the transport ended.
+ *
+ * An open stream that goes quiet says nothing about whether the run is still
+ * progressing, so callers must not treat it as evidence of an exhausted
+ * session the way they treat a connection that ends without events.
+ */
+export function isStreamIdleTimeoutError(error: unknown): boolean {
+  return error instanceof StreamIdleTimeoutError;
+}
+
+interface ReadNdjsonStreamOptions {
+  readonly idleTimeoutMs?: number;
+}
+
+type StreamReadResult = Awaited<ReturnType<ReadableStreamDefaultReader<Uint8Array>["read"]>>;
+
 /**
  * Reads newline-delimited JSON events from a `ReadableStream<Uint8Array>`.
  *
@@ -35,6 +67,7 @@ export function isStreamDisconnectError(error: unknown): boolean {
  */
 export async function* readNdjsonStream(
   body: ReadableStream<Uint8Array>,
+  options: ReadNdjsonStreamOptions = {},
 ): AsyncGenerator<HandleMessageStreamEvent> {
   const reader = body.getReader();
   const decoder = new TextDecoder();
@@ -43,7 +76,7 @@ export async function* readNdjsonStream(
 
   try {
     while (true) {
-      const result = await reader.read();
+      const result = await readWithIdleTimeout(reader, options.idleTimeoutMs);
 
       if (result.done) {
         reachedEof = true;
@@ -84,3 +117,25 @@ export async function* readNdjsonStream(
     reader.releaseLock();
   }
 }
+
+async function readWithIdleTimeout(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  idleTimeoutMs: number | undefined,
+): Promise<StreamReadResult> {
+  if (idleTimeoutMs === undefined) {
+    return await reader.read();
+  }
+
+  let timeout: ReturnType<typeof setTimeout> | undefined;
+  const timeoutPromise = new Promise<never>((_resolve, reject) => {
+    timeout = setTimeout(() => reject(new StreamIdleTimeoutError(idleTimeoutMs)), idleTimeoutMs);
+  });
+
+  try {
+    return await Promise.race([reader.read(), timeoutPromise]);
+  } finally {
+    if (timeout !== undefined) {
+      clearTimeout(timeout);
+    }
+  }
+}

packages/eve/src/client/open-stream.ts

@@ -1,7 +1,11 @@
 import type { HandleMessageStreamEvent } from "#protocol/message.js";
 import { createEveMessageStreamRoutePath } from "#protocol/routes.js";
 import { ClientError } from "#client/client-error.js";
-import { isStreamDisconnectError, readNdjsonStream } from "#client/ndjson.js";
+import {
+  isStreamDisconnectError,
+  isStreamIdleTimeoutError,
+  readNdjsonStream,
+} from "#client/ndjson.js";
 import type {
   ClientRedirectPolicy,
   ResolvedStreamReconnectPolicy as StreamReconnectPolicyOptions,
@@ -19,12 +23,14 @@ interface RetryPolicy {
 interface ResolvedStreamReconnectPolicy {
   readonly retryableErrorStatuses: ReadonlySet<number>;
   readonly streamIdleReconnectPolicy: RetryPolicy;
+  readonly streamIdleTimeoutMs: number | undefined;
   readonly streamOpenReconnectPolicy: RetryPolicy;
 }
 
 const DEFAULT_STREAM_RECONNECT_POLICY: ResolvedStreamReconnectPolicy = {
   retryableErrorStatuses: new Set([404, 409, 425, 500, 502, 503, 504]),
   streamIdleReconnectPolicy: { baseDelayMs: 250, maxAttempts: 5, maxDelayMs: 4_000 },
+  streamIdleTimeoutMs: 30_000,
   streamOpenReconnectPolicy: { baseDelayMs: 250, maxAttempts: 12, maxDelayMs: 5_000 },
 };
 
@@ -34,6 +40,7 @@ const NO_STREAM_RECONNECT_POLICY: ResolvedStreamReconnectPolicy = {
     ...DEFAULT_STREAM_RECONNECT_POLICY.streamIdleReconnectPolicy,
     maxAttempts: 0,
   },
+  streamIdleTimeoutMs: undefined,
   streamOpenReconnectPolicy: {
     ...DEFAULT_STREAM_RECONNECT_POLICY.streamOpenReconnectPolicy,
     maxAttempts: 1,
@@ -63,13 +70,26 @@ function resolveStreamReconnectPolicy(
       configured?.streamIdleReconnectPolicy,
       DEFAULT_STREAM_RECONNECT_POLICY.streamIdleReconnectPolicy,
     ),
+    streamIdleTimeoutMs: resolveStreamIdleTimeoutMs(configured?.streamIdleTimeoutMs),
     streamOpenReconnectPolicy: resolveRetryPolicy(
       configured?.streamOpenReconnectPolicy,
       DEFAULT_STREAM_RECONNECT_POLICY.streamOpenReconnectPolicy,
     ),
   };
 }
 
+function resolveStreamIdleTimeoutMs(value: number | undefined): number | undefined {
+  if (value === undefined) {
+    return DEFAULT_STREAM_RECONNECT_POLICY.streamIdleTimeoutMs;
+  }
+
+  if (!Number.isFinite(value) || value < 0) {
+    throw new Error("streamIdleTimeoutMs must be a non-negative finite number.");
+  }
+
+  return value === 0 ? undefined : value;
+}
+
 /**
  * Internal configuration for following a durable event stream.
  */
@@ -114,8 +134,11 @@ export async function* followStreamIterable(
     }
 
     let deliveredEvent = false;
+    let timedOutIdle = false;
     try {
-      for await (const event of readNdjsonStream(body)) {
+      for await (const event of readNdjsonStream(body, {
+        idleTimeoutMs: input.startIndex < 0 ? undefined : retryPolicy.streamIdleTimeoutMs,
+      })) {
         startIndex += 1;
         deliveredEvent = true;
         reconnectDelayMs = idleRetryPolicy.baseDelayMs;
@@ -126,13 +149,20 @@ export async function* followStreamIterable(
       if (!isStreamDisconnectError(error)) {
         throw error;
       }
+      timedOutIdle = isStreamIdleTimeoutError(error);
     }
 
     if (input.signal?.aborted || input.startIndex < 0 || idleRetryPolicy.maxAttempts === 0) {
       return;
     }
 
-    if (
+    if (timedOutIdle) {
+      // A stream the server is still holding open is not an exhausted session,
+      // and eve emits no bytes at all across a slow step. Charging these to the
+      // empty-stream budget would end the follow mid-turn. The idle deadline
+      // itself already paces the reconnect.
+      reconnectDelayMs = idleRetryPolicy.baseDelayMs;
+    } else if (
       !deliveredEvent &&
       !initialConnection &&
       (idleReconnects += 1) >= idleRetryPolicy.maxAttempts

packages/eve/src/client/session.test.ts

@@ -508,6 +508,46 @@ describe("ClientSession", () => {
     expect(session.state.streamIndex).toBe(1);
   });
 
+  it("does not arm an idle timeout when stream reconnection is disabled", async () => {
+    const abortController = new AbortController();
+    const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(async (_request, init) => {
+      return new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            init?.signal?.addEventListener("abort", () => {
+              controller.error(new DOMException("The operation was aborted.", "AbortError"));
+            });
+          },
+        }),
+      );
+    });
+    const session = createSession({ sessionId: "session_1", streamIndex: 0 });
+
+    vi.useFakeTimers();
+    try {
+      let settled = false;
+      const consumed = (async () => {
+        for await (const _event of session.stream({
+          signal: abortController.signal,
+          streamReconnectPolicy: { reconnect: false },
+        })) {
+          // The stream intentionally remains open and silent.
+        }
+      })().finally(() => {
+        settled = true;
+      });
+
+      await vi.advanceTimersByTimeAsync(31_000);
+      expect(settled).toBe(false);
+      expect(fetchMock).toHaveBeenCalledOnce();
+
+      abortController.abort();
+      await consumed;
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("does not reconnect a sent turn's response stream when disabled", async () => {
     const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(async (_request, init) => {
       if ((init?.method ?? "GET") === "POST") {
@@ -572,6 +612,23 @@ describe("ClientSession", () => {
     expect(fetchMock).toHaveBeenCalledTimes(2);
   });
 
+  it.each([-1, Number.NaN, Number.POSITIVE_INFINITY])(
+    "rejects an invalid stream idle timeout (%s)",
+    async (streamIdleTimeoutMs) => {
+      const fetchMock = vi.spyOn(globalThis, "fetch");
+      const session = createSession({ sessionId: "session_1", streamIndex: 0 });
+
+      await expect(async () => {
+        for await (const _event of session.stream({
+          streamReconnectPolicy: { streamIdleTimeoutMs },
+        })) {
+          // Policy validation happens before opening the stream.
+        }
+      }).rejects.toThrow("streamIdleTimeoutMs must be a non-negative finite number.");
+      expect(fetchMock).not.toHaveBeenCalled();
+    },
+  );
+
   it("retries a transient fetch failure while reopening an active turn stream", async () => {
     const encoder = new TextEncoder();
     const streamUrls: string[] = [];