fix(eve): reconnect silent client streams

.changeset/calm-streams-reconnect.md

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

docs/guides/client/streaming.mdx

@@ -126,7 +126,17 @@ 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 canceled 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("Run the long operation.", {
+  streamReconnectPolicy: { streamIdleTimeoutMs: 15_000 },
+});
+```
+
+A step can legitimately run for minutes without emitting an event, so reopening a quiet stream does not count against `streamIdleReconnectPolicy`. 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.
 
 If your consumer persists events, key on `event.meta.id`. It is stable across reconnects and rewinds, so an overlapping replay is safe to ingest twice. See [the event envelope](../../concepts/sessions-runs-and-streaming#the-event-envelope).
 

docs/subagents.mdx

@@ -192,7 +192,7 @@ Without the opt-in, delegated children run as one-shot tasks: they answer once a
 
 With it enabled, a child parks after answering instead of terminating, keeping its session and conversation history alive. A failed child turn can also leave the child parked — its latest status shows the error and the parent may message it again. Pass a parked child's `agentId` to the same subagent tool with a new `message` to continue that session. Omitting `agentId` (or passing an empty string or `null`) always starts a new child, and an `agentId` that matches no known agent falls back to starting a new child rather than failing. Passing a known `agentId` through a different subagent tool fails with `AGENT_MISMATCH`, and messaging a child that is still starting or working on its previous request fails with `AGENT_BUSY` — wait for its result before continuing it.
 
-Whenever the set of parked (resumable) children changes, eve appends a framework-injected note to the conversation — labeled `[Agents]` and carrying an `<agents>` block — listing each child's `agentId`, name, and latest status. The static system prompt tells the model the note is injected by eve, not written by the user. The note is appended only when the listing changes (an append-only design that preserves the provider prompt cache), the most recent note is authoritative, and children that are starting or running do not appear until they park again.
+On every model call, eve injects a system message containing an `<agents>` block listing the currently parked (resumable) children — each entry carries the `agentId`, name, and latest status. It is a per-call injection, not part of the conversation history: it always reflects the current state, never accumulates stale copies, and children that are starting or running do not appear until they park again.
 
 The parent holds agent handles only for its session lifetime. When the parent session ends, eve terminates its local children. Remote children are not terminated on parent shutdown — that is a known gap: a parked remote child stays alive on its own deployment until that deployment's own lifecycle ends it. For [remote agents](./guides/remote-agents), both deployments must run the same eve version to use agent messaging.
 

e2e/fixtures/agent-subagents/package.json

@@ -22,8 +22,5 @@
   "devDependencies": {
     "@types/node": "catalog:",
     "typescript": "catalog:"
-  },
-  "e2e": {
-    "modelMatrix": "full"
   }
 }

packages/eve/src/client/ndjson.ts

@@ -24,6 +24,21 @@ export function isStreamDisconnectError(error: unknown): boolean {
   );
 }
 
+class StreamIdleTimeoutError extends Error {
+  constructor(timeoutMs: number) {
+    super(`Message stream produced no bytes for ${timeoutMs}ms.`);
+    this.name = "StreamIdleTimeoutError";
+  }
+}
+
+export function isStreamIdleTimeoutError(error: unknown): boolean {
+  return error instanceof StreamIdleTimeoutError;
+}
+
+interface ReadNdjsonStreamOptions {
+  readonly idleTimeoutMs?: number;
+}
+
 /**
  * Reads newline-delimited JSON events from a `ReadableStream<Uint8Array>`.
  *
@@ -35,6 +50,7 @@ export function isStreamDisconnectError(error: unknown): boolean {
  */
 export async function* readNdjsonStream(
   body: ReadableStream<Uint8Array>,
+  options: ReadNdjsonStreamOptions = {},
 ): AsyncGenerator<MessageStreamEvent> {
   const reader = body.getReader();
   const decoder = new TextDecoder();
@@ -43,7 +59,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 +100,27 @@ export async function* readNdjsonStream(
     reader.releaseLock();
   }
 }
+
+async function readWithIdleTimeout(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+  idleTimeoutMs: number | undefined,
+): ReturnType<ReadableStreamDefaultReader<Uint8Array>["read"]> {
+  if (idleTimeoutMs === undefined) {
+    return await reader.read();
+  }
+
+  let timeout: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      reader.read(),
+      new Promise<never>((_resolve, reject) => {
+        timeout = setTimeout(
+          () => reject(new StreamIdleTimeoutError(idleTimeoutMs)),
+          idleTimeoutMs,
+        );
+      }),
+    ]);
+  } finally {
+    if (timeout !== undefined) clearTimeout(timeout);
+  }
+}

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

@@ -2,7 +2,11 @@ import type { MessageStreamEvent } from "#protocol/message.js";
 import { EVE_STREAM_TAIL_INDEX_HEADER } from "#protocol/message.js";
 import { createEveSessionStreamRoutePath } 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,
@@ -20,12 +24,16 @@ interface RetryPolicy {
 interface ResolvedStreamReconnectPolicy {
   readonly retryableErrorStatuses: ReadonlySet<number>;
   readonly streamIdleReconnectPolicy: RetryPolicy;
+  readonly streamIdleTimeoutMs: number | undefined;
   readonly streamOpenReconnectPolicy: RetryPolicy;
 }
 
+const MAX_TIMER_DELAY_MS = 2_147_483_647;
+
 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 },
 };
 
@@ -35,6 +43,7 @@ const NO_STREAM_RECONNECT_POLICY: ResolvedStreamReconnectPolicy = {
     ...DEFAULT_STREAM_RECONNECT_POLICY.streamIdleReconnectPolicy,
     maxAttempts: 0,
   },
+  streamIdleTimeoutMs: undefined,
   streamOpenReconnectPolicy: {
     ...DEFAULT_STREAM_RECONNECT_POLICY.streamOpenReconnectPolicy,
     maxAttempts: 1,
@@ -64,13 +73,27 @@ function resolveStreamReconnectPolicy(
       configured?.streamIdleReconnectPolicy,
       DEFAULT_STREAM_RECONNECT_POLICY.streamIdleReconnectPolicy,
     ),
+    streamIdleTimeoutMs: resolveStreamIdleTimeoutMs(configured?.streamIdleTimeoutMs),
     streamOpenReconnectPolicy: resolveRetryPolicy(
       configured?.streamOpenReconnectPolicy,
       DEFAULT_STREAM_RECONNECT_POLICY.streamOpenReconnectPolicy,
     ),
   };
 }
 
+/** @internal Validates reconnect options before a turn is submitted. */
+export function validateStreamReconnectPolicy(policy: StreamReconnectPolicy | undefined): void {
+  resolveStreamReconnectPolicy(policy);
+}
+
+function resolveStreamIdleTimeoutMs(value: number | undefined): number | undefined {
+  if (value === undefined) return DEFAULT_STREAM_RECONNECT_POLICY.streamIdleTimeoutMs;
+  if (!Number.isInteger(value) || value < 0 || value > MAX_TIMER_DELAY_MS) {
+    throw new Error(`streamIdleTimeoutMs must be an integer between 0 and ${MAX_TIMER_DELAY_MS}.`);
+  }
+  return value === 0 ? undefined : value;
+}
+
 /**
  * Internal configuration for following a durable event stream.
  */
@@ -154,8 +177,14 @@ export async function* followStreamIterable(
     }
 
     let deliveredEvent = false;
+    let timedOutIdle = false;
     try {
-      for await (const event of readNdjsonStream(connection.body)) {
+      for await (const event of readNdjsonStream(connection.body, {
+        idleTimeoutMs:
+          input.startIndex < 0 || idleRetryPolicy.maxAttempts === 0
+            ? undefined
+            : retryPolicy.streamIdleTimeoutMs,
+      })) {
         startIndex += 1;
         deliveredEvent = true;
         reconnectDelayMs = idleRetryPolicy.baseDelayMs;
@@ -167,7 +196,9 @@ export async function* followStreamIterable(
         }
       }
     } catch (error) {
-      if (!isStreamDisconnectError(error)) {
+      if (isStreamIdleTimeoutError(error)) {
+        timedOutIdle = true;
+      } else if (!isStreamDisconnectError(error)) {
         throw error;
       }
     }
@@ -176,7 +207,9 @@ export async function* followStreamIterable(
       return;
     }
 
-    if (
+    if (timedOutIdle) {
+      reconnectDelayMs = idleRetryPolicy.baseDelayMs;
+    } else if (
       !deliveredEvent &&
       !initialConnection &&
       (idleReconnects += 1) >= idleRetryPolicy.maxAttempts

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

@@ -647,22 +647,154 @@ describe("ClientSession", () => {
     expect(fetchMock).toHaveBeenCalledOnce();
   });
 
-  it("does not reconnect a manually opened stream when disabled", async () => {
-    const fetchMock = vi
-      .spyOn(globalThis, "fetch")
-      .mockResolvedValue(createStreamResponse([{ type: "turn.started", data: {} }]));
+  it("does not apply idle timeouts to tail-relative streams", async () => {
+    const encoder = new TextEncoder();
+    let streamController: ReadableStreamDefaultController<Uint8Array> | undefined;
+    const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(
+      new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            streamController = controller;
+          },
+        }),
+      ),
+    );
     const session = createSession({ sessionId: "session_1", streamIndex: 0 });
 
-    const eventTypes: string[] = [];
-    for await (const event of session.stream({ streamReconnectPolicy: { reconnect: false } })) {
-      eventTypes.push(event.type);
+    vi.useFakeTimers();
+    try {
+      const consumed = (async () => {
+        for await (const _event of session.stream({
+          startIndex: -1,
+          streamReconnectPolicy: { streamIdleTimeoutMs: 10 },
+        })) {
+          // Drain the delayed event.
+        }
+      })();
+      await vi.advanceTimersByTimeAsync(20);
+      streamController?.enqueue(
+        encoder.encode(
+          `${JSON.stringify({
+            type: "session.waiting",
+            data: { continuationToken: "session-id", wait: "next-user-message" },
+          })}\n`,
+        ),
+      );
+      streamController?.close();
+      await consumed;
+    } finally {
+      vi.useRealTimers();
+    }
+
+    expect(fetchMock).toHaveBeenCalledOnce();
+  });
+
+  it("does not apply idle timeouts when stream reconnection is disabled", async () => {
+    const encoder = new TextEncoder();
+    let streamController: ReadableStreamDefaultController<Uint8Array> | undefined;
+    const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(
+      new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            streamController = controller;
+          },
+        }),
+      ),
+    );
+    const session = createSession({ sessionId: "session_1", streamIndex: 0 });
+
+    vi.useFakeTimers();
+    try {
+      const consumed = (async () => {
+        for await (const _event of session.stream({
+          streamReconnectPolicy: { reconnect: false },
+        })) {
+          // Drain the delayed event.
+        }
+      })();
+      await vi.advanceTimersByTimeAsync(31_000);
+      streamController?.enqueue(
+        encoder.encode(`${JSON.stringify({ type: "turn.started", data: {} })}\n`),
+      );
+      streamController?.close();
+      await consumed;
+    } finally {
+      vi.useRealTimers();
     }
 
-    expect(eventTypes).toEqual(["turn.started"]);
     expect(fetchMock).toHaveBeenCalledOnce();
     expect(session.state.streamIndex).toBe(1);
   });
 
+  it("does not apply idle timeouts when idle reconnect attempts are disabled", async () => {
+    const encoder = new TextEncoder();
+    let streamController: ReadableStreamDefaultController<Uint8Array> | undefined;
+    const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(
+      new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            streamController = controller;
+          },
+        }),
+      ),
+    );
+    const session = createSession({ sessionId: "session_1", streamIndex: 0 });
+
+    vi.useFakeTimers();
+    try {
+      const consumed = (async () => {
+        for await (const _event of session.stream({
+          streamReconnectPolicy: {
+            streamIdleReconnectPolicy: { maxAttempts: 0 },
+            streamIdleTimeoutMs: 10,
+          },
+        })) {
+          // Drain the delayed event.
+        }
+      })();
+      await vi.advanceTimersByTimeAsync(20);
+      streamController?.enqueue(
+        encoder.encode(`${JSON.stringify({ type: "turn.started", data: {} })}\n`),
+      );
+      streamController?.close();
+      await consumed;
+    } finally {
+      vi.useRealTimers();
+    }
+
+    expect(fetchMock).toHaveBeenCalledOnce();
+    expect(session.state.streamIndex).toBe(1);
+  });
+
+  it.each([-1, 1.5, 2_147_483_648, 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 an integer between 0 and 2147483647.");
+      expect(fetchMock).not.toHaveBeenCalled();
+    },
+  );
+
+  it("rejects an invalid stream idle timeout before submitting a turn", async () => {
+    const fetchMock = vi.spyOn(globalThis, "fetch");
+    const session = createSession({ sessionId: "session_1", streamIndex: 0 });
+
+    await expect(
+      session.send("first", {
+        streamReconnectPolicy: { streamIdleTimeoutMs: -1 },
+      }),
+    ).rejects.toThrow("streamIdleTimeoutMs must be an integer between 0 and 2147483647.");
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
   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") {