From b5faa8355d8b92b8e5d9e54ed452885abbf5f810 Mon Sep 17 00:00:00 2001 From: Andrew Barba Date: Fri, 24 Jul 2026 13:52:49 -0400 Subject: [PATCH 1/2] fix(eve): client streams - reconnect stalled open responses Signed-off-by: Andrew Barba --- .changeset/calm-streams-reconnect.md | 5 + docs/guides/client/streaming.mdx | 16 +++- packages/eve/src/client/ndjson.ts | 45 ++++++++- packages/eve/src/client/open-stream.ts | 20 +++- packages/eve/src/client/session.test.ts | 57 +++++++++++ .../client/stream-follow.integration.test.ts | 96 ++++++++++++++++++- packages/eve/src/client/types.ts | 8 ++ 7 files changed, 242 insertions(+), 5 deletions(-) create mode 100644 .changeset/calm-streams-reconnect.md diff --git a/.changeset/calm-streams-reconnect.md b/.changeset/calm-streams-reconnect.md new file mode 100644 index 000000000..7661f0e8b --- /dev/null +++ b/.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. diff --git a/docs/guides/client/streaming.mdx b/docs/guides/client/streaming.mdx index 8ba6fb3fd..a2eeaacfb 100644 --- a/docs/guides/client/streaming.mdx +++ b/docs/guides/client/streaming.mdx @@ -103,7 +103,21 @@ 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. + +Configure the idle-read deadline and the no-progress retry budget together: + +```ts +const response = await session.send({ + message: "Run the long operation.", + streamReconnectPolicy: { + streamIdleTimeoutMs: 15_000, + streamIdleReconnectPolicy: { maxAttempts: 10 }, + }, +}); +``` + +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: diff --git a/packages/eve/src/client/ndjson.ts b/packages/eve/src/client/ndjson.ts index 3187a369a..fba2df39f 100644 --- a/packages/eve/src/client/ndjson.ts +++ b/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,22 @@ 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; + } +} + +interface ReadNdjsonStreamOptions { + readonly idleTimeoutMs?: number; +} + +type StreamReadResult = Awaited["read"]>>; + /** * Reads newline-delimited JSON events from a `ReadableStream`. * @@ -35,6 +55,7 @@ export function isStreamDisconnectError(error: unknown): boolean { */ export async function* readNdjsonStream( body: ReadableStream, + options: ReadNdjsonStreamOptions = {}, ): AsyncGenerator { const reader = body.getReader(); const decoder = new TextDecoder(); @@ -43,7 +64,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 +105,25 @@ export async function* readNdjsonStream( reader.releaseLock(); } } + +async function readWithIdleTimeout( + reader: ReadableStreamDefaultReader, + idleTimeoutMs: number | undefined, +): Promise { + if (idleTimeoutMs === undefined) { + return await reader.read(); + } + + let timeout: ReturnType | undefined; + const timeoutPromise = new Promise((_resolve, reject) => { + timeout = setTimeout(() => reject(new StreamIdleTimeoutError(idleTimeoutMs)), idleTimeoutMs); + }); + + try { + return await Promise.race([reader.read(), timeoutPromise]); + } finally { + if (timeout !== undefined) { + clearTimeout(timeout); + } + } +} diff --git a/packages/eve/src/client/open-stream.ts b/packages/eve/src/client/open-stream.ts index e5d01265c..4c555f820 100644 --- a/packages/eve/src/client/open-stream.ts +++ b/packages/eve/src/client/open-stream.ts @@ -19,12 +19,14 @@ interface RetryPolicy { interface ResolvedStreamReconnectPolicy { readonly retryableErrorStatuses: ReadonlySet; 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 +36,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,6 +66,7 @@ function resolveStreamReconnectPolicy( configured?.streamIdleReconnectPolicy, DEFAULT_STREAM_RECONNECT_POLICY.streamIdleReconnectPolicy, ), + streamIdleTimeoutMs: resolveStreamIdleTimeoutMs(configured?.streamIdleTimeoutMs), streamOpenReconnectPolicy: resolveRetryPolicy( configured?.streamOpenReconnectPolicy, DEFAULT_STREAM_RECONNECT_POLICY.streamOpenReconnectPolicy, @@ -70,6 +74,18 @@ function resolveStreamReconnectPolicy( }; } +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. */ @@ -115,7 +131,9 @@ export async function* followStreamIterable( let deliveredEvent = 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; diff --git a/packages/eve/src/client/session.test.ts b/packages/eve/src/client/session.test.ts index c9f184d56..64755357f 100644 --- a/packages/eve/src/client/session.test.ts +++ b/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({ + 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[] = []; diff --git a/packages/eve/src/client/stream-follow.integration.test.ts b/packages/eve/src/client/stream-follow.integration.test.ts index a50fe39e4..0d07388b7 100644 --- a/packages/eve/src/client/stream-follow.integration.test.ts +++ b/packages/eve/src/client/stream-follow.integration.test.ts @@ -4,6 +4,7 @@ import { afterEach, describe, expect, it } from "vitest"; import { Client } from "./client.js"; import { followStreamIterable } from "./open-stream.js"; +import type { StreamReconnectPolicy } from "./types.js"; const servers: Server[] = []; @@ -27,12 +28,13 @@ function startIndexOf(url: string | undefined): number { return Number(new URL(url ?? "", "http://127.0.0.1").searchParams.get("startIndex") ?? "0"); } -function follow(host: string) { +function follow(host: string, streamReconnectPolicy?: StreamReconnectPolicy, startIndex = 0) { return followStreamIterable({ host, resolveHeaders: () => Promise.resolve(new Headers()), sessionId: "s", - startIndex: 0, + startIndex, + streamReconnectPolicy, }); } @@ -132,4 +134,94 @@ describe("stream following over real sockets", () => { expect(received).toEqual(events); expect(connections).toBe(3 * events.length); }, 30_000); + + it("reconnects an open silent stream from the last durable cursor", async () => { + const startIndexes: number[] = []; + let firstConnectionClosed = false; + const host = await listen( + createServer((req, res) => { + const startIndex = startIndexOf(req.url); + startIndexes.push(startIndex); + res.writeHead(200, { "content-type": "application/x-ndjson" }); + + if (startIndex === 0) { + req.once("close", () => { + firstConnectionClosed = true; + }); + res.write(`${JSON.stringify({ type: "message.received", data: {} })}\n`); + return; + } + + res.end(`${JSON.stringify({ type: "session.completed", data: {} })}\n`); + }), + ); + + const received: string[] = []; + for await (const event of follow(host, { + streamIdleReconnectPolicy: { baseDelayMs: 1, maxAttempts: 2, maxDelayMs: 1 }, + streamIdleTimeoutMs: 25, + })) { + received.push(event.type); + if (event.type === "session.completed") break; + } + + expect(received).toEqual(["message.received", "session.completed"]); + expect(startIndexes).toEqual([0, 1]); + expect(firstConnectionClosed).toBe(true); + }); + + it("does not apply idle-read reconnects to tail-relative streams", async () => { + let connections = 0; + const host = await listen( + createServer((_req, res) => { + connections += 1; + res.writeHead(200, { "content-type": "application/x-ndjson" }); + setTimeout( + () => res.end(`${JSON.stringify({ type: "session.completed", data: {} })}\n`), + 40, + ); + }), + ); + + const received: string[] = []; + for await (const event of follow( + host, + { + streamIdleReconnectPolicy: { baseDelayMs: 1, maxAttempts: 1, maxDelayMs: 1 }, + streamIdleTimeoutMs: 10, + }, + -1, + )) { + received.push(event.type); + } + + expect(received).toEqual(["session.completed"]); + expect(connections).toBe(1); + }); + + it("allows idle-read reconnects to be disabled for absolute cursors", async () => { + let connections = 0; + const host = await listen( + createServer((_req, res) => { + connections += 1; + res.writeHead(200, { "content-type": "application/x-ndjson" }); + setTimeout( + () => res.end(`${JSON.stringify({ type: "session.completed", data: {} })}\n`), + 40, + ); + }), + ); + + const received: string[] = []; + for await (const event of follow(host, { + streamIdleReconnectPolicy: { baseDelayMs: 1, maxAttempts: 1, maxDelayMs: 1 }, + streamIdleTimeoutMs: 0, + })) { + received.push(event.type); + if (event.type === "session.completed") break; + } + + expect(received).toEqual(["session.completed"]); + expect(connections).toBe(1); + }); }); diff --git a/packages/eve/src/client/types.ts b/packages/eve/src/client/types.ts index c9e07e6f4..56d12b096 100644 --- a/packages/eve/src/client/types.ts +++ b/packages/eve/src/client/types.ts @@ -184,6 +184,14 @@ export interface StreamReconnectRetryPolicy { /** Configurable policy used when automatic stream reconnection is enabled. */ export interface ResolvedStreamReconnectPolicy { + /** + * Milliseconds without stream bytes before reconnecting from the durable + * cursor. Set to `0` to disable idle-read reconnects. + * + * @default 30000 + */ + readonly streamIdleTimeoutMs?: number; + /** Retry policy for opening an HTTP stream connection. */ readonly streamOpenReconnectPolicy?: StreamReconnectRetryPolicy; From 9fd3015d4b68407700c07e67a913fc1220ec36cb Mon Sep 17 00:00:00 2001 From: Andrew Barba Date: Fri, 24 Jul 2026 15:40:12 -0400 Subject: [PATCH 2/2] fix(eve): client streams - keep following a stream that went quiet eve emits no bytes between actions.requested and action.result, so a slow step looks identical to a stalled connection. Charging idle-read reconnects to streamIdleReconnectPolicy ended the follow after about three minutes of legitimate quiet, with no terminal event. That budget now covers only reconnects that end without delivering events. Signed-off-by: Andrew Barba --- docs/guides/client/streaming.mdx | 9 ++-- packages/eve/src/client/ndjson.ts | 12 +++++ packages/eve/src/client/open-stream.ts | 16 +++++- .../client/stream-follow.integration.test.ts | 52 +++++++++++++++++++ 4 files changed, 82 insertions(+), 7 deletions(-) diff --git a/docs/guides/client/streaming.mdx b/docs/guides/client/streaming.mdx index a2eeaacfb..b4c51ad39 100644 --- a/docs/guides/client/streaming.mdx +++ b/docs/guides/client/streaming.mdx @@ -105,18 +105,17 @@ If you support refresh while an authorization prompt is pending, keep the sessio 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. -Configure the idle-read deadline and the no-progress retry budget together: +Tune the idle-read deadline with `streamIdleTimeoutMs`: ```ts const response = await session.send({ message: "Run the long operation.", - streamReconnectPolicy: { - streamIdleTimeoutMs: 15_000, - streamIdleReconnectPolicy: { maxAttempts: 10 }, - }, + 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: diff --git a/packages/eve/src/client/ndjson.ts b/packages/eve/src/client/ndjson.ts index fba2df39f..5fae56dcc 100644 --- a/packages/eve/src/client/ndjson.ts +++ b/packages/eve/src/client/ndjson.ts @@ -38,6 +38,18 @@ class StreamIdleTimeoutError extends Error { } } +/** + * 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; } diff --git a/packages/eve/src/client/open-stream.ts b/packages/eve/src/client/open-stream.ts index 4c555f820..2d4860e8c 100644 --- a/packages/eve/src/client/open-stream.ts +++ b/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, @@ -130,6 +134,7 @@ export async function* followStreamIterable( } let deliveredEvent = false; + let timedOutIdle = false; try { for await (const event of readNdjsonStream(body, { idleTimeoutMs: input.startIndex < 0 ? undefined : retryPolicy.streamIdleTimeoutMs, @@ -144,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 diff --git a/packages/eve/src/client/stream-follow.integration.test.ts b/packages/eve/src/client/stream-follow.integration.test.ts index 0d07388b7..67648e7b8 100644 --- a/packages/eve/src/client/stream-follow.integration.test.ts +++ b/packages/eve/src/client/stream-follow.integration.test.ts @@ -170,6 +170,58 @@ describe("stream following over real sockets", () => { expect(firstConnectionClosed).toBe(true); }); + it("keeps following a stream the server holds open across a silent step", async () => { + // eve emits no bytes at all between `actions.requested` and `action.result`, + // so a slow step looks exactly like this to the client. + const silentConnections = 4; + let connections = 0; + const host = await listen( + createServer((_req, res) => { + connections += 1; + res.writeHead(200, { "content-type": "application/x-ndjson" }); + res.write("\n"); + + if (connections > silentConnections) { + res.end(`${JSON.stringify({ type: "session.completed", data: {} })}\n`); + } + }), + ); + + const received: string[] = []; + for await (const event of follow(host, { + streamIdleReconnectPolicy: { baseDelayMs: 1, maxAttempts: 2, maxDelayMs: 1 }, + streamIdleTimeoutMs: 25, + })) { + received.push(event.type); + if (event.type === "session.completed") break; + } + + expect(received).toEqual(["session.completed"]); + expect(connections).toBe(silentConnections + 1); + }, 30_000); + + it("still stops following when reconnects keep ending without events", async () => { + let connections = 0; + const host = await listen( + createServer((_req, res) => { + connections += 1; + res.writeHead(200, { "content-type": "application/x-ndjson" }); + res.end("\n"); + }), + ); + + const received: string[] = []; + for await (const event of follow(host, { + streamIdleReconnectPolicy: { baseDelayMs: 1, maxAttempts: 2, maxDelayMs: 1 }, + streamIdleTimeoutMs: 25, + })) { + received.push(event.type); + } + + expect(received).toEqual([]); + expect(connections).toBe(3); + }); + it("does not apply idle-read reconnects to tail-relative streams", async () => { let connections = 0; const host = await listen(