From 2a231735c483953c562980b2f3e9b29d4b41bb70 Mon Sep 17 00:00:00 2001 From: "vercel-gh-bot-5[bot]" <312521305+vercel-gh-bot-5[bot]@users.noreply.github.com> Date: Thu, 6 Aug 2026 10:30:48 +0000 Subject: [PATCH] fix(eve): reconnect silent client streams Co-authored-by: ruiconti <1834568+ruiconti@users.noreply.github.com> --- .changeset/agents-listing-system-channel.md | 5 - .changeset/calm-streams-reconnect.md | 5 + .../flush-dev-workflow-stream-headers.md | 5 - docs/guides/client/streaming.mdx | 12 +- docs/subagents.mdx | 2 +- e2e/fixtures/agent-subagents/package.json | 3 - packages/eve/src/client/ndjson.ts | 42 ++++- packages/eve/src/client/open-stream.ts | 41 ++++- packages/eve/src/client/session.test.ts | 148 +++++++++++++++++- packages/eve/src/client/session.ts | 3 +- .../client/stream-follow.integration.test.ts | 127 ++++++++++++++- packages/eve/src/client/types.ts | 8 + .../agent-messaging.scenario.test.ts | 4 +- .../eve/src/harness/handles/prompt.test.ts | 19 +-- packages/eve/src/harness/handles/prompt.ts | 28 +--- packages/eve/src/harness/tool-loop.test.ts | 102 +----------- packages/eve/src/harness/tool-loop.ts | 8 +- .../internal/nitro/host/dev-server-http.ts | 4 - .../drained-nitro-dev-server.scenario.test.ts | 77 --------- .../runtime/agent/bootstrap-model-utils.ts | 20 --- .../runtime/agent/mock-model-adapter.test.ts | 62 -------- .../src/runtime/agent/mock-model-adapter.ts | 9 -- .../src/runtime/agent/mock-model-fixtures.ts | 11 +- packages/eve/src/runtime/prompt/compose.ts | 2 +- 24 files changed, 388 insertions(+), 359 deletions(-) delete mode 100644 .changeset/agents-listing-system-channel.md create mode 100644 .changeset/calm-streams-reconnect.md delete mode 100644 .changeset/flush-dev-workflow-stream-headers.md diff --git a/.changeset/agents-listing-system-channel.md b/.changeset/agents-listing-system-channel.md deleted file mode 100644 index 245e434f5..000000000 --- a/.changeset/agents-listing-system-channel.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"eve": patch ---- - -Agent-messaging `` listings are now announced as framework-injected user-role notes instead of assistant messages appended to history. This fixes parent resume failures on models that reject assistant-final requests (e.g. `This model does not support assistant message prefill` from Claude via AI Gateway) after a persistent child parks, keeps the announcement append-only so provider prompt caches stay warm, and the agent-messaging system prompt now declares the `[Agents]` note as framework-injected. diff --git a/.changeset/calm-streams-reconnect.md b/.changeset/calm-streams-reconnect.md new file mode 100644 index 000000000..ac1e6c076 --- /dev/null +++ b/.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. diff --git a/.changeset/flush-dev-workflow-stream-headers.md b/.changeset/flush-dev-workflow-stream-headers.md deleted file mode 100644 index 2cdb1aa39..000000000 --- a/.changeset/flush-dev-workflow-stream-headers.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"eve": patch ---- - -Flush local development streaming response headers immediately so pending Workflow streams can be cancelled without accumulating listeners. diff --git a/docs/guides/client/streaming.mdx b/docs/guides/client/streaming.mdx index 836d13c02..810476674 100644 --- a/docs/guides/client/streaming.mdx +++ b/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). diff --git a/docs/subagents.mdx b/docs/subagents.mdx index 9f597b432..4dfbccb78 100644 --- a/docs/subagents.mdx +++ b/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 `` 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 `` 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. diff --git a/e2e/fixtures/agent-subagents/package.json b/e2e/fixtures/agent-subagents/package.json index 0ee8d2fd7..927e0e7a6 100644 --- a/e2e/fixtures/agent-subagents/package.json +++ b/e2e/fixtures/agent-subagents/package.json @@ -22,8 +22,5 @@ "devDependencies": { "@types/node": "catalog:", "typescript": "catalog:" - }, - "e2e": { - "modelMatrix": "full" } } diff --git a/packages/eve/src/client/ndjson.ts b/packages/eve/src/client/ndjson.ts index 317c391c1..309decc66 100644 --- a/packages/eve/src/client/ndjson.ts +++ b/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`. * @@ -35,6 +50,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 +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, + idleTimeoutMs: number | undefined, +): ReturnType["read"]> { + if (idleTimeoutMs === undefined) { + return await reader.read(); + } + + let timeout: ReturnType | undefined; + try { + return await Promise.race([ + reader.read(), + new Promise((_resolve, reject) => { + timeout = setTimeout( + () => reject(new StreamIdleTimeoutError(idleTimeoutMs)), + idleTimeoutMs, + ); + }), + ]); + } 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 b0d8007d5..0519ab9aa 100644 --- a/packages/eve/src/client/open-stream.ts +++ b/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; 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,6 +73,7 @@ function resolveStreamReconnectPolicy( configured?.streamIdleReconnectPolicy, DEFAULT_STREAM_RECONNECT_POLICY.streamIdleReconnectPolicy, ), + streamIdleTimeoutMs: resolveStreamIdleTimeoutMs(configured?.streamIdleTimeoutMs), streamOpenReconnectPolicy: resolveRetryPolicy( configured?.streamOpenReconnectPolicy, DEFAULT_STREAM_RECONNECT_POLICY.streamOpenReconnectPolicy, @@ -71,6 +81,19 @@ function resolveStreamReconnectPolicy( }; } +/** @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 diff --git a/packages/eve/src/client/session.test.ts b/packages/eve/src/client/session.test.ts index c0818ca50..283271ede 100644 --- a/packages/eve/src/client/session.test.ts +++ b/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 | undefined; + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + new ReadableStream({ + 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 | undefined; + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + new ReadableStream({ + 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 | undefined; + const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + new ReadableStream({ + 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") { diff --git a/packages/eve/src/client/session.ts b/packages/eve/src/client/session.ts index e1f80317c..7c3563921 100644 --- a/packages/eve/src/client/session.ts +++ b/packages/eve/src/client/session.ts @@ -3,7 +3,7 @@ import { EVE_SESSION_ID_HEADER, isCurrentTurnBoundaryEvent } from "#protocol/mes import { EVE_SESSION_ROUTE_PATH, createEveSessionRoutePath } from "#protocol/routes.js"; import { ClientError } from "#client/client-error.js"; import { MessageResponse } from "#client/message-response.js"; -import { followStreamIterable } from "#client/open-stream.js"; +import { followStreamIterable, validateStreamReconnectPolicy } from "#client/open-stream.js"; import { cancelClientSession, clearClientSession, @@ -248,6 +248,7 @@ async function postTurn( input: SendTurnPayload, requireMessage: boolean, ): Promise { + validateStreamReconnectPolicy(input.streamReconnectPolicy); const body = createMessageBody(input, requireMessage); if (body === null) { throw new Error( diff --git a/packages/eve/src/client/stream-follow.integration.test.ts b/packages/eve/src/client/stream-follow.integration.test.ts index 32295239c..fe8090639 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,13 +28,17 @@ function startIndexOf(url: string | undefined): number { return Number(new URL(url ?? "", "http://127.0.0.1").searchParams.get("startIndex") ?? "0"); } -function follow(host: string, options?: { follow?: boolean }) { +function follow( + host: string, + options?: { follow?: boolean; streamReconnectPolicy?: StreamReconnectPolicy }, +) { return followStreamIterable({ follow: options?.follow, host, resolveHeaders: () => Promise.resolve(new Headers()), sessionId: "s", startIndex: 0, + streamReconnectPolicy: options?.streamReconnectPolicy, }); } @@ -221,4 +226,124 @@ describe("stream following over real sockets", () => { expect(received).toEqual(events); expect(connections).toBe(3 * events.length); }, 30_000); + + it("reconnects an open silent response 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: "step.started", data: {} })}\n`); + return; + } + + res.end(`${JSON.stringify({ type: "session.completed", data: {} })}\n`); + }), + ); + + const received: string[] = []; + for await (const event of follow(host, { + streamReconnectPolicy: { + streamIdleReconnectPolicy: { baseDelayMs: 1, maxAttempts: 1, maxDelayMs: 1 }, + streamIdleTimeoutMs: 25, + }, + })) { + received.push(event.type); + if (event.type === "session.completed") break; + } + + expect(received).toEqual(["step.started", "session.completed"]); + expect(startIndexes).toEqual([0, 1]); + expect(firstConnectionClosed).toBe(true); + }); + + it("keeps a connection when bytes arrive within the idle deadline", async () => { + let connections = 0; + const event = `${JSON.stringify({ type: "session.completed", data: {} })}\n`; + const host = await listen( + createServer(async (_req, res) => { + connections += 1; + res.writeHead(200, { "content-type": "application/x-ndjson" }); + for (let index = 0; index < event.length; index += 10) { + res.write(event.slice(index, index + 10)); + await new Promise((resolve) => setTimeout(resolve, 15)); + } + res.end(); + }), + ); + + const received: string[] = []; + for await (const streamEvent of follow(host, { + streamReconnectPolicy: { streamIdleTimeoutMs: 25 }, + })) { + received.push(streamEvent.type); + if (streamEvent.type === "session.completed") break; + } + + expect(received).toEqual(["session.completed"]); + expect(connections).toBe(1); + }); + + it("allows idle-read reconnects to be disabled", 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, { + streamReconnectPolicy: { + 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); + }); + + it("does not spend the empty-stream retry budget on silent open responses", async () => { + let connections = 0; + const host = await listen( + createServer((_req, res) => { + connections += 1; + res.writeHead(200, { "content-type": "application/x-ndjson" }); + res.flushHeaders(); + if (connections === 3) { + res.end(`${JSON.stringify({ type: "session.completed", data: {} })}\n`); + } + }), + ); + + const received: string[] = []; + for await (const event of follow(host, { + streamReconnectPolicy: { + streamIdleReconnectPolicy: { baseDelayMs: 1, maxAttempts: 1, maxDelayMs: 1 }, + streamIdleTimeoutMs: 25, + }, + })) { + received.push(event.type); + if (event.type === "session.completed") break; + } + + expect(received).toEqual(["session.completed"]); + expect(connections).toBe(3); + }); }); diff --git a/packages/eve/src/client/types.ts b/packages/eve/src/client/types.ts index 6167083fe..9da16633b 100644 --- a/packages/eve/src/client/types.ts +++ b/packages/eve/src/client/types.ts @@ -174,6 +174,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; diff --git a/packages/eve/src/execution/agent-messaging.scenario.test.ts b/packages/eve/src/execution/agent-messaging.scenario.test.ts index 0317e9d54..4ae1d7f04 100644 --- a/packages/eve/src/execution/agent-messaging.scenario.test.ts +++ b/packages/eve/src/execution/agent-messaging.scenario.test.ts @@ -44,8 +44,8 @@ const model = mockModel((request) => { } if (childResults.length === 1) { - // The agents listing rides the conversation as a framework-injected - // user-role announcement, so scan every message for the latest listing. + // The agents listing rides the conversation as an assistant announcement + // (not the system prompt), so scan every message for the latest listing. const agentsSnippet = request.messages.map((message) => message.text).join("\\n"); const agentId = AGENT_ID_PATTERN.exec(agentsSnippet)?.[1]; if (agentId === undefined) { diff --git a/packages/eve/src/harness/handles/prompt.test.ts b/packages/eve/src/harness/handles/prompt.test.ts index 906a7dd6d..64a450863 100644 --- a/packages/eve/src/harness/handles/prompt.test.ts +++ b/packages/eve/src/harness/handles/prompt.test.ts @@ -80,12 +80,12 @@ describe("resolveAgentsAnnouncement", () => { expect(messages).toEqual([{ content: "Earlier response", role: "assistant" }]); }); - it("does not repeat an unchanged listing already announced as user content", () => { + it("does not repeat an unchanged listing", () => { const content = renderAgentsSnippet({ handles: [parkedRemoteHandle] }); expect( resolveAgentsAnnouncement({ - messages: [{ content, role: "user" }], + messages: [{ content, role: "assistant" }], store: { handles: [parkedRemoteHandle] }, }), ).toBeUndefined(); @@ -96,7 +96,7 @@ describe("resolveAgentsAnnouncement", () => { expect( resolveAgentsAnnouncement({ - messages: [{ content: previous, role: "user" }], + messages: [{ content: previous, role: "assistant" }], store: { handles: [runningHandle] }, }), ).toBe("[Agents]\n\n"); @@ -105,17 +105,4 @@ describe("resolveAgentsAnnouncement", () => { it("skips empty scaffolding when no listing was previously announced", () => { expect(resolveAgentsAnnouncement({ messages: [], store: undefined })).toBeUndefined(); }); - - it("ignores an assistant message that happens to start with the label", () => { - const content = renderAgentsSnippet({ handles: [parkedRemoteHandle] }); - - // Only framework-injected user-role announcements gate re-announcement; - // model output echoing the label must not suppress a fresh listing. - expect( - resolveAgentsAnnouncement({ - messages: [{ content, role: "assistant" }], - store: { handles: [parkedRemoteHandle] }, - }), - ).toBe(content); - }); }); diff --git a/packages/eve/src/harness/handles/prompt.ts b/packages/eve/src/harness/handles/prompt.ts index b6399ad55..c2fe0a6d1 100644 --- a/packages/eve/src/harness/handles/prompt.ts +++ b/packages/eve/src/harness/handles/prompt.ts @@ -2,12 +2,7 @@ import type { ModelMessage } from "ai"; import type { AgentHandle, AgentHandleStore } from "#harness/handles/store.js"; -/** - * Label prefixing every framework-injected agents announcement. Mock model - * adapters use it to treat announcements as transparent scaffolding rather - * than authored user input. - */ -export const AGENTS_SNIPPET_LABEL = "[Agents]"; +const AGENTS_SNIPPET_LABEL = "[Agents]"; /** Returns the resumable handles: the only phase the model may continue. */ export function projectParkedAgentHandles( @@ -30,22 +25,9 @@ export function renderAgentsSnippet(store: AgentHandleStore): string { } /** - * Returns an append-only announcement when the visible handle listing - * changed since the last one in history, or `undefined` when it is - * unchanged. - * - * The announcement is framework-injected `user`-role conversation content - * (the pattern system-reminder notes use in Claude Code and OpenCode), not - * an `assistant` or `system` entry: - * - * - `assistant` breaks providers that reject assistant-final requests - * (a settle resume carries no new user input, so the announcement would - * end the request) and invites the model to imitate the listing. - * - `system` busts the provider prompt cache for the entire conversation - * every time a child settles; append-only history preserves the prefix. - * - * The static agent-messaging prompt block declares the `[Agents]` label as - * eve-injected so the model does not attribute it to the user. + * Returns an append-only model announcement when the visible handle listing + * changed. Keeping volatile handles in conversation history preserves the + * stable system prompt cache prefix. */ export function resolveAgentsAnnouncement(input: { readonly messages: readonly ModelMessage[]; @@ -53,7 +35,7 @@ export function resolveAgentsAnnouncement(input: { }): string | undefined { const latest = input.messages.findLast( (message) => - message.role === "user" && + message.role === "assistant" && typeof message.content === "string" && message.content.startsWith(AGENTS_SNIPPET_LABEL), ); diff --git a/packages/eve/src/harness/tool-loop.test.ts b/packages/eve/src/harness/tool-loop.test.ts index d5ab19cef..bbfe76275 100644 --- a/packages/eve/src/harness/tool-loop.test.ts +++ b/packages/eve/src/harness/tool-loop.test.ts @@ -692,7 +692,7 @@ describe("createToolLoopHarness", () => { expect(agentCall!.tools).not.toHaveProperty("Workflow"); }); - it("announces parked agents as user-role content before the user message, outside the system prompt", async () => { + it("announces current agents outside the cacheable system prompt", async () => { setupMockAgent({ finishReason: "stop", response: { messages: [{ content: "Hello!", role: "assistant" }] }, @@ -742,25 +742,18 @@ describe("createToolLoopHarness", () => { generate: ReturnType; }; const messages = agent.generate.mock.calls[0]?.[0].messages as ModelMessage[]; - // The volatile listing stays out of the system prompt (prompt cache) and - // rides history as a labeled, framework-injected user message. expect(instructions).toBe("You are a test assistant."); expect(messages).toContainEqual({ content: expect.stringContaining( 'waiting', ), - role: "user", + role: "assistant", }); - // The announcement precedes the turn's actual user message. expect(messages.at(-1)).toEqual({ content: "Hi", role: "user" }); - expect(messages.at(-2)).toEqual({ - content: expect.stringContaining("[Agents]"), - role: "user", - }); expect(JSON.stringify({ instructions, messages })).not.toContain("private-token"); expect(result.session.history).toContainEqual({ content: expect.stringContaining(' { expect(JSON.stringify(messages)).not.toContain(""); }); - // Regression: a child settling used to append the updated listing - // to history as an assistant message. On a resume with no new user input the - // request then ended with `assistant`, which Anthropic rejects ("this model - // does not support assistant message prefill"). The announcement is - // user-role content, so on a no-input resume it trails the tool results - // and the request stays user-final. - it("keeps a no-input resume provider-valid when a parked handle is announced", async () => { - setupMockAgent({ - finishReason: "stop", - response: { messages: [{ content: "Done.", role: "assistant" }] }, - text: "Done.", - toolCalls: [], - toolResults: [], - }); - - const runStep = createToolLoopHarness( - createTestConfig("conversation", undefined, { persistentSubagentSessions: true }), - ); - const session = createTestSession({ - history: [ - { content: "Delegate this.", role: "user" }, - { - content: [ - { - input: { message: "do it" }, - toolCallId: "call-1", - toolName: "research", - type: "tool-call", - }, - ], - role: "assistant", - }, - { - content: [ - { - output: { type: "text", value: "child answered" }, - toolCallId: "call-1", - toolName: "research", - type: "tool-result", - }, - ], - role: "tool", - }, - ], - state: { - [AGENT_HANDLES_STATE_KEY]: { - handles: [ - { - address: { - continuationToken: "private-token", - kind: "agent/local", - sessionId: "child-session-123456789012", - }, - identity: { - id: "ag_research:123456789012", - name: "research", - nodeId: "subagents/research", - }, - lastStatus: "child answered", - phase: "parked", - }, - ], - }, - }, - }); - - const result = await runStep(session); - - const { instructions } = vi.mocked(ToolLoopAgent).mock.calls[0]![0]; - const agent = vi.mocked(ToolLoopAgent).mock.results[0]?.value as { - generate: ReturnType; - }; - const messages = agent.generate.mock.calls[0]?.[0].messages as ModelMessage[]; - // The request ends user-final: the announcement trails the tool results. - expect(messages.at(-1)).toEqual({ - content: expect.stringContaining(' message.role === "assistant")).toHaveLength(1); - // The volatile listing never rides the system prompt (prompt cache). - expect(JSON.stringify(instructions ?? "")).not.toContain(""); - // The announcement persists append-only so the next step's diff gate - // sees it and does not re-announce an unchanged listing. - expect(result.session.history.at(-2)).toEqual({ - content: expect.stringContaining("[Agents]"), - role: "user", - }); - }); - it("uses dynamic model selection for the model call", async () => { setupMockAgent({ finishReason: "stop", diff --git a/packages/eve/src/harness/tool-loop.ts b/packages/eve/src/harness/tool-loop.ts index fbf2c0350..110b80b65 100644 --- a/packages/eve/src/harness/tool-loop.ts +++ b/packages/eve/src/harness/tool-loop.ts @@ -754,19 +754,13 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { } session = continuation.session; - // Announce the parked-agents listing as framework-injected user-role - // content, before any new user input so a message present on this step - // stays the turn's focus. On a no-input settle resume it trails the tool - // results, keeping the request user-final for providers that reject - // assistant-final histories. See resolveAgentsAnnouncement for the role - // rationale (assistant-final rejection, prompt-cache preservation). if (config.persistentSubagentSessions === true) { const announcement = resolveAgentsAnnouncement({ messages, store: getAgentHandleStore(session.state), }); if (announcement !== undefined) { - messages.push({ content: announcement, role: "user" }); + messages.push({ content: announcement, role: "assistant" }); } } diff --git a/packages/eve/src/internal/nitro/host/dev-server-http.ts b/packages/eve/src/internal/nitro/host/dev-server-http.ts index de3123826..ff16c30d9 100644 --- a/packages/eve/src/internal/nitro/host/dev-server-http.ts +++ b/packages/eve/src/internal/nitro/host/dev-server-http.ts @@ -59,10 +59,6 @@ export async function writeResponse( return; } - // Flush the headers now so the client can obtain / cancel the response body - // avoiding, the accumulation of listeners if the stream has a long delay - // before first byte - response.flushHeaders(); const body = Readable.fromWeb(webResponse.body as import("node:stream/web").ReadableStream); const cancelBody = () => body.destroy(signal.reason as Error | undefined); signal.addEventListener("abort", cancelBody, { once: true }); diff --git a/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts b/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts index 74d411998..85ba4b94f 100644 --- a/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts +++ b/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts @@ -1,7 +1,4 @@ -import { mkdtemp, rm } from "node:fs/promises"; import { connect } from "node:net"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; import { describe, expect, it, vi } from "vitest"; @@ -18,12 +15,6 @@ import type { DevelopmentRunner, DevelopmentRunnerFactory, } from "#internal/nitro/host/dev-runner.js"; -import { createDevelopmentWorkflowWorld } from "#internal/workflow/development-world-client.js"; -import { createParentDevelopmentWorkflowWorld } from "#internal/workflow/development-world-server.js"; -import { - DEVELOPMENT_WORKER_APP_ROOT_ENV, - DEVELOPMENT_WORKFLOW_SECRET_ENV, -} from "#internal/workflow/development-world-protocol.js"; const TEST_DEADLINE_MS = 5_000; const LOGGER = { error: () => undefined }; @@ -585,50 +576,6 @@ describe("drained Nitro dev server", () => { await server.close(); }); - it("cancels parent-owned Workflow streams before their first chunk", async () => { - const appRoot = await mkdtemp(join(tmpdir(), "eve-dev-world-stream-cancel-")); - const secret = "scenario-workflow-transport-secret"; - const world = createParentDevelopmentWorkflowWorld({ - agentName: "dev-world-stream-cancel-test", - appRoot, - resolveActiveGenerationId: () => "generation-a", - transportSecret: secret, - }); - const server = new DrainedNitroDevServer(LOGGER); - server.setControlHandler(async (request) => await world.handleRequest(request)); - const listener = await listen(server); - const previousBaseUrl = process.env.WORKFLOW_LOCAL_BASE_URL; - const previousSecret = process.env[DEVELOPMENT_WORKFLOW_SECRET_ENV]; - const previousAppRoot = process.env[DEVELOPMENT_WORKER_APP_ROOT_ENV]; - process.env.WORKFLOW_LOCAL_BASE_URL = listener.url; - process.env[DEVELOPMENT_WORKFLOW_SECRET_ENV] = secret; - process.env[DEVELOPMENT_WORKER_APP_ROOT_ENV] = appRoot; - const emitWarning = vi.spyOn(process, "emitWarning").mockImplementation(() => undefined); - - try { - await world.start(); - const worker = createDevelopmentWorkflowWorld(); - const streamName = "strm_http_cancel_test_system_abort"; - for (let index = 0; index < 12; index += 1) { - const stream = await withinDeadline( - worker.streams.get("wrun_http_cancel_test", streamName), - "Timed out waiting for Workflow stream response headers.", - ); - await stream.cancel(); - } - - expect(maxListenerWarningTypes(emitWarning.mock.calls)).toEqual([]); - } finally { - emitWarning.mockRestore(); - restoreEnvironmentVariable("WORKFLOW_LOCAL_BASE_URL", previousBaseUrl); - restoreEnvironmentVariable(DEVELOPMENT_WORKFLOW_SECRET_ENV, previousSecret); - restoreEnvironmentVariable(DEVELOPMENT_WORKER_APP_ROOT_ENV, previousAppRoot); - await server.close(); - await world.close(); - await rm(appRoot, { force: true, recursive: true }); - } - }); - it("answers control requests without admitting them to the worker", async () => { const fetchMock = vi.fn(async () => new Response("worker")); const { createRunner } = createRunnerFactory(fetchMock); @@ -650,27 +597,3 @@ describe("drained Nitro dev server", () => { await server.close(); }); }); - -function maxListenerWarningTypes(calls: readonly (readonly unknown[])[]): string[] { - return calls.flatMap(([warning]) => { - if ( - typeof warning !== "object" || - warning === null || - !("name" in warning) || - warning.name !== "MaxListenersExceededWarning" || - !("type" in warning) || - typeof warning.type !== "string" - ) { - return []; - } - return [warning.type]; - }); -} - -function restoreEnvironmentVariable(name: string, value: string | undefined): void { - if (value === undefined) { - delete process.env[name]; - } else { - process.env[name] = value; - } -} diff --git a/packages/eve/src/runtime/agent/bootstrap-model-utils.ts b/packages/eve/src/runtime/agent/bootstrap-model-utils.ts index 9f5cac87a..131819117 100644 --- a/packages/eve/src/runtime/agent/bootstrap-model-utils.ts +++ b/packages/eve/src/runtime/agent/bootstrap-model-utils.ts @@ -1,7 +1,5 @@ import type { MockLanguageModelV3 } from "ai/test"; -import { AGENTS_SNIPPET_LABEL } from "#harness/handles/prompt.js"; - export type BootstrapGenerateOptions = Parameters[0]; export type BootstrapPrompt = BootstrapGenerateOptions["prompt"]; export type BootstrapGenerateResult = Awaited>; @@ -157,10 +155,6 @@ export function getPromptContentText(content: BootstrapPrompt[number]["content"] /** * Returns the text from the last user message in the prompt, or `null`. - * - * Skips framework-injected `[Agents]` announcements: they ride the user - * role in conversation history, but they are not authored input and must - * not drive mock directive parsing. */ export function getLastUserPromptText(prompt: BootstrapPrompt): string | null { for (const message of [...prompt].reverse()) { @@ -170,10 +164,6 @@ export function getLastUserPromptText(prompt: BootstrapPrompt): string | null { const text = getPromptContentText(message.content).trim(); - if (isAgentsAnnouncementText(text)) { - continue; - } - if (text.length > 0) { return text; } @@ -182,16 +172,6 @@ export function getLastUserPromptText(prompt: BootstrapPrompt): string | null { return null; } -/** - * True when the text is a framework-injected `[Agents]` announcement. - * Announcements are user-role scaffolding, not authored input: mock model - * heuristics must scan past them instead of treating them as the turn's - * message or as a turn boundary. - */ -export function isAgentsAnnouncementText(text: string): boolean { - return text.startsWith(AGENTS_SNIPPET_LABEL); -} - /** * Joins all message content in the prompt into a single string. */ diff --git a/packages/eve/src/runtime/agent/mock-model-adapter.test.ts b/packages/eve/src/runtime/agent/mock-model-adapter.test.ts index d30c0a182..3e5127761 100644 --- a/packages/eve/src/runtime/agent/mock-model-adapter.test.ts +++ b/packages/eve/src/runtime/agent/mock-model-adapter.test.ts @@ -699,68 +699,6 @@ describe("createMockAuthoredRuntimeModel", () => { ]); }); - // Regression: the [Agents] announcement is user-role scaffolding injected - // after a subagent settles. Treating it as a turn boundary masked the tool - // result, and the adapter re-issued the same deterministic tool call — a - // duplicate start operation that fatally failed the parent session in the - // mock world suites. - it("replies to a tool result behind a framework [Agents] announcement instead of re-calling", async () => { - const result = await generateWithPrompt( - [ - { - content: "Call conditional-marker exactly once.", - role: "user", - }, - { - content: [ - { - input: JSON.stringify({ message: "run" }), - toolCallId: "call_conditional_marker", - toolName: "conditional-marker", - type: "tool-call", - }, - ], - role: "assistant", - }, - { - content: [ - { - output: { type: "json", value: "DYNAMIC_SUBAGENT_ENABLED" }, - toolCallId: "call_conditional_marker", - toolName: "conditional-marker", - type: "tool-result", - }, - ], - role: "tool", - }, - { - content: - '[Agents]\n\nDYNAMIC_SUBAGENT_ENABLED\n', - role: "user", - }, - ], - [ - { - inputSchema: { - properties: { message: { type: "string" } }, - required: ["message"], - type: "object", - }, - name: "conditional-marker", - type: "function", - }, - ], - ); - - expect(result.finishReason).toEqual({ raw: undefined, unified: "stop" }); - expect(result.content).toEqual([ - { - text: 'Used conditional-marker for "Call conditional-marker exactly once.": DYNAMIC_SUBAGENT_ENABLED', - type: "text", - }, - ]); - }); - it("does not reuse a prior turn's tool result after a later user message", async () => { const result = await generateWithPrompt([ { diff --git a/packages/eve/src/runtime/agent/mock-model-adapter.ts b/packages/eve/src/runtime/agent/mock-model-adapter.ts index fc3c2bcd0..58d1a8066 100644 --- a/packages/eve/src/runtime/agent/mock-model-adapter.ts +++ b/packages/eve/src/runtime/agent/mock-model-adapter.ts @@ -23,7 +23,6 @@ import { getLastUserPromptText, getPromptContentText, getPromptText, - isAgentsAnnouncementText, } from "#runtime/agent/bootstrap-model-utils.js"; import { findRelevantSkill, @@ -485,14 +484,6 @@ function getAvailableTools(options: BootstrapGenerateOptions): AvailableBootstra function getLastAuthoredToolResult(prompt: BootstrapPrompt): BootstrapToolResult | null { for (const message of [...prompt].reverse()) { if (message.role === "user") { - // A framework-injected [Agents] announcement is scaffolding, not a - // turn boundary. Treating it as one masks the tool result behind it, - // and the adapter then re-issues the same deterministic tool call — - // for subagent starts that collides on the derived operation id and - // fatally fails the parent session. - if (isAgentsAnnouncementText(getPromptContentText(message.content).trim())) { - continue; - } return null; } diff --git a/packages/eve/src/runtime/agent/mock-model-fixtures.ts b/packages/eve/src/runtime/agent/mock-model-fixtures.ts index ec8378abd..13f4829e2 100644 --- a/packages/eve/src/runtime/agent/mock-model-fixtures.ts +++ b/packages/eve/src/runtime/agent/mock-model-fixtures.ts @@ -1,8 +1,5 @@ import type { BootstrapPrompt } from "#runtime/agent/bootstrap-model-utils.js"; -import { - getPromptContentText, - isAgentsAnnouncementText, -} from "#runtime/agent/bootstrap-model-utils.js"; +import { getPromptContentText } from "#runtime/agent/bootstrap-model-utils.js"; import { createJsonSchemaSample } from "#runtime/agent/mock-structured-output.js"; import { LOAD_SKILL_TOOL_NAME } from "#runtime/skills/fragment-context.js"; @@ -141,11 +138,7 @@ function getTrailingUserText(prompt: BootstrapPrompt): string { for (const message of [...prompt].reverse()) { if (message.role === "system") continue; if (message.role !== "user") break; - const text = getPromptContentText(message.content); - // Framework-injected [Agents] announcements are scaffolding, not part - // of the turn's authored ask. - if (isAgentsAnnouncementText(text.trim())) continue; - texts.unshift(text); + texts.unshift(getPromptContentText(message.content)); } return texts.join("\n"); diff --git a/packages/eve/src/runtime/prompt/compose.ts b/packages/eve/src/runtime/prompt/compose.ts index 2b2327b53..b9233018c 100644 --- a/packages/eve/src/runtime/prompt/compose.ts +++ b/packages/eve/src/runtime/prompt/compose.ts @@ -12,7 +12,7 @@ const PARALLEL_ACTION_INSTRUCTION = "Tool execution\nA single tool or subagent call runs as one serial action. If you call multiple independent tools or subagents in one response, eve treats that batch as parallel work. Only batch work that is independent and does not rely on another call in the same response."; const AGENT_MESSAGING_INSTRUCTION = - "Agent messaging\nAgents you have already delegated to stay available after they answer. eve injects the current `` list into the conversation as a note labeled `[Agents]`; it is added automatically by the framework, not written by the user, and never requires a reply. The list is only a record of those existing agents — their `agentId`, name, and latest status. It does not limit which subagent tools you can call: your tool list is the source of truth, and any subagent tool can always be called without `agentId` to start a new agent, including when the `` list is empty or absent. Pass `agentId` to the same subagent tool only to continue one of those existing agents' sessions."; + "Agent messaging\nAgents you have already delegated to stay available after they answer. The system-message `` list is only a record of those existing agents — their `agentId`, name, and latest status. It does not limit which subagent tools you can call: your tool list is the source of truth, and any subagent tool can always be called without `agentId` to start a new agent, including when the `` list is empty or absent. Pass `agentId` to the same subagent tool only to continue one of those existing agents' sessions."; /** * Input for composing the base authored instructions prompt for one