From 2dd4f3976a1af37f8b8c81f1156efeb96d84641d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Tue, 9 Jun 2026 21:18:51 -0400 Subject: [PATCH] Enforce deadlines across the whole provider request lifecycle The abort timer was cleared as soon as fetch resolved with headers, so a server that stalled the response body hung the request forever - and one hung run froze the autonomous queue behind the worker re-entrancy guard. JSON requests now carry a header deadline plus a longer body deadline on one abort controller, and event streams fail when no chunk arrives within an idle window, cancelling the reader. Timeouts surface as ProviderHttpError so the run records a clear failure and the heartbeat retries. --- .../src/main/services/provider-http.ts | 81 +++++++++++++++---- .../tests/unit/provider-http.test.ts | 64 +++++++++++++++ 2 files changed, 128 insertions(+), 17 deletions(-) create mode 100644 AgentCompany-New/tests/unit/provider-http.test.ts diff --git a/AgentCompany-New/src/main/services/provider-http.ts b/AgentCompany-New/src/main/services/provider-http.ts index 06fb01f2..d0f3c314 100644 --- a/AgentCompany-New/src/main/services/provider-http.ts +++ b/AgentCompany-New/src/main/services/provider-http.ts @@ -19,7 +19,15 @@ export interface ProviderEventStreamResponse { rateLimitState: ProviderObservedRateLimitState | null; } -const PROVIDER_REQUEST_TIMEOUT_MS = 60_000; +const HEADERS_TIMEOUT_MS = 60_000; +const BODY_TIMEOUT_MS = 300_000; +const STREAM_IDLE_TIMEOUT_MS = 60_000; + +export interface ProviderRequestTimeouts { + headersTimeoutMs?: number; + bodyTimeoutMs?: number; + idleTimeoutMs?: number; +} export class ProviderHttpError extends Error { constructor( @@ -35,7 +43,12 @@ export function providerRateLimitStateFromError(error: unknown): ProviderObserve return error instanceof ProviderHttpError ? error.rateLimitState : null; } -export async function postJson(endpoint: string, headers: Record, body: JsonRecord): Promise { +export async function postJson( + endpoint: string, + headers: Record, + body: JsonRecord, + timeouts: ProviderRequestTimeouts = {}, +): Promise { return requestJson(endpoint, { method: "POST", headers: { @@ -43,7 +56,7 @@ export async function postJson(endpoint: string, headers: Record ...headers, }, body: JSON.stringify(body), - }); + }, timeouts); } export async function getJson(endpoint: string, headers: Record): Promise { @@ -54,8 +67,10 @@ export async function postEventStream( endpoint: string, headers: Record, body: JsonRecord, + timeouts: ProviderRequestTimeouts = {}, ): Promise { - const response = await requestProvider(endpoint, { + const controller = new AbortController(); + const response = await withDeadline(controller, timeouts.headersTimeoutMs ?? HEADERS_TIMEOUT_MS, () => fetch(endpoint, { method: "POST", headers: { accept: "text/event-stream", @@ -63,22 +78,24 @@ export async function postEventStream( ...headers, }, body: JSON.stringify(body), - }); + signal: controller.signal, + })); const rateLimitState = parseProviderRateLimitHeaders(response.headers); if (!response.ok) { - const raw = await response.text(); + const raw = await withDeadline(controller, timeouts.bodyTimeoutMs ?? BODY_TIMEOUT_MS, () => response.text()); const data = parseProviderJson(raw, false); const message = readProviderError(data) ?? (raw.trim() || `Provider request failed with HTTP ${response.status}.`); throw new ProviderHttpError(message, rateLimitState); } if (!response.body) throw new ProviderHttpError("Provider returned an empty event stream.", rateLimitState); - return { messages: readServerSentEvents(response.body), rateLimitState }; + return { messages: readServerSentEvents(response.body, timeouts.idleTimeoutMs ?? STREAM_IDLE_TIMEOUT_MS), rateLimitState }; } -async function requestJson(endpoint: string, init: RequestInit): Promise { - const response = await requestProvider(endpoint, init); +async function requestJson(endpoint: string, init: RequestInit, timeouts: ProviderRequestTimeouts = {}): Promise { + const controller = new AbortController(); + const response = await withDeadline(controller, timeouts.headersTimeoutMs ?? HEADERS_TIMEOUT_MS, () => fetch(endpoint, { ...init, signal: controller.signal })); const rateLimitState = parseProviderRateLimitHeaders(response.headers); - const raw = await response.text(); + const raw = await withDeadline(controller, timeouts.bodyTimeoutMs ?? BODY_TIMEOUT_MS, () => response.text()); const data = parseProviderJson(raw, response.ok); if (!response.ok) { const message = readProviderError(data) ?? (raw.trim() || `Provider request failed with HTTP ${response.status}.`); @@ -87,24 +104,35 @@ async function requestJson(endpoint: string, init: RequestInit): Promise { - const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), PROVIDER_REQUEST_TIMEOUT_MS); +/** + * Runs one phase of a provider request under a hard deadline. The abort + * controller covers the WHOLE request — a server that returns headers and + * then stalls the body must not hang the caller, so the body read gets its + * own (longer) deadline after the headers arrive. + */ +async function withDeadline(controller: AbortController, timeoutMs: number, operation: () => Promise): Promise { + const timer = setTimeout(() => controller.abort(), timeoutMs); try { - return await fetch(endpoint, { ...init, signal: controller.signal }); + return await operation(); + } catch (error) { + if (controller.signal.aborted) throw new ProviderHttpError("Provider request timed out.", null); + throw error; } finally { - clearTimeout(timeout); + clearTimeout(timer); } } -export async function* readServerSentEvents(body: ReadableStream): AsyncIterable { +export async function* readServerSentEvents( + body: ReadableStream, + idleTimeoutMs: number = STREAM_IDLE_TIMEOUT_MS, +): AsyncIterable { const decoder = new TextDecoder(); const state: EventStreamParseState = { eventName: null, dataLines: [] }; let lineBuffer = ""; const reader = body.getReader(); try { for (;;) { - const { value, done } = await reader.read(); + const { value, done } = await readWithIdleDeadline(reader, idleTimeoutMs); if (done) break; lineBuffer += decoder.decode(value, { stream: true }); const lines = lineBuffer.split("\n"); @@ -121,11 +149,30 @@ export async function* readServerSentEvents(body: ReadableStream): A } const message = flushServerSentEvent(state); if (message) yield message; + } catch (error) { + await reader.cancel().catch(() => undefined); + throw error; } finally { reader.releaseLock(); } } +/** A healthy event stream emits chunks continuously; a read that produces nothing within the idle window means the connection is dead. */ +async function readWithIdleDeadline( + reader: ReadableStreamDefaultReader, + idleTimeoutMs: number, +): Promise> { + let timer: NodeJS.Timeout | undefined; + const idle = new Promise((_, reject) => { + timer = setTimeout(() => reject(new ProviderHttpError("Provider stream timed out.", null)), idleTimeoutMs); + }); + try { + return await Promise.race([reader.read(), idle]); + } finally { + clearTimeout(timer); + } +} + function parseServerSentEventLine(state: EventStreamParseState, rawLine: string): ProviderEventStreamMessage | null { const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; if (line === "") return flushServerSentEvent(state); diff --git a/AgentCompany-New/tests/unit/provider-http.test.ts b/AgentCompany-New/tests/unit/provider-http.test.ts new file mode 100644 index 00000000..7d14aafe --- /dev/null +++ b/AgentCompany-New/tests/unit/provider-http.test.ts @@ -0,0 +1,64 @@ +import { createServer, type Server } from "node:http"; +import { afterEach, describe, expect, test } from "vitest"; +import { postEventStream, postJson, ProviderHttpError } from "../../src/main/services/provider-http"; + +let server: Server | null = null; + +function listen(handler: Parameters[1]): Promise { + return new Promise((resolve) => { + server = createServer(handler); + server.listen(0, "127.0.0.1", () => { + const address = server!.address() as { port: number }; + resolve(`http://127.0.0.1:${address.port}`); + }); + }); +} + +afterEach(() => { + server?.closeAllConnections(); + server?.close(); + server = null; +}); + +describe("provider http deadlines", () => { + test("a body that stalls after headers fails instead of hanging", async () => { + const baseUrl = await listen((_request, response) => { + response.writeHead(200, { "content-type": "application/json" }); + response.write('{"partial":'); + // Never ends the body. + }); + await expect(postJson(baseUrl, {}, {}, { bodyTimeoutMs: 100 })) + .rejects.toThrow(/timed out/); + }); + + test("a server that never sends headers fails at the header deadline", async () => { + const baseUrl = await listen(() => { + // Accept the request and never respond. + }); + await expect(postJson(baseUrl, {}, {}, { headersTimeoutMs: 100 })) + .rejects.toThrow(/timed out/); + }); + + test("an event stream that goes idle fails instead of hanging", async () => { + const baseUrl = await listen((_request, response) => { + response.writeHead(200, { "content-type": "text/event-stream" }); + response.write('data: {"first":true}\n\n'); + // Never sends another event. + }); + const stream = await postEventStream(baseUrl, {}, {}, { idleTimeoutMs: 100 }); + const seen: unknown[] = []; + await expect((async () => { + for await (const message of stream.messages) seen.push(message.data); + })()).rejects.toThrow(ProviderHttpError); + expect(seen).toEqual([{ first: true }]); + }); + + test("a healthy response is returned unchanged", async () => { + const baseUrl = await listen((_request, response) => { + response.writeHead(200, { "content-type": "application/json" }); + response.end('{"ok":true}'); + }); + const result = await postJson(baseUrl, {}, {}); + expect(result.data).toEqual({ ok: true }); + }); +});