fix(providers): content-idle stream watchdog that keepalive pings cannot reset

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<TApi>,
 	): AssistantMessageEventStream {
-		return lazyStream(model, async () => {
-			const prepared = await this.prepareRequest(
-				model,
-				options as (StreamOptions & ModelsRequestTransforms) | undefined,
-			);
-			return prepared.provider.stream(
-				prepared.model as Model<TApi>,
-				context,
-				prepared.options as ApiStreamOptions<TApi>,
-			);
-		});
+		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<TApi>,
+					context,
+					prepared.options as ApiStreamOptions<TApi>,
+				);
+			}),
+		);
 	}
 
 	complete<TApi extends Api>(
@@ -609,10 +612,12 @@ export class ModelRuntime implements Models {
 	}
 
 	streamSimple(model: Model<Api>, 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<Api>, context: Context, options?: ModelsSimpleStreamOptions): Promise<AssistantMessage> {

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,

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 });
 

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<TApi extends Api>(
 		return builtinProvider.stream(model, context, withEnvApiKey(model, options) as ApiStreamOptions<TApi>);
 	}
 	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<TApi extends Api>(
@@ -260,7 +263,9 @@ export function streamSimple<TApi extends Api>(
 		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<TApi extends Api>(

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";

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<TApi>,
 	): AssistantMessageEventStream {
-		return lazyStream(model, async () => {
-			const provider = this.requireProvider(model);
-			const { requestModel, requestOptions } = await this.applyAuth(
-				model,
-				options as ModelsApiStreamOptions<Api> | undefined,
-			);
-			return provider.stream(requestModel as Model<TApi>, context, requestOptions as ApiStreamOptions<TApi>);
-		});
+		return withStreamIdleTimeout(model, options, (guardedOptions) =>
+			lazyStream(model, async () => {
+				const provider = this.requireProvider(model);
+				const { requestModel, requestOptions } = await this.applyAuth(
+					model,
+					guardedOptions as ModelsApiStreamOptions<Api> | undefined,
+				);
+				return provider.stream(requestModel as Model<TApi>, context, requestOptions as ApiStreamOptions<TApi>);
+			}),
+		);
 	}
 
 	async complete<TApi extends Api>(
@@ -688,11 +691,13 @@ class ModelsImpl implements MutableModels {
 	}
 
 	streamSimple(model: Model<Api>, 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<TApi extends Api = Api>(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);

packages/providers/src/types.ts

@@ -188,6 +188,13 @@ export interface StreamOptions extends ProviderRequestOptions<Model<Api>> {
 	 * 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.

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<AssistantMessageEvent, { type: "done" | "error" }> {
+	return event.type === "done" || event.type === "error";
+}
+
+function createIdleErrorMessage(
+	model: Model<Api>,
+	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<TOptions extends StreamOptions>(
+	model: Model<Api>,
+	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<typeof setTimeout> | 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;
+}