From 91d17fd268ff9b612569cea5acfc67af041eb375 Mon Sep 17 00:00:00 2001 From: Jason Kneen Date: Tue, 22 Sep 2026 20:26:49 +0100 Subject: [PATCH] fix(providers): content-idle stream watchdog that keepalive pings cannot reset Invariant: stream liveness timers reset on content events only, never on keepalive pings. Cause: the only liveness guard was undici bodyTimeout (httpIdleTimeoutMs, default 300s), which resets on any body bytes including SSE pings and comment keepalives, so a stalled generation that keeps pinging never timed out. Library users of providers/agent-core had no idle guard at all. Fix: add StreamOptions.streamIdleTimeoutMs and withStreamIdleTimeout, applied at the stream-dispatch entry points (createProvider, Models, compat fallback, coding-agent ModelRuntime). When > 0 it runs the request under a child AbortController linked to the caller signal, resets on every emitted stream event, and on expiry aborts the request and terminates with stopReason "error" and "Stream idle timeout: no content for Nms (phase: ..., last content at )", which isRetryableAssistantError treats as retryable. Caller aborts still yield "aborted". coding-agent passes the existing httpIdleTimeoutMs setting (0 disables). --- .../coding-agent/src/core/model-runtime.ts | 35 +-- packages/coding-agent/src/core/sdk.ts | 3 + .../test/sdk-stream-options.test.ts | 15 ++ packages/providers/src/compat.ts | 9 +- packages/providers/src/index.ts | 1 + packages/providers/src/models.ts | 40 ++-- packages/providers/src/types.ts | 7 + .../src/utils/stream-idle-timeout.ts | 153 +++++++++++++ .../test/stream-idle-timeout.test.ts | 201 ++++++++++++++++++ 9 files changed, 432 insertions(+), 32 deletions(-) create mode 100644 packages/providers/src/utils/stream-idle-timeout.ts create mode 100644 packages/providers/test/stream-idle-timeout.test.ts diff --git a/packages/coding-agent/src/core/model-runtime.ts b/packages/coding-agent/src/core/model-runtime.ts index ca5d04b9..f42a171d 100644 --- a/packages/coding-agent/src/core/model-runtime.ts +++ b/packages/coding-agent/src/core/model-runtime.ts @@ -35,6 +35,7 @@ import { type ProviderRequestOptions, type SimpleStreamOptions, type StreamOptions, + withStreamIdleTimeout, } from "@step-harness/providers"; import * as builtinProviderCatalog from "@step-harness/providers/providers/all"; import { getAgentDir, STEP_ENTRYPOINT } from "../config.ts"; @@ -587,17 +588,19 @@ export class ModelRuntime implements Models { context: Context, options?: ModelsApiStreamOptions, ): AssistantMessageEventStream { - return lazyStream(model, async () => { - const prepared = await this.prepareRequest( - model, - options as (StreamOptions & ModelsRequestTransforms) | undefined, - ); - return prepared.provider.stream( - prepared.model as Model, - context, - prepared.options as ApiStreamOptions, - ); - }); + return withStreamIdleTimeout(model, options, (guardedOptions) => + lazyStream(model, async () => { + const prepared = await this.prepareRequest( + model, + guardedOptions as (StreamOptions & ModelsRequestTransforms) | undefined, + ); + return prepared.provider.stream( + prepared.model as Model, + context, + prepared.options as ApiStreamOptions, + ); + }), + ); } complete( @@ -609,10 +612,12 @@ export class ModelRuntime implements Models { } streamSimple(model: Model, context: Context, options?: ModelsSimpleStreamOptions): AssistantMessageEventStream { - return lazyStream(model, async () => { - const prepared = await this.prepareRequest(model, options); - return prepared.provider.streamSimple(prepared.model, context, prepared.options as SimpleStreamOptions); - }); + return withStreamIdleTimeout(model, options, (guardedOptions) => + lazyStream(model, async () => { + const prepared = await this.prepareRequest(model, guardedOptions); + return prepared.provider.streamSimple(prepared.model, context, prepared.options as SimpleStreamOptions); + }), + ); } completeSimple(model: Model, context: Context, options?: ModelsSimpleStreamOptions): Promise { diff --git a/packages/coding-agent/src/core/sdk.ts b/packages/coding-agent/src/core/sdk.ts index f07d5a34..82dcd346 100644 --- a/packages/coding-agent/src/core/sdk.ts +++ b/packages/coding-agent/src/core/sdk.ts @@ -353,6 +353,9 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} const requestOptions: ModelsSimpleStreamOptions = { ...options, timeoutMs, + // Content-idle watchdog: same budget as the transport idle timeout, but + // keepalive pings do not reset it (0 disables). + streamIdleTimeoutMs: options?.streamIdleTimeoutMs ?? httpIdleTimeoutMs, websocketConnectTimeoutMs, maxRetries: options?.maxRetries ?? providerRetrySettings.maxRetries, maxRetryDelayMs: options?.maxRetryDelayMs ?? providerRetrySettings.maxRetryDelayMs, diff --git a/packages/coding-agent/test/sdk-stream-options.test.ts b/packages/coding-agent/test/sdk-stream-options.test.ts index 5045d313..b701f2e4 100644 --- a/packages/coding-agent/test/sdk-stream-options.test.ts +++ b/packages/coding-agent/test/sdk-stream-options.test.ts @@ -143,6 +143,21 @@ describe("createAgentSession stream options", () => { expect(options?.timeoutMs).toBe(0); }); + it("guards the stream with a content-idle watchdog from httpIdleTimeoutMs", async () => { + const options = await captureStreamOptions("openai-completions", { httpIdleTimeoutMs: 1234 }); + + // The watchdog consumes the option and links its own abort signal to the request. + expect(options).not.toHaveProperty("streamIdleTimeoutMs"); + expect(options?.signal).toBeInstanceOf(AbortSignal); + }); + + it("installs no content-idle watchdog when httpIdleTimeoutMs is 0", async () => { + const options = await captureStreamOptions("openai-completions", { httpIdleTimeoutMs: 0 }); + + expect(options?.streamIdleTimeoutMs).toBe(0); + expect(options?.signal).toBeUndefined(); + }); + it("forwards websocketConnectTimeoutMs from settings", async () => { const options = await captureStreamOptions("openai-responses", { websocketConnectTimeoutMs: 1234 }); diff --git a/packages/providers/src/compat.ts b/packages/providers/src/compat.ts index 3fd4e98d..3f38887d 100644 --- a/packages/providers/src/compat.ts +++ b/packages/providers/src/compat.ts @@ -40,6 +40,7 @@ import type { StreamFunction, StreamOptions, } from "./types.ts"; +import { withStreamIdleTimeout } from "./utils/stream-idle-timeout.ts"; /** @deprecated Static catalog read. Use `getBuiltinModel` from "@step-harness/providers/providers/all" or `Models.getModel()`. */ export const getModel = getBuiltinModel; @@ -235,7 +236,9 @@ export function stream( return builtinProvider.stream(model, context, withEnvApiKey(model, options) as ApiStreamOptions); } const provider = resolveApiProvider(model.api); - return provider.stream(model, context, withEnvApiKey(model, options) as StreamOptions); + return withStreamIdleTimeout(model, withEnvApiKey(model, options) as StreamOptions | undefined, (guardedOptions) => + provider.stream(model, context, guardedOptions), + ); } export async function complete( @@ -260,7 +263,9 @@ export function streamSimple( return builtinProvider.streamSimple(model, context, withEnvApiKey(model, options)); } const provider = resolveApiProvider(model.api); - return provider.streamSimple(model, context, withEnvApiKey(model, options)); + return withStreamIdleTimeout(model, withEnvApiKey(model, options), (guardedOptions) => + provider.streamSimple(model, context, guardedOptions), + ); } export async function completeSimple( diff --git a/packages/providers/src/index.ts b/packages/providers/src/index.ts index ba3d48b2..2015bf1a 100644 --- a/packages/providers/src/index.ts +++ b/packages/providers/src/index.ts @@ -42,6 +42,7 @@ export * from "./utils/event-stream.ts"; export * from "./utils/json-parse.ts"; export * from "./utils/overflow.ts"; export * from "./utils/retry.ts"; +export * from "./utils/stream-idle-timeout.ts"; export { contentText } from "./utils/text.ts"; export * from "./utils/typebox-helpers.ts"; export { uuidv7 } from "./utils/uuid.ts"; diff --git a/packages/providers/src/models.ts b/packages/providers/src/models.ts index dfc5202c..e08b3ce5 100644 --- a/packages/providers/src/models.ts +++ b/packages/providers/src/models.ts @@ -33,6 +33,7 @@ import type { Usage, } from "./types.ts"; import { operationSignal, raceWithAbortSignal } from "./utils/abort.ts"; +import { withStreamIdleTimeout } from "./utils/stream-idle-timeout.ts"; export { ModelsError, type ModelsErrorCode } from "./auth/resolve.ts"; @@ -669,14 +670,16 @@ class ModelsImpl implements MutableModels { context: Context, options?: ModelsApiStreamOptions, ): AssistantMessageEventStream { - return lazyStream(model, async () => { - const provider = this.requireProvider(model); - const { requestModel, requestOptions } = await this.applyAuth( - model, - options as ModelsApiStreamOptions | undefined, - ); - return provider.stream(requestModel as Model, context, requestOptions as ApiStreamOptions); - }); + return withStreamIdleTimeout(model, options, (guardedOptions) => + lazyStream(model, async () => { + const provider = this.requireProvider(model); + const { requestModel, requestOptions } = await this.applyAuth( + model, + guardedOptions as ModelsApiStreamOptions | undefined, + ); + return provider.stream(requestModel as Model, context, requestOptions as ApiStreamOptions); + }), + ); } async complete( @@ -688,11 +691,13 @@ class ModelsImpl implements MutableModels { } streamSimple(model: Model, context: Context, options?: ModelsSimpleStreamOptions): AssistantMessageEventStream { - return lazyStream(model, async () => { - const provider = this.requireProvider(model); - const { requestModel, requestOptions } = await this.applyAuth(model, options); - return provider.streamSimple(requestModel, context, requestOptions as SimpleStreamOptions); - }); + return withStreamIdleTimeout(model, options, (guardedOptions) => + lazyStream(model, async () => { + const provider = this.requireProvider(model); + const { requestModel, requestOptions } = await this.applyAuth(model, guardedOptions); + return provider.streamSimple(requestModel, context, requestOptions as SimpleStreamOptions); + }), + ); } async completeSimple( @@ -826,9 +831,14 @@ export function createProvider(input: CreateProviderOpti } : undefined, filterModels: input.filterModels, - stream: (model, context, options) => dispatch(model, (streams) => streams.stream(model, context, options)), + stream: (model, context, options) => + withStreamIdleTimeout(model, options, (guardedOptions) => + dispatch(model, (streams) => streams.stream(model, context, guardedOptions)), + ), streamSimple: (model, context, options) => - dispatch(model, (streams) => streams.streamSimple(model, context, options)), + withStreamIdleTimeout(model, options, (guardedOptions) => + dispatch(model, (streams) => streams.streamSimple(model, context, guardedOptions)), + ), }; const streams = single ? [single] : Object.values(byApi ?? {}).filter((entry) => entry !== undefined); diff --git a/packages/providers/src/types.ts b/packages/providers/src/types.ts index 7fa967e4..fa2c1ebf 100644 --- a/packages/providers/src/types.ts +++ b/packages/providers/src/types.ts @@ -188,6 +188,13 @@ export interface StreamOptions extends ProviderRequestOptions> { * stream idleness after connection uses timeoutMs. */ websocketConnectTimeoutMs?: number; + /** + * Content-idle watchdog in milliseconds. When > 0, the stream fails with + * `stopReason: "error"` if no stream event (start, deltas, block start/end) + * is emitted for this long; keepalive pings do not reset it. The request is + * aborted on expiry. 0 or undefined disables the watchdog. + */ + streamIdleTimeoutMs?: number; /** * Optional metadata to include in API requests. * Providers extract the fields they understand and ignore the rest. diff --git a/packages/providers/src/utils/stream-idle-timeout.ts b/packages/providers/src/utils/stream-idle-timeout.ts new file mode 100644 index 00000000..8e171b32 --- /dev/null +++ b/packages/providers/src/utils/stream-idle-timeout.ts @@ -0,0 +1,153 @@ +import type { Api, AssistantMessage, AssistantMessageEvent, Model, StreamOptions } from "../types.ts"; +import { AssistantMessageEventStream } from "./event-stream.ts"; + +function isTerminal(event: AssistantMessageEvent): event is Extract { + return event.type === "done" || event.type === "error"; +} + +function createIdleErrorMessage( + model: Model, + partial: AssistantMessage | undefined, + errorMessage: string, +): AssistantMessage { + return { + ...(partial ?? { + role: "assistant", + content: [], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + timestamp: Date.now(), + }), + stopReason: "error", + errorMessage, + }; +} + +/** + * Guards a provider stream with a content-idle watchdog when + * `options.streamIdleTimeoutMs` is > 0. The timer starts when the request is + * issued and resets on every emitted stream event. Adapters only emit content + * events (SSE pings/comment keepalives are filtered below this layer), so a + * stalled generation that keeps pinging still times out, unlike transport body + * timeouts that reset on any byte. + * + * On expiry the request is aborted and the stream terminates with + * `stopReason: "error"` (not "aborted": the caller did not abort) and a message + * naming the phase and last activity. The message matches + * `isRetryableAssistantError`. Caller aborts are forwarded unchanged. + * + * The option is consumed here, so nested dispatch layers do not re-guard. + */ +export function withStreamIdleTimeout( + model: Model, + options: TOptions | undefined, + start: (options: TOptions | undefined) => AssistantMessageEventStream, +): AssistantMessageEventStream { + const timeoutMs = options?.streamIdleTimeoutMs; + if (!options || timeoutMs === undefined || !Number.isFinite(timeoutMs) || timeoutMs <= 0) { + return start(options); + } + + const { streamIdleTimeoutMs: _consumed, ...rest } = options; + const callerSignal = options.signal; + const controller = new AbortController(); + const outer = new AssistantMessageEventStream(); + const requestStartedAt = Date.now(); + let timer: ReturnType | undefined; + let finished = false; + let lastPartial: AssistantMessage | undefined; + let lastEventType: AssistantMessageEvent["type"] | undefined; + let lastEventAt = requestStartedAt; + + const onCallerAbort = () => { + clearTimeout(timer); + controller.abort(callerSignal?.reason); + }; + const finish = () => { + finished = true; + clearTimeout(timer); + callerSignal?.removeEventListener("abort", onCallerAbort); + }; + const onIdle = () => { + if (finished || callerSignal?.aborted) return; + const phase = + lastEventType === undefined || lastEventType === "start" + ? "waiting for first token" + : `streaming ${lastEventType}`; + const activity = + lastEventType === undefined + ? `request started at ${new Date(requestStartedAt).toISOString()}` + : `last content at ${new Date(lastEventAt).toISOString()}`; + const errorMessage = `Stream idle timeout: no content for ${timeoutMs}ms (phase: ${phase}, ${activity})`; + const error = createIdleErrorMessage(model, lastPartial, errorMessage); + finish(); + controller.abort(new Error(errorMessage)); + outer.push({ type: "error", reason: "error", error }); + outer.end(error); + }; + const arm = () => { + clearTimeout(timer); + if (finished || callerSignal?.aborted) return; + timer = setTimeout(onIdle, timeoutMs); + }; + + if (callerSignal?.aborted) controller.abort(callerSignal.reason); + else callerSignal?.addEventListener("abort", onCallerAbort, { once: true }); + + arm(); + let inner: AssistantMessageEventStream; + try { + inner = start({ ...rest, signal: controller.signal } as TOptions); + } catch (error) { + finish(); + throw error; + } + + void (async () => { + let terminal: AssistantMessage | undefined; + try { + for await (const event of inner) { + if (finished) return; + if (isTerminal(event)) { + terminal = event.type === "done" ? event.message : event.error; + finish(); + } else { + lastPartial = event.partial; + lastEventType = event.type; + lastEventAt = Date.now(); + arm(); + } + outer.push(event); + } + } catch (error) { + if (finished) return; + finish(); + const message = createIdleErrorMessage( + model, + lastPartial, + error instanceof Error ? error.message : String(error), + ); + outer.push({ type: "error", reason: "error", error: message }); + outer.end(message); + return; + } + if (terminal) { + outer.end(terminal); + return; + } + if (finished) return; + finish(); + outer.end(await inner.result()); + })(); + + return outer; +} diff --git a/packages/providers/test/stream-idle-timeout.test.ts b/packages/providers/test/stream-idle-timeout.test.ts new file mode 100644 index 00000000..77f614ac --- /dev/null +++ b/packages/providers/test/stream-idle-timeout.test.ts @@ -0,0 +1,201 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createProvider } from "../src/models.ts"; +import type { Api, AssistantMessage, Context, Model, StreamOptions } from "../src/types.ts"; +import { AssistantMessageEventStream } from "../src/utils/event-stream.ts"; +import { isRetryableAssistantError } from "../src/utils/retry.ts"; + +const model: Model = { + id: "idle-model", + name: "idle-model", + api: "idle-api", + provider: "idle-provider", + baseUrl: "https://example.test/v1", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 10000, + maxTokens: 1000, +}; + +const context: Context = { messages: [{ role: "user", content: "hi", timestamp: 0 }] }; + +function message(stopReason: AssistantMessage["stopReason"], text = ""): AssistantMessage { + return { + role: "assistant", + content: text ? [{ type: "text", text }] : [], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason, + timestamp: Date.now(), + }; +} + +/** + * A wire API that behaves like an adapter whose upstream only sends keepalive + * pings: pings are filtered below the event layer, so after `start` (and any + * scripted deltas) nothing is emitted until the request signal aborts. + */ +function scriptedProvider(script: (stream: AssistantMessageEventStream, signal: AbortSignal | undefined) => void) { + const seen: { signal?: AbortSignal; streamIdleTimeoutMs?: number }[] = []; + const provider = createProvider({ + id: model.provider, + auth: { apiKey: { name: "key", resolve: async () => ({ auth: { apiKey: "k" } }) } } as never, + models: [model], + api: { + stream: (_model, _context, options?: StreamOptions) => { + seen.push({ signal: options?.signal, streamIdleTimeoutMs: options?.streamIdleTimeoutMs }); + const stream = new AssistantMessageEventStream(); + const signal = options?.signal; + signal?.addEventListener("abort", () => { + const aborted = message("aborted"); + stream.push({ type: "error", reason: "aborted", error: aborted }); + stream.end(aborted); + }); + script(stream, signal); + return stream; + }, + streamSimple: () => { + throw new Error("unused"); + }, + }, + }); + return { provider, seen }; +} + +describe("stream content-idle watchdog", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it("fails with stopReason error naming the phase when only keepalives follow start", async () => { + const { provider } = scriptedProvider((stream) => { + stream.push({ type: "start", partial: message("stop") }); + }); + const stream = provider.stream(model, context, { streamIdleTimeoutMs: 1000 }); + const events: string[] = []; + const consume = (async () => { + for await (const event of stream) events.push(event.type); + })(); + + await vi.advanceTimersByTimeAsync(999); + expect(events).toEqual(["start"]); + await vi.advanceTimersByTimeAsync(1); + await consume; + + const result = await stream.result(); + expect(events).toEqual(["start", "error"]); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toBe( + "Stream idle timeout: no content for 1000ms (phase: waiting for first token, last content at 2026-01-01T00:00:00.000Z)", + ); + expect(isRetryableAssistantError(result)).toBe(true); + }); + + it("names the last streamed event and keeps partial content when a stream stalls mid-generation", async () => { + const { provider } = scriptedProvider((stream) => { + const partial = message("stop", "hel"); + stream.push({ type: "start", partial: message("stop") }); + setTimeout(() => stream.push({ type: "text_start", contentIndex: 0, partial }), 500); + setTimeout(() => stream.push({ type: "text_delta", contentIndex: 0, delta: "hel", partial }), 900); + }); + const stream = provider.stream(model, context, { streamIdleTimeoutMs: 1000 }); + + await vi.advanceTimersByTimeAsync(1899); + let settled = false; + void stream.result().then(() => { + settled = true; + }); + await vi.advanceTimersByTimeAsync(0); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.content).toEqual([{ type: "text", text: "hel" }]); + expect(result.errorMessage).toBe( + "Stream idle timeout: no content for 1000ms (phase: streaming text_delta, last content at 2026-01-01T00:00:00.900Z)", + ); + }); + + it("aborts the underlying request when the watchdog fires", async () => { + const { provider, seen } = scriptedProvider(() => {}); + const stream = provider.stream(model, context, { streamIdleTimeoutMs: 1000 }); + await vi.advanceTimersByTimeAsync(1000); + + const result = await stream.result(); + expect(result.stopReason).toBe("error"); + expect(result.errorMessage).toBe( + "Stream idle timeout: no content for 1000ms (phase: waiting for first token, request started at 2026-01-01T00:00:00.000Z)", + ); + expect(seen[0].signal?.aborted).toBe(true); + // The option is consumed by the outermost guard so nested dispatch layers do not double-guard. + expect(seen[0].streamIdleTimeoutMs).toBeUndefined(); + }); + + it("never trips while content arrives more often than the timeout", async () => { + const { provider } = scriptedProvider((stream) => { + const partial = message("stop", "x"); + stream.push({ type: "start", partial: message("stop") }); + stream.push({ type: "text_start", contentIndex: 0, partial }); + for (let i = 1; i <= 10; i++) { + setTimeout(() => stream.push({ type: "text_delta", contentIndex: 0, delta: "x", partial }), i * 800); + } + setTimeout(() => { + const done = message("stop", "x"); + stream.push({ type: "done", reason: "stop", message: done }); + stream.end(done); + }, 8500); + }); + const stream = provider.stream(model, context, { streamIdleTimeoutMs: 1000 }); + + await vi.advanceTimersByTimeAsync(8500); + const result = await stream.result(); + expect(result.stopReason).toBe("stop"); + expect(vi.getTimerCount()).toBe(0); + }); + + it("still reports a caller abort as aborted", async () => { + const { provider } = scriptedProvider((stream) => { + stream.push({ type: "start", partial: message("stop") }); + }); + const controller = new AbortController(); + const stream = provider.stream(model, context, { streamIdleTimeoutMs: 1000, signal: controller.signal }); + + await vi.advanceTimersByTimeAsync(500); + controller.abort(); + await vi.advanceTimersByTimeAsync(5000); + + const result = await stream.result(); + expect(result.stopReason).toBe("aborted"); + expect(vi.getTimerCount()).toBe(0); + }); + + it.each([0, undefined])("installs no watchdog when streamIdleTimeoutMs is %s", async (streamIdleTimeoutMs) => { + const { provider, seen } = scriptedProvider((stream) => { + stream.push({ type: "start", partial: message("stop") }); + }); + const stream = provider.stream(model, context, { streamIdleTimeoutMs }); + let settled = false; + void stream.result().then(() => { + settled = true; + }); + + await vi.advanceTimersByTimeAsync(10 * 60_000); + expect(settled).toBe(false); + expect(vi.getTimerCount()).toBe(0); + expect(seen[0].signal).toBeUndefined(); + }); +});