From bb42032739e54411a4d2cb2694127c699b292766 Mon Sep 17 00:00:00 2001 From: vastsa Date: Wed, 16 Sep 2026 22:12:26 +0800 Subject: [PATCH 1/2] fix(agent-runtime): rebuild the shared provider transport after unanswered failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #234: one Codex turn retried ten times and every attempt reported a bare `NETWORK_ERROR: fetch failed` while a new turn on the same provider, model, and network recovered immediately. The recorded fingerprint (`phase=stream`, `streamMs=1~2`) reads like a failure after the response headers, but pi-agent-core emits `message_start` for a stream that ended without ever emitting `start`, so it is the opposite: all ten attempts died before a response, and each spent seconds in the `phase=request` retries. Two provider-side gaps follow from that: - The real cause never reached the log. `classifyAgentError` walks the cause chain only when it is handed the Error, and pi-ai flattens a rejected request into `errorMessage` first, so `fetch failed` collapsed to `networkCategory: unknown` with no errno. The fetch wrapper in `provider-retry.ts` is the last layer that still holds the object, so it describes the cause there and merges the same validated fields the classifier produces (`networkCategory`, `networkCode`, `networkSyscall`, `networkHost`) plus `networkRoute` into the retried and terminal errors. - The shared transport was never invalidated. `node-proxy.ts` installs one process-wide undici dispatcher, and a pooled connection that died unnoticed stayed in use for the whole retry budget. `createProviderTransportHealth` counts consecutive unanswered failures per origin and rebuilds the transport after the second one, once per streak, at most once every 30 seconds process-wide, and never for `dns`. The replacement is installed before the previous dispatcher is closed, the previous one is closed gracefully (`close()`, never `destroy()`), and the configured route — including an environment-proxy bootstrap — is reproduced, so another session's in-flight request finishes on the pool it started on. A captured cause also reports `phase: request`, which is what the attempt actually did, instead of the synthetic start that surfaced it. The retry budget, backoff curve, classification, and user-visible codes are unchanged. New tests cover the capture, its redaction (no token, Authorization value, or URL query secret can reach a field), the per-origin threshold, the throttle, graceful replacement, route preservation, and an in-flight request surviving a rebuild; reverting the source leaves them red. --- packages/agent-runtime/src/agent-errors.ts | 20 ++ packages/agent-runtime/src/node-proxy.test.ts | 114 ++++++++- packages/agent-runtime/src/node-proxy.ts | 177 +++++++++++++- .../agent-runtime/src/provider-retry.test.ts | 82 ++++++- packages/agent-runtime/src/provider-retry.ts | 34 ++- .../src/provider-transport-recovery.test.ts | 231 ++++++++++++++++++ .../src/provider-transport-recovery.ts | 206 ++++++++++++++++ packages/agent-runtime/src/runtime.test.ts | 67 +++++ packages/agent-runtime/src/runtime.ts | 101 +++++++- 9 files changed, 996 insertions(+), 36 deletions(-) create mode 100644 packages/agent-runtime/src/provider-transport-recovery.test.ts create mode 100644 packages/agent-runtime/src/provider-transport-recovery.ts diff --git a/packages/agent-runtime/src/agent-errors.ts b/packages/agent-runtime/src/agent-errors.ts index be37504fda..3ac3360a45 100644 --- a/packages/agent-runtime/src/agent-errors.ts +++ b/packages/agent-runtime/src/agent-errors.ts @@ -315,6 +315,26 @@ export function describeNetworkFailure( }; } +/** + * Network diagnosis for a caller that still holds the rejected request's Error. + * + * `classifyAgentError` normally sees pi-ai's flattened `errorMessage`, so an + * errno that only lives in `error.cause` is invisible to it and `fetch failed` + * collapses to `networkCategory: "unknown"` (issue #234). The fetch wrapper in + * `provider-retry.ts` is the last layer that still sees the original object, so + * it describes the cause chain there and hands these validated fields to the + * runtime. The fields come from `describeNetworkFailure`, so a captured cause is + * exactly as narrow as a classified one: an errno-shaped code, a lowercase + * syscall, and a bare hostname — never a URL, port, path, query, or credential. + */ +export function networkFailureDiagnostics( + err: unknown, + message: string, +): { category: NetworkFailureCategory; fields: Record } { + const network = describeNetworkFailure(err, message); + return { category: network.category, fields: networkDetailFields(network) }; +} + export function classifyAgentError(err: unknown): ClassifiedAgentError { const rawMessage = typeof err === "string" diff --git a/packages/agent-runtime/src/node-proxy.test.ts b/packages/agent-runtime/src/node-proxy.test.ts index 96d35810ab..d0e8c03026 100644 --- a/packages/agent-runtime/src/node-proxy.test.ts +++ b/packages/agent-runtime/src/node-proxy.test.ts @@ -1,7 +1,14 @@ import { createServer as createHttpServer, type Server as HttpServer } from "node:http"; import { connect, createServer, type Server } from "node:net"; -import { afterEach, describe, expect, it } from "vitest"; -import { applyNodeNetworkProxy, proxyBypassMatcher } from "./node-proxy.js"; +import { getGlobalDispatcher } from "undici"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + activeNodeTransportRoute, + applyNodeNetworkProxy, + PROVIDER_TRANSPORT_REBUILD_MIN_INTERVAL_MS, + proxyBypassMatcher, + rebuildNodeNetworkTransport, +} from "./node-proxy.js"; type TestServer = Server | HttpServer; @@ -191,3 +198,106 @@ describe("proxyBypassMatcher", () => { expect(proxyBypassMatcher("")("localhost", 80)).toBe(false); }); }); + +/** + * A rebuild is a process-wide action, so its proof is that the swap is not a + * teardown: a request already dispatched keeps running on the pool it started + * on, and the replaced pool is closed gracefully instead of being destroyed + * under another session's request (issue #234). + */ +describe("provider transport rebuild", () => { + /** + * Monotonic synthetic clock. Each test takes a fresh window and may spend one + * inside itself (the throttle is process-wide module state), so a window is + * two intervals wide: no test can be throttled by the one before it. + */ + let clock = 0; + const nextWindow = (): number => { + clock += 2 * PROVIDER_TRANSPORT_REBUILD_MIN_INTERVAL_MS; + return clock; + }; + + afterEach(() => { + applyNodeNetworkProxy({ mode: "direct" }); + }); + + it("keeps a request that is already in flight and closes the old pool gracefully", async () => { + const target = createHttpServer((_request, response) => { + setTimeout(() => { + response.writeHead(200, { "content-type": "text/plain" }); + response.end("slow but complete"); + }, 250); + }); + const port = await listen(target); + applyNodeNetworkProxy({ mode: "direct" }); + const before = getGlobalDispatcher(); + const closeSpy = vi.spyOn(before, "close"); + const destroySpy = vi.spyOn(before, "destroy"); + + const inFlight = fetch(`http://127.0.0.1:${port}/slow`); + await new Promise((resolve) => setTimeout(resolve, 50)); + const result = rebuildNodeNetworkTransport(nextWindow()); + + expect(result.rebuilt).toBe(true); + expect(result.route).toBe("direct"); + expect(getGlobalDispatcher()).not.toBe(before); + // `close()` waits for the requests already dispatched; `destroy()` would + // have aborted them. (undici's own `close` re-enters the method, so the + // count is not asserted — the mode is.) + expect(closeSpy).toHaveBeenCalled(); + expect(destroySpy).not.toHaveBeenCalled(); + const response = await inFlight; + expect(response.status).toBe(200); + expect(response.status).toBe(200); + await expect(response.text()).resolves.toBe("slow but complete"); + + await close(target); + }); + + it("throttles rebuilds so a long outage cannot churn every session's pool", () => { + applyNodeNetworkProxy({ mode: "direct" }); + const window = nextWindow(); + + expect(rebuildNodeNetworkTransport(window).rebuilt).toBe(true); + const swapped = getGlobalDispatcher(); + + expect(rebuildNodeNetworkTransport(window + 1)).toEqual({ + rebuilt: false, + route: "direct", + }); + expect(getGlobalDispatcher()).toBe(swapped); + expect( + rebuildNodeNetworkTransport( + window + PROVIDER_TRANSPORT_REBUILD_MIN_INTERVAL_MS, + ), + ).toEqual({ rebuilt: true, route: "direct" }); + expect(getGlobalDispatcher()).not.toBe(swapped); + }); + + it("reproduces a custom proxy route instead of falling back to direct", () => { + applyNodeNetworkProxy({ + mode: "custom", + url: "http://127.0.0.1:3128", + }); + expect(activeNodeTransportRoute()).toBe("http-proxy"); + const before = getGlobalDispatcher(); + const closeSpy = vi.spyOn(before, "close"); + const destroySpy = vi.spyOn(before, "destroy"); + + expect(rebuildNodeNetworkTransport(nextWindow())).toEqual({ + rebuilt: true, + route: "http-proxy", + }); + expect(getGlobalDispatcher()).not.toBe(before); + expect(activeNodeTransportRoute()).toBe("http-proxy"); + expect(closeSpy).toHaveBeenCalled(); + expect(destroySpy).not.toHaveBeenCalled(); + }); + + it("names the route a proxy scheme and a SOCKS tunnel take", () => { + applyNodeNetworkProxy({ mode: "custom", url: "socks5://127.0.0.1:1080" }); + expect(activeNodeTransportRoute()).toBe("socks5-proxy"); + applyNodeNetworkProxy({ mode: "direct" }); + expect(["direct", "environment-proxy"]).toContain(activeNodeTransportRoute()); + }); +}); diff --git a/packages/agent-runtime/src/node-proxy.ts b/packages/agent-runtime/src/node-proxy.ts index 6fa1254692..9f7c3793f2 100644 --- a/packages/agent-runtime/src/node-proxy.ts +++ b/packages/agent-runtime/src/node-proxy.ts @@ -19,6 +19,7 @@ import { } from "@pi-desktop/shared"; import { Agent, + EnvHttpProxyAgent, ProxyAgent, fetch as undiciFetch, getGlobalDispatcher, @@ -27,6 +28,23 @@ import { type buildConnector, } from "undici"; +/** + * Whether Node built its own fetch transport on the environment proxy. Node + * reads `NODE_USE_ENV_PROXY` once, at startup, and installs an + * `EnvHttpProxyAgent`; replacing the global dispatcher (which this module does + * for a custom proxy) therefore has to remember that choice, or a later + * transport rebuild would silently drop proxy routing. + */ +const envProxyAtStartup = process.env.NODE_USE_ENV_PROXY === "1"; + +/** + * A transport rebuild swaps a dispatcher every session's provider traffic + * shares. Closing idle sockets is cheap, but churning the pool while a network + * is down for minutes would make every other session pay a fresh TCP/TLS + * handshake on its next request, so rebuilds are rate limited. + */ +export const PROVIDER_TRANSPORT_REBUILD_MIN_INTERVAL_MS = 30_000; + let originalFetch: typeof fetch | null = null; let originalDispatcher: Dispatcher | null = null; let installed = false; @@ -34,6 +52,38 @@ let activeDispatcher: Dispatcher | null = null; /** Plain agent the bypass list routes to; closed together with the proxy. */ let activeDirectDispatcher: Dispatcher | null = null; +/** Route the sidecar's provider traffic takes, for failure diagnostics. */ +export type NodeTransportRoute = + | "direct" + | "environment-proxy" + | "http-proxy" + | "socks5-proxy"; + +/** Settings behind the installed pair, so a rebuild reproduces exactly them. */ +let activeCustomSettings: NetworkProxySettings | null = null; +/** Negative infinity: a process that has never rebuilt is never throttled. */ +let lastTransportRebuildAt = Number.NEGATIVE_INFINITY; + +function defaultRoute(): NodeTransportRoute { + return envProxyAtStartup ? "environment-proxy" : "direct"; +} + +function routeForSettings(settings: NetworkProxySettings): NodeTransportRoute { + if (settings.mode !== "custom") return defaultRoute(); + return /^socks/i.test(settings.url ?? "") ? "socks5-proxy" : "http-proxy"; +} + +/** + * Route the next provider request takes. Reported as `networkRoute` on a + * transport failure, because "the provider is down" and "the tunnel died" look + * identical in an errno (issue #234). + */ +export function activeNodeTransportRoute(): NodeTransportRoute { + return activeCustomSettings + ? routeForSettings(activeCustomSettings) + : defaultRoute(); +} + function ensurePatched(): void { if (installed) return; installed = true; @@ -41,21 +91,26 @@ function ensurePatched(): void { originalDispatcher = getGlobalDispatcher(); } +/** + * The dispatcher Node would install by itself. A rebuild must reproduce it, not + * fall back to a direct connection while the process runs on an env proxy. + */ +function defaultDispatcher(): Dispatcher { + return envProxyAtStartup ? new EnvHttpProxyAgent() : new Agent(); +} + function closeActiveDispatchers(): void { - if (activeDispatcher && typeof activeDispatcher.close === "function") { - void activeDispatcher.close(); - } - if (activeDirectDispatcher && typeof activeDirectDispatcher.close === "function") { - void activeDirectDispatcher.close(); - } + const dispatchers = [activeDispatcher, activeDirectDispatcher]; activeDispatcher = null; activeDirectDispatcher = null; + for (const dispatcher of dispatchers) closeDispatcher(dispatcher); } function restoreDefault(): void { closeActiveDispatchers(); + activeCustomSettings = null; if (originalDispatcher) setGlobalDispatcher(originalDispatcher); - else setGlobalDispatcher(new Agent()); + else setGlobalDispatcher(defaultDispatcher()); if (originalFetch) globalThis.fetch = originalFetch; } @@ -69,11 +124,27 @@ export function applyNodeNetworkProxy( restoreDefault(); return; } - const parsed = parseProxyUrl(settings.url ?? ""); - if (!parsed.ok) { + const built = createCustomProxyDispatchers(settings); + if (!built) { restoreDefault(); return; } + installCustomProxyDispatchers(built); + activeCustomSettings = settings; +} + +type CustomProxyDispatchers = { + dispatcher: Dispatcher; + direct: Dispatcher; + route: NodeTransportRoute; +}; + +/** Build the proxied dispatcher pair for one custom-proxy configuration. */ +function createCustomProxyDispatchers( + settings: NetworkProxySettings, +): CustomProxyDispatchers | null { + const parsed = parseProxyUrl(settings.url ?? ""); + if (!parsed.ok) return null; const proxied: Dispatcher = parsed.value.isSocks ? new Agent({ connect: socksConnector(parsed.value) }) : new ProxyAgent(parsed.value.href); @@ -87,11 +158,91 @@ export function applyNodeNetworkProxy( ? direct.dispatch(options, handler) : next(options, handler), ); - closeActiveDispatchers(); - activeDispatcher = dispatcher; - activeDirectDispatcher = direct; - setGlobalDispatcher(dispatcher); + return { + dispatcher, + direct, + route: parsed.value.isSocks ? "socks5-proxy" : "http-proxy", + }; +} + +/** + * Install a freshly built pair and retire the previous one. + * + * Order matters: the new dispatcher becomes global before the old one is + * closed, so no request can be dispatched into a dispatcher that is already + * closing. `close()` is graceful — undici lets the requests already in flight + * finish on their own sockets and only then tears the pool down — so a rebuild + * triggered by one failing session can never abort another session's request. + * The old pair is released by its owner here, which is why this module keeps the + * reference: nothing else is left holding those sockets. + */ +function installCustomProxyDispatchers(built: CustomProxyDispatchers): void { + const previous = [activeDispatcher, activeDirectDispatcher]; + activeDispatcher = built.dispatcher; + activeDirectDispatcher = built.direct; + setGlobalDispatcher(built.dispatcher); globalThis.fetch = undiciFetch as unknown as typeof fetch; + for (const dispatcher of previous) closeDispatcher(dispatcher); +} + +/** + * Close one replaced dispatcher. A rejection means a socket failed to close: it + * must stay visible without rejecting an unawaited promise. + */ +function closeDispatcher(dispatcher: Dispatcher | null): void { + if (!dispatcher || typeof dispatcher.close !== "function") return; + void dispatcher.close().catch((error: unknown) => { + process.stderr.write( + `[agent-runtime] provider transport close failed: ${ + error instanceof Error ? error.message : String(error) + }\n`, + ); + }); +} + +export type TransportRebuildResult = { + /** False when the process-wide throttle skipped this rebuild. */ + rebuilt: boolean; + route: NodeTransportRoute; +}; + +/** + * Rebuild the shared transport after the same origin failed repeatedly without + * ever answering (issue #234). + * + * undici keeps its sockets in one pool per dispatcher, and a socket that died + * without the pool noticing is not retired by the failure itself, so a replay + * can fail the same way ten times. Closing the pool is the public way to + * invalidate it; `createProviderTransportHealth` in + * `provider-transport-recovery.ts` owns the question of when that is justified. + * Returns whether a rebuild happened, so the caller can record it. + */ +export function rebuildNodeNetworkTransport( + now = Date.now(), +): TransportRebuildResult { + ensurePatched(); + const route = activeNodeTransportRoute(); + if (now - lastTransportRebuildAt < PROVIDER_TRANSPORT_REBUILD_MIN_INTERVAL_MS) { + return { rebuilt: false, route }; + } + lastTransportRebuildAt = now; + const customSettings = activeCustomSettings; + const rebuilt = customSettings + ? createCustomProxyDispatchers(customSettings) + : null; + if (rebuilt) { + installCustomProxyDispatchers(rebuilt); + return { rebuilt: true, route: rebuilt.route }; + } + + const previous = getGlobalDispatcher(); + const fresh = defaultDispatcher(); + setGlobalDispatcher(fresh); + // `restoreDefault` reinstalls the captured dispatcher, so never leave that + // reference pointing at the pool this rebuild just closed. + if (previous === originalDispatcher) originalDispatcher = fresh; + closeDispatcher(previous); + return { rebuilt: true, route }; } export type ProxyBypassMatcher = (hostname: string, port: number) => boolean; diff --git a/packages/agent-runtime/src/provider-retry.test.ts b/packages/agent-runtime/src/provider-retry.test.ts index 4f6dd0b918..a01e8d13f1 100644 --- a/packages/agent-runtime/src/provider-retry.test.ts +++ b/packages/agent-runtime/src/provider-retry.test.ts @@ -18,12 +18,14 @@ import { stripOutputLimitFields, withoutDerivedOutputLimit, } from "./provider-retry.js"; +import { activeNodeTransportRoute } from "./node-proxy.js"; +import { describeProviderFetchFailure } from "./provider-transport-recovery.js"; +import type { ClassifiedAgentError } from "./agent-errors.js"; const model = { id: "model", api: "openai-completions", provider: "provider", - name: "Model", reasoning: false, input: ["text"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, @@ -251,6 +253,84 @@ describe("provider rate-limit retry", () => { expect(snapshot).toBeUndefined(); }); + it("describes the transport cause of a rejected fetch and keeps the error", async () => { + const failures: Array> = []; + const original = Object.assign(new TypeError("fetch failed"), { + cause: Object.assign(new Error("read ECONNRESET"), { + code: "ECONNRESET", + syscall: "read", + }), + }); + const wrapped = captureProviderResponse( + async () => { + throw original; + }, + (_response, _requestBytes, failure) => { + failures.push(failure); + }, + ); + + await expect( + wrapped("https://chatgpt.com/backend-api/codex/responses", {}), + ).rejects.toBe(original); + // The first entry is the clear before the attempt, the last is the cause. + expect(failures).toEqual([ + undefined, + { + origin: "https://chatgpt.com", + category: "reset", + fields: { + networkCategory: "reset", + networkCode: "ECONNRESET", + networkSyscall: "read", + networkRoute: activeNodeTransportRoute(), + }, + }, + ]); + }); + it("names the failing layer on the retried error, not just fetch failed", async () => { + let attempts = 0; + const retried: ClassifiedAgentError[] = []; + const failure = describeProviderFetchFailure( + Object.assign(new TypeError("fetch failed"), { + cause: Object.assign(new Error("read ECONNRESET"), { + code: "ECONNRESET", + }), + }), + "https://chatgpt.com/backend-api/codex/responses", + ); + const stream = createProviderRetryStream( + model, + context, + {}, + () => { + attempts += 1; + return failedStream({ errorMessage: "fetch failed" }); + }, + { + claim: () => (attempts <= 1 ? attempts : undefined), + headers: () => undefined, + status: () => undefined, + failure: () => failure, + onRetry: ({ error }) => { + retried.push(error); + }, + }, + ); + + const final = await stream.result(); + + // The retry indicator renders this object, so the errno has to be on it; + // the provider message itself stays untouched. + expect(retried).toHaveLength(1); + expect(retried[0]).toMatchObject({ + code: "NETWORK_ERROR", + message: "fetch failed", + details: { networkCategory: "reset", networkCode: "ECONNRESET" }, + }); + expect(final.errorMessage).toBe("fetch failed"); + }); + it("reports the request body size even when the request dies first", async () => { const seen: Array<[number | undefined, number | undefined]> = []; const body = JSON.stringify({ model: "gpt-5.6-sol", input: "hello" }); diff --git a/packages/agent-runtime/src/provider-retry.ts b/packages/agent-runtime/src/provider-retry.ts index 69da128dc8..ac816ac2b8 100644 --- a/packages/agent-runtime/src/provider-retry.ts +++ b/packages/agent-runtime/src/provider-retry.ts @@ -13,7 +13,11 @@ import { classifyAgentError, type ClassifiedAgentError, } from "./agent-errors.js"; - +import { + describeProviderFetchFailure, + type ProviderFetchFailure, + withProviderFetchFailure, +} from "./provider-transport-recovery.js"; /** Maximum number of retries after the first rate-limited request. */ export const PROVIDER_RATE_LIMIT_MAX_RETRIES = PROVIDER_RETRY_MAX_RETRIES; export const PROVIDER_RATE_LIMIT_INITIAL_DELAY_MS = 2_000; @@ -132,6 +136,8 @@ export type ProviderRetryController = { headers: () => Readonly> | undefined; /** Status captured even when the provider body omits the HTTP code. */ status?: () => number | undefined; + /** Cause captured for the attempt that just failed, when the fetch rejected. */ + failure?: () => ProviderFetchFailure | undefined; onRetry?: (input: { error: ClassifiedAgentError; phase: ProviderRetryPhase; @@ -319,13 +325,15 @@ function requestBodyBytes(body: unknown): number | undefined { /** Capture HTTP status/headers, including failed 429 responses that pi-ai's * onResponse callback intentionally does not expose. The second argument is the - * outgoing request size, reported even when the request dies before headers — - * exactly the case worth correlating with a network failure (issue #234). */ + * outgoing request size, reported even when the request dies before headers, and + * the third is the transport cause of a rejection — the same two signals the + * reporter of issue #234 had no way to read. */ export function captureProviderResponse( fetchFn: FetchFunction | undefined, onResponse: ( response?: ProviderResponseSnapshot, requestBytes?: number, + failure?: ProviderFetchFailure, ) => void, ): FetchFunction { const baseFetch = fetchFn ?? globalThis.fetch; @@ -343,7 +351,15 @@ export function captureProviderResponse( onResponse({ status: response.status, headers }, requestBytes); return response; } catch (error) { - onResponse(undefined, requestBytes); + // Last point that still sees the original Error: pi-ai hands the runtime a + // flattened `errorMessage`, in which the errno undici keeps in + // `error.cause` is already gone (issue #234). Describing it here does not + // change the rejection the provider sees. + onResponse( + undefined, + requestBytes, + describeProviderFetchFailure(error, input), + ); throw error; } }; @@ -434,9 +450,13 @@ export function createProviderRetryStream( typeof event.error.errorMessage === "string" ? event.error.errorMessage : event.error; - const error = classifyProviderError( - errorMessage, - controller.status?.(), + // Fold in the cause the fetch wrapper captured for this attempt: the + // message pi-ai hands over is already flattened, so without it the + // retry indicator and the terminal error read `fetch failed` with no + // errno (issue #234). + const error = withProviderFetchFailure( + classifyProviderError(errorMessage, controller.status?.()), + controller.failure?.(), ); if (!limitRepairTried && isOpaqueBadRequest(error)) { opaqueLimitRejection = error; diff --git a/packages/agent-runtime/src/provider-transport-recovery.test.ts b/packages/agent-runtime/src/provider-transport-recovery.test.ts new file mode 100644 index 0000000000..39c1aa0256 --- /dev/null +++ b/packages/agent-runtime/src/provider-transport-recovery.test.ts @@ -0,0 +1,231 @@ +import { describe, expect, it } from "vitest"; +import { activeNodeTransportRoute } from "./node-proxy.js"; +import { + createProviderTransportHealth, + describeProviderFetchFailure, + explainsProviderFetchFailure, + PROVIDER_TRANSPORT_REBUILD_THRESHOLD, + type ProviderFetchFailure, + withProviderFetchFailure, +} from "./provider-transport-recovery.js"; + +/** The reporter's endpoint (issue #234): a provider fetch that never answered. */ +const CODEX_URL = "https://chatgpt.com/backend-api/codex/responses"; + +function fetchFailed(cause?: unknown): TypeError { + return Object.assign(new TypeError("fetch failed"), { + ...(cause === undefined ? {} : { cause }), + }); +} + +function coded(code: string, extra: Record = {}): Error { + return Object.assign(new Error(`socket failure ${code}`), { code, ...extra }); +} + +function buildFailure( + origin: string | undefined, + category: ProviderFetchFailure["category"], +): ProviderFetchFailure { + return { + ...(origin === undefined ? {} : { origin }), + category, + fields: { networkCategory: category }, + }; +} + +describe("describeProviderFetchFailure", () => { + it("names the transport layer behind a bare fetch failed", () => { + // Exactly what pi-ai flattens into `errorMessage` before classification can + // see it: the errno only exists in the cause chain. + const failure = describeProviderFetchFailure( + fetchFailed(coded("ECONNRESET", { syscall: "read" })), + CODEX_URL, + ); + + expect(failure).toEqual({ + origin: "https://chatgpt.com", + category: "reset", + fields: { + networkCategory: "reset", + networkCode: "ECONNRESET", + networkSyscall: "read", + networkRoute: activeNodeTransportRoute(), + }, + }); + }); + + it("reads the errno out of an undici happy-eyeballs aggregate", () => { + const aggregate = new AggregateError( + [ + coded("ECONNREFUSED", { hostname: "chatgpt.com" }), + coded("ECONNREFUSED", { hostname: "chatgpt.com" }), + ], + "all connection attempts failed", + ); + + expect( + describeProviderFetchFailure(fetchFailed(aggregate), CODEX_URL), + ).toMatchObject({ + category: "refused", + fields: { + networkCategory: "refused", + networkCode: "ECONNREFUSED", + networkHost: "chatgpt.com", + }, + }); + }); + + it("reads the origin of an object-shaped request input", () => { + expect( + describeProviderFetchFailure(fetchFailed(coded("ENOTFOUND")), { + url: CODEX_URL, + })?.origin, + ).toBe("https://chatgpt.com"); + }); + + it("treats an abort as a stopped request, not a broken connection", () => { + // A stop must not count towards a rebuild, and must not be reported as a + // transport cause: the connection was never the problem. + expect( + describeProviderFetchFailure( + Object.assign(new Error("Request aborted"), { name: "AbortError" }), + CODEX_URL, + ), + ).toBeUndefined(); + expect( + describeProviderFetchFailure(fetchFailed(coded("ECONNRESET")), { + url: "not a url", + })?.origin, + ).toBeUndefined(); + }); + + it("never reports a credential, a URL or a query string", () => { + const error = fetchFailed( + Object.assign( + new Error( + "connect ECONNRESET https://api.example.com/v1/chat?api_key=sk-live-secret Authorization: Bearer token-secret", + ), + { + code: "ECONNRESET", + syscall: "read", + hostname: "user:pass@api.example.com", + }, + ), + ); + + const failure = describeProviderFetchFailure( + error, + "https://user:pass@api.example.com/v1/chat?api_key=sk-live-secret", + ); + const serialized = JSON.stringify(failure); + + expect(failure?.origin).toBe("https://api.example.com"); + expect(serialized).not.toContain("sk-live-secret"); + expect(serialized).not.toContain("token-secret"); + expect(serialized).not.toContain("user:pass"); + // A `hostname` that is not a bare hostname is dropped rather than trimmed. + expect(failure?.fields).not.toHaveProperty("networkHost"); + }); +}); + +describe("provider transport health", () => { + it("rebuilds only after the same origin fails unbounded times in a row", () => { + const health = createProviderTransportHealth(); + const first = buildFailure("https://a.example.com", "reset"); + const other = buildFailure("https://b.example.com", "reset"); + + expect(PROVIDER_TRANSPORT_REBUILD_THRESHOLD).toBe(2); + expect(health.observeFailure(first)).toBe(false); + // A different origin is a different connection: two unrelated failures must + // not add up into one rebuild. + expect(health.observeFailure(other)).toBe(false); + expect(health.observeFailure(other)).toBe(true); + }); + + it("drops the streak when an attempt reaches the provider", () => { + const health = createProviderTransportHealth(); + const failure = buildFailure("https://a.example.com", "timeout"); + + expect(health.observeFailure(failure)).toBe(false); + health.observeResponse(); + expect(health.observeFailure(failure)).toBe(false); + expect(health.observeFailure(failure)).toBe(true); + // A new turn starts from no evidence at all. + health.reset(); + expect(health.observeFailure(failure)).toBe(false); + }); + + it("never rebuilds for a name that a fresh pool cannot resolve", () => { + const health = createProviderTransportHealth(); + const dns = buildFailure("https://a.example.com", "dns"); + const reset = buildFailure("https://a.example.com", "reset"); + + expect(health.observeFailure(dns)).toBe(false); + expect(health.observeFailure(dns)).toBe(false); + // DNS also clears the streak: a resolution failure is not evidence that the + // pooled connection is broken. + expect(health.observeFailure(reset)).toBe(false); + expect(health.observeFailure(reset)).toBe(true); + }); + + it("spends one rebuild per streak", () => { + const health = createProviderTransportHealth(); + const failure = buildFailure("https://a.example.com", "reset"); + + expect(health.observeFailure(failure)).toBe(false); + expect(health.observeFailure(failure)).toBe(true); + // A long outage must not churn the pool on every attempt. + expect(health.observeFailure(failure)).toBe(false); + expect(health.observeFailure(failure)).toBe(true); + }); + + it("counts an unparseable origin in one fallback bucket", () => { + const health = createProviderTransportHealth(); + + expect(health.observeFailure(buildFailure(undefined, "reset"))).toBe(false); + expect(health.observeFailure(buildFailure(undefined, "reset"))).toBe(true); + }); +}); + +describe("withProviderFetchFailure", () => { + it("replaces the unknown the text classifier fell back to", () => { + const classified = { + code: "NETWORK_ERROR", + message: "fetch failed", + retriable: true, + details: { networkCategory: "unknown" }, + }; + const failure = describeProviderFetchFailure( + fetchFailed(coded("ECONNRESET")), + CODEX_URL, + ); + + expect(withProviderFetchFailure(classified, failure)).toEqual({ + code: "NETWORK_ERROR", + message: "fetch failed", + retriable: true, + details: { + networkCategory: "reset", + networkCode: "ECONNRESET", + networkRoute: activeNodeTransportRoute(), + }, + }); + }); + + it("leaves a reached provider and a missing capture untouched", () => { + const limited = { + code: "PROVIDER_RATE_LIMITED", + message: "429: slow down", + retriable: true, + }; + const failure = describeProviderFetchFailure( + fetchFailed(coded("ECONNRESET")), + CODEX_URL, + ); + + expect(withProviderFetchFailure(limited, failure)).toBe(limited); + expect(withProviderFetchFailure(limited, undefined)).toBe(limited); + expect(explainsProviderFetchFailure("NETWORK_ERROR")).toBe(true); + expect(explainsProviderFetchFailure("PROVIDER_RATE_LIMITED")).toBe(false); + }); +}); diff --git a/packages/agent-runtime/src/provider-transport-recovery.ts b/packages/agent-runtime/src/provider-transport-recovery.ts new file mode 100644 index 0000000000..a3ff6637e1 --- /dev/null +++ b/packages/agent-runtime/src/provider-transport-recovery.ts @@ -0,0 +1,206 @@ +/** + * Recovery policy for a provider transport that keeps failing (issue #234). + * + * One turn retried a Codex request ten times and every attempt died with a bare + * `fetch failed`, while a new turn on the same network and provider recovered + * immediately. `phase=stream` / `streamMs=1~2` looked like a failure after the + * response headers, but pi-agent-core emits `message_start` for a stream that + * ended without ever emitting `start`, so the pair actually means the opposite: + * the request never produced a response, on all ten attempts, each of which + * spent seconds in `provider-retry.ts`'s `phase=request` retries. + * + * A rejected provider fetch says two things beyond its errno. It is the only + * transport event that proves the connection never worked, and it is the last + * moment the original Error (with the cause chain node/undici put the real errno + * in) still exists — pi-ai only forwards a flattened `errorMessage`. So the + * capture and the decision live together here: + * + * - `describeProviderFetchFailure` turns the rejection into the same bounded, + * validated `network*` fields a directly classified network error carries. + * - `createProviderTransportHealth` decides when the shared pool is worth + * rebuilding. It is deliberately not "rebuild on every rejection": the + * dispatcher in `node-proxy.ts` is process-wide, so a rebuild closes the idle + * sockets of every session, including ones that are not failing. One failure + * is a blip and a replay over the same pool is what a retry is supposed to do; + * the *same origin* failing twice in a row without ever answering is the point + * where the pool's view of that origin is not recovering on its own, and it + * leaves eight of the ten attempts to prove that a fresh pool helped. + * - `dns` never triggers a rebuild: name resolution happens before a socket + * exists, so a fresh pool cannot change the answer. + */ +import type { + ClassifiedAgentError, + NetworkFailureCategory, +} from "./agent-errors.js"; +import { networkFailureDiagnostics } from "./agent-errors.js"; +import { activeNodeTransportRoute } from "./node-proxy.js"; + +/** Consecutive unanswered failures for one origin that justify a rebuild. */ +export const PROVIDER_TRANSPORT_REBUILD_THRESHOLD = 2; + +/** A rejected provider fetch, described for the log and the error details. */ +export type ProviderFetchFailure = { + /** + * Origin of the failed request. Compared to decide whether two failures are + * consecutive; never reported, because a URL can carry credentials. + */ + origin?: string; + category: NetworkFailureCategory; + /** Validated `network*` fields for `AppError.details`. */ + fields: Record; +}; + +/** + * Error codes a captured fetch cause explains. Mirrors the transient provider + * codes: everything else is either a reached provider (`PROVIDER_RATE_LIMITED`, + * an HTTP status) or a failure a different recovery path repairs. + */ +const EXPLAINED_ERROR_CODES = new Set([ + "NETWORK_ERROR", + "TIMEOUT", + "STREAM_FAILED", + "PROVIDER_ERROR", +]); + +/** Whether a captured fetch cause explains an error about to be surfaced. */ +export function explainsProviderFetchFailure(code: string): boolean { + return EXPLAINED_ERROR_CODES.has(code); +} + +/** + * Categories a fresh pool cannot fix. DNS is the whole list: the lookup happens + * before a connection exists, so rebuilding the pool would close other + * sessions' idle sockets and still resolve the name the same way. + */ +const UNREBUILDABLE_CATEGORIES: ReadonlySet = new Set([ + "dns", +]); + +function isAbort(error: unknown): boolean { + return ( + (error instanceof Error && error.name === "AbortError") || + (typeof error === "object" && + error !== null && + (error as { name?: unknown }).name === "AbortError") + ); +} + +/** + * Origin of the request that failed, for the consecutive-failure comparison. + * `undefined` (an input the URL parser rejects) still works: the failures share + * one fallback bucket instead of never reaching the threshold. + */ +function requestOrigin(input: unknown): string | undefined { + const raw = + typeof input === "string" + ? input + : input instanceof URL + ? input.href + : typeof input === "object" && + input !== null && + typeof (input as { url?: unknown }).url === "string" + ? (input as { url: string }).url + : undefined; + if (raw === undefined) return undefined; + try { + return new URL(raw).origin; + } catch { + return undefined; + } +} + +function errorMessageOf(error: unknown): string { + if (error instanceof Error) return error.message; + return typeof error === "string" ? error : ""; +} + +/** + * Describe a rejected provider fetch, or return undefined when the rejection is + * not a transport fault: an abort the user asked for (a stopped turn must not + * look like a broken connection) or an error carrying an HTTP status, which means + * a response did arrive. + */ +export function describeProviderFetchFailure( + error: unknown, + input?: unknown, +): ProviderFetchFailure | undefined { + if (isAbort(error)) return undefined; + const { category, fields } = networkFailureDiagnostics( + error, + errorMessageOf(error), + ); + const origin = requestOrigin(input); + return { + ...(origin !== undefined ? { origin } : {}), + category, + fields: { ...fields, networkRoute: activeNodeTransportRoute() }, + }; +} + +/** + * Consecutive-failure policy for the shared transport. One instance per runtime, + * so one failing session cannot spend another session's evidence; the rebuild + * itself is throttled by its owner in `node-proxy.ts`. + */ +export type ProviderTransportHealth = { + /** Record an unanswered failure; true when the shared pool should be rebuilt. */ + observeFailure: (failure: ProviderFetchFailure) => boolean; + /** Record that an attempt reached the provider. */ + observeResponse: () => void; + /** Drop the streak; called when a new turn starts. */ + reset: () => void; +}; + +export function createProviderTransportHealth(): ProviderTransportHealth { + let origin: string | undefined; + let consecutive = 0; + + const clear = (): void => { + origin = undefined; + consecutive = 0; + }; + + return { + observeFailure(failure) { + if (UNREBUILDABLE_CATEGORIES.has(failure.category)) { + clear(); + return false; + } + if (consecutive > 0 && failure.origin !== origin) { + // A different origin is a different connection. Restart the streak + // rather than adding two unrelated failures together. + origin = failure.origin; + consecutive = 1; + return false; + } + origin = failure.origin; + consecutive += 1; + if (consecutive < PROVIDER_TRANSPORT_REBUILD_THRESHOLD) return false; + // One rebuild per streak: a further rebuild has to earn its own evidence, + // so a long outage cannot churn the pool on every attempt. + clear(); + return true; + }, + observeResponse: clear, + reset: clear, + }; +} + +/** + * Apply a captured cause to an error that was classified from a flattened + * message. The errno the capture carries is first-hand, so it replaces the + * `networkCategory: "unknown"` the text-only classifier falls back to; an error + * the capture cannot explain is returned untouched. + */ +export function withProviderFetchFailure( + error: ClassifiedAgentError, + failure: ProviderFetchFailure | undefined, +): ClassifiedAgentError { + if (failure === undefined || !explainsProviderFetchFailure(error.code)) { + return error; + } + return { + ...error, + details: { ...(error.details ?? {}), ...failure.fields }, + }; +} diff --git a/packages/agent-runtime/src/runtime.test.ts b/packages/agent-runtime/src/runtime.test.ts index 533f89d777..49d9f5dd9c 100644 --- a/packages/agent-runtime/src/runtime.test.ts +++ b/packages/agent-runtime/src/runtime.test.ts @@ -18,6 +18,8 @@ import { PROVIDER_RATE_LIMIT_MAX_RETRIES, PROVIDER_TRANSIENT_MAX_RETRIES, } from "./provider-retry.js"; +import { getGlobalDispatcher } from "undici"; +import { describeProviderFetchFailure } from "./provider-transport-recovery.js"; /** * The delegate loop itself is covered in `subagent.test.ts`; here only the * `Task` wiring around it is under test, so `SubagentRun` is replaced by a @@ -1434,6 +1436,71 @@ describe("DesktopAgentRuntime live activity", () => { await runtime.dispose(); }); + it("reports the captured transport cause instead of the flattened message", async () => { + const runtime = createRuntime({ onEvent: vi.fn() }); + // The reporter's log line (issue #234): a bare `fetch failed`, a stream + // phase, a two-millisecond "stream", and 112s of provider wait. pi-ai had + // already flattened the Error, so the text-only classifier can only say + // "unknown" — the fetch wrapper still held the cause chain. + const flattened = classifyAgentError("fetch failed"); + expect(flattened).toMatchObject({ + code: "NETWORK_ERROR", + details: { networkCategory: "unknown" }, + }); + (runtime as any).providerFetchFailure = describeProviderFetchFailure( + Object.assign(new TypeError("fetch failed"), { + cause: Object.assign(new Error("read ECONNRESET"), { + code: "ECONNRESET", + syscall: "read", + }), + }), + "https://chatgpt.com/backend-api/codex/responses", + ); + + expect( + (runtime as any).providerErrorWithDiagnostics( + flattened, + "stream", + 112_442, + 2, + ).details, + ).toMatchObject({ + networkCategory: "reset", + networkCode: "ECONNRESET", + networkSyscall: "read", + // No response ever arrived, so this is not a started stream — the + // synthetic `message_start` pi-agent-core emits for a stream that ended + // without `start` must not read as one. + phase: "request", + providerWaitMs: 112_442, + streamMs: 2, + }); + + await runtime.dispose(); + }); + + it("rebuilds the shared transport only once the same origin fails twice", async () => { + const runtime = createRuntime({ onEvent: vi.fn() }); + const failure = describeProviderFetchFailure( + Object.assign(new TypeError("fetch failed"), { + cause: Object.assign(new Error("read ECONNRESET"), { + code: "ECONNRESET", + }), + }), + "https://chatgpt.com/backend-api/codex/responses", + ); + const before = getGlobalDispatcher(); + + (runtime as any).recoverProviderTransport(failure); + // One unanswered attempt is what a retry is for; the pool stays. + expect(getGlobalDispatcher()).toBe(before); + (runtime as any).recoverProviderTransport(failure); + // The second one has spent its retry on the same dead pool. + expect(getGlobalDispatcher()).not.toBe(before); + + await runtime.dispose(); + }); + it("emits status phases for quiet provider and delegation intervals", async () => { const onEvent = vi.fn(); const runtime = createRuntime({ onEvent }); diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index e5c9f0efad..b28584b22a 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -183,6 +183,14 @@ import { providerSetupRetryDelayMs, } from "./provider-retry.js"; +import { rebuildNodeNetworkTransport } from "./node-proxy.js"; +import { + createProviderTransportHealth, + explainsProviderFetchFailure, + type ProviderFetchFailure, + type ProviderTransportHealth, +} from "./provider-transport-recovery.js"; + export type { RuntimeProviderConfig } from "./provider-binding.js"; export type RuntimePromptAttachment = AgentPromptAttachment & { @@ -1466,6 +1474,21 @@ export class DesktopAgentRuntime { */ private providerRequestBytes?: number; private providerRequestMessages?: number; + /** + * Transport cause of the last provider attempt that rejected before any + * response arrived, captured where the original Error still exists. pi-ai + * only forwards a flattened `errorMessage`, so without this the real errno is + * gone before classification and `fetch failed` reaches the log as + * `networkCategory: "unknown"` (issue #234). + */ + private providerFetchFailure?: ProviderFetchFailure; + /** + * Consecutive-failure policy for the shared undici pool. Rebuilding the pool + * is a process-wide action, so the evidence stays per session and the pool is + * rebuilt only when the same origin keeps failing unanswered (issue #234). + */ + private readonly providerTransportHealth: ProviderTransportHealth = + createProviderTransportHealth(); private pendingProviderRetry?: ReturnType; /** * Shared bounded retry count for non-rate-limit transient failures, counted @@ -1662,6 +1685,10 @@ Delegation rules: this.providerRetryHeaders = undefined; this.providerRequestBytes = undefined; this.providerRequestMessages = context.messages?.length; + // A new model request starts a new transport streak: the evidence that + // justified a rebuild does not carry into the next request (issue #234). + this.providerFetchFailure = undefined; + this.providerTransportHealth.reset(); const requestOptions: SimpleStreamOptions = withProviderHeaders( withOpenCodeSessionHeaders( { @@ -1669,18 +1696,25 @@ Delegation rules: maxRetries: PROVIDER_REQUEST_MAX_RETRIES, sessionId: this.sessionId, // pi-ai only exposes onResponse after a request succeeds. Capture the - // failed response separately so a 429 can honor Retry-After headers. - fetch: captureProviderResponse(options?.fetch, (response, requestBytes) => { - this.providerResponseStatus = response?.status; - this.providerRequestBytes = requestBytes; - // A gateway 502/503 can also state Retry-After, so keep headers for - // every status whose delay is usable instead of only for 429. - this.providerRetryHeaders = carriesRetryDelayHeaders( - response?.status, - ) - ? response?.headers - : undefined; - }), + // failed response separately so a 429 can honor Retry-After headers, + // and capture the transport cause of a rejection while the original + // Error still exists (issue #234). + fetch: captureProviderResponse( + options?.fetch, + (response, requestBytes, failure) => { + this.providerResponseStatus = response?.status; + this.providerRequestBytes = requestBytes; + this.providerFetchFailure = failure; + if (failure) this.recoverProviderTransport(failure); + // A gateway 502/503 can also state Retry-After, so keep headers + // for every status whose delay is usable, not only for 429. + this.providerRetryHeaders = carriesRetryDelayHeaders( + response?.status, + ) + ? response?.headers + : undefined; + }, + ), onResponse: async (response, responseModel) => { this.providerResponseStatus = response.status; await options?.onResponse?.(response, responseModel); @@ -1706,6 +1740,7 @@ Delegation rules: claim: (error, phase) => this.claimProviderRetry(error, phase), headers: () => this.providerRetryHeaders, status: () => this.providerResponseStatus, + failure: () => this.providerFetchFailure, onRetry: ({ error, phase, attempt, delayMs }) => { this.setAgentActivity({ phase: "retrying", @@ -4811,11 +4846,24 @@ Delegation rules: streamMs?: number, ): ReturnType { const existingDetails = isRecord(error.details) ? error.details : {}; + // A capture exists only for an attempt that rejected before any response, so + // it is also the honest phase: whatever the message lifecycle that surfaced + // the failure looks like, this request never reached the provider, and + // pi-agent-core's synthetic `message_start` must not read as a started + // stream (issue #234). + const captured = + this.providerFetchFailure !== undefined && + explainsProviderFetchFailure(error.code) + ? this.providerFetchFailure + : undefined; return { ...error, details: { ...existingDetails, - phase, + // First-hand cause, so it replaces the `unknown` the text classifier + // falls back to for a bare `fetch failed`. + ...(captured ? captured.fields : {}), + phase: captured ? "request" : phase, // Correlation for a failure that produced no response to inspect: how // much context and how many bytes the attempt carried, and which // compaction generation the session was on (issue #234). Counts, flags @@ -4846,6 +4894,31 @@ Delegation rules: }; } + /** + * Rebuild the shared provider transport once the same origin has failed + * repeatedly without ever answering (issue #234). + * + * A rejection that produced no response is the only transport event that says + * the connection itself never worked, so `createProviderTransportHealth` + * counts them per origin and asks for a rebuild after the second one. The + * rebuild swaps a dispatcher every session shares, which is safe precisely + * because the swap is not a teardown: undici resolves the global dispatcher per + * dispatch, in-flight requests keep finishing on the pool they started on, and + * the replaced pool is closed gracefully by its owner in `node-proxy.ts`. A + * process-wide throttle there bounds how often any session can spend that cost, + * and the capture is reported either way, so a rebuild that does not help + * still leaves an accurate record. + */ + private recoverProviderTransport(failure: ProviderFetchFailure): void { + if (!this.providerTransportHealth.observeFailure(failure)) return; + const result = rebuildNodeNetworkTransport(); + process.stderr.write( + `[agent-runtime] provider transport ${ + result.rebuilt ? "rebuilt after" : "rebuild skipped within throttling for" + } a ${failure.category} failure (session=${this.sessionId} route=${result.route})\n`, + ); + } + /** * Clear the per-run recovery state that every entry point driving the agent * loop must start from. Each recovery arms itself mid-run by setting a @@ -4861,6 +4934,8 @@ Delegation rules: this.providerTransientRetryAttempt = 0; this.providerRateLimitRetryAttempt = 0; this.providerRetryHeaders = undefined; + this.providerFetchFailure = undefined; + this.providerTransportHealth.reset(); this.activeProviderRetryAttempt = 0; this.providerRetryInProgress = false; this.suppressProviderRetryRunEnd = false; From e122c8d5e764ea9dddd8ac68c80b44eb7f10368b Mon Sep 17 00:00:00 2001 From: vastsa Date: Wed, 16 Sep 2026 22:12:39 +0800 Subject: [PATCH 2/2] docs(runtime): record the transport diagnosis and its rebuild policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sync the runtime specs (EN and zh-CN) with the new network diagnosis: the captured cause and `networkRoute`, the honest `phase: request` for a request that never answered, and the bounded, conditional rebuild of the shared transport. Add ADR 0271 for the rebuild policy — the thresholds, the graceful replacement, and why the pattern is "same origin twice" rather than every failure — and list it in the ADR index. --- docs/adr/0271-provider-transport-rebuild.md | 88 +++++++++++++++++++ docs/adr/README.md | 1 + docs/spec/03-runtime/01-ipc-protocol.md | 5 +- docs/spec/03-runtime/02-agent-runtime.md | 12 ++- docs/spec/03-runtime/08-error-codes.md | 25 +++++- docs/zh-CN/spec/03-runtime/01-ipc-protocol.md | 4 +- .../zh-CN/spec/03-runtime/02-agent-runtime.md | 12 ++- docs/zh-CN/spec/03-runtime/08-error-codes.md | 18 +++- 8 files changed, 155 insertions(+), 10 deletions(-) create mode 100644 docs/adr/0271-provider-transport-rebuild.md diff --git a/docs/adr/0271-provider-transport-rebuild.md b/docs/adr/0271-provider-transport-rebuild.md new file mode 100644 index 0000000000..c12c8c1d7a --- /dev/null +++ b/docs/adr/0271-provider-transport-rebuild.md @@ -0,0 +1,88 @@ +# ADR 0271: Rebuild the shared provider transport after repeated unanswered failures + +- Status: Accepted for implementation +- Date: 2026-09-16 +- Related: [Error codes](../spec/03-runtime/08-error-codes.md) · [Agent runtime](../spec/03-runtime/02-agent-runtime.md) · [ADR 0212](0212-remove-diagnostic-timing-log-streams.md) · issue #234 + +## Context + +One turn on a long session retried a Codex request ten times over ~112 seconds and +reported `NETWORK_ERROR: fetch failed` for every attempt; a new turn on the same +provider, model, and network succeeded immediately. The log record carried +`phase=stream`, `streamMs=1~2`, and `retryAttempt=10`, and nothing else. + +Two readings of that evidence were possible, and the first one was wrong: + +- `phase=stream` with a two-millisecond stream looks like a failure *after* the + response headers arrived, which would mean the connection worked and the retry + hypothesis is empty. +- The pair is actually the fingerprint of the opposite: pi-agent-core emits + `message_start` for a stream that ended **without** ever emitting `start`, so + `streamMs` measures the synthetic start/end pair, not a started stream. With + `providerWaitMs=112442` covering the whole retry loop and each attempt spending + seconds in `phase=request` retries, every one of the ten attempts died before + any response. + +The consequence of the wrong reading was that the shared transport was never +touched: `node-proxy.ts` installs one process-wide undici dispatcher (a +`ProxyAgent`, a SOCKS5 agent, or the plain agent) and `closeActiveDispatchers()` +ran only when settings changed. A pooled connection that died without the pool +noticing therefore stayed in use for the whole retry budget, and the first-hand +`error.cause` — where node and undici keep `ENOTFOUND`, `ECONNRESET`, +`UND_ERR_SOCKET`, or a TLS code — was already flattened into an `errorMessage` +string by pi-ai before classification, so the log could only say +`networkCategory: unknown`. + +## Decision + +1. Describe the rejected provider fetch where the original Error still exists: + the fetch wrapper in `provider-retry.ts` runs `describeNetworkFailure` on the + live cause chain and reports the same validated fields a directly classified + network error carries (`networkCategory`, `networkCode`, `networkSyscall`, + `networkHost`) instead of letting a bare `fetch failed` fall back to + `unknown`. Field shapes, bounds, and redaction stay owned by + `agent-errors.ts`; no new error code and no new locale string is introduced. +2. Report the transport route (`networkRoute`: `direct`, `environment-proxy`, + `http-proxy`, `socks5-proxy`) with the diagnosis, so a failure at the proxy + hop is readable without inferring it from an errno. +3. Report `phase: request` for a failure that never received a response. The + phase describes what actually happened to the attempt, not which message + lifecycle surfaced it. +4. Rebuild the shared transport when one origin fails this way **repeatedly**: + after the second consecutive unanswered failure for the same origin inside a + turn, once per streak, at most once every 30 seconds process-wide, never for a + `dns` failure (name resolution happens before a socket exists). Rebuild means + rebuilding the configured route from the settings already applied. +5. Swap before closing: the replacement dispatcher is installed, then the + previous one is closed with `close()` — never `destroy()` — and the captured + original-dispatcher reference is repointed if the closed pool was it. In-flight + requests finish on the pool they were dispatched on. + +## Consequences + +- The next occurrence of issue #234 is diagnosable: the log names the layer and + the route, and the retry indicator names the errno. +- A turn can now spend one of its retry attempts on a fresh pool, so a transport + that never recovers on its own stops being reused. A rebuild does not change + the retry budget, the backoff curve, or the error classification. +- Other sessions can observe a rebuild: their idle pooled sockets are closed, so + their next request pays a new TCP/TLS handshake. That is a latency cost, never + a correctness cost, and the 30-second throttle plus the per-origin streak bound + it. +- A transport failure during a `dns` outage triggers no rebuild, which is + deliberate: throttling the pool would spend the cost for no possible benefit. + +## Alternatives considered + +- **Rebuild on every failure.** Rejected: the dispatcher is process-wide, so one + session's single transient failure would churn every other session's pool, and + a long outage would rebuild once per attempt. +- **Never rebuild.** Rejected: it leaves a repeated unanswered failure on the + same origin with no recovery path, which is exactly the reported behavior. +- **Per-session dispatchers.** Rejected for this change: it changes the ownership + model of the transport and the proxy/bypass composition for every provider + request, which is a larger and separate architectural decision. +- **Retain the raw cause message and its `name`.** Rejected: it would put + free-form provider text (URLs, header values, query strings) into logs and the + transcript for a small diagnostic gain over the errno, the category, and the + route. diff --git a/docs/adr/README.md b/docs/adr/README.md index 2b83ebdd1a..42f836cdc3 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -295,5 +295,6 @@ Each ADR includes: | 0266 | [Plugin fs roots follow the calling session](0266-plugin-fs-root-follows-the-calling-session.md) | Accepted (D093) | | 0267 | [Plugin labels follow the app language](0267-plugin-labels-follow-the-app-language.md) | Accepted (amends 0160; ADR 0182) | | 0270 | [Builtin subagents can be switched off](0270-builtin-subagents-can-be-disabled.md) | Accepted for implementation (amends ADR 0063; ADR 0112) | +| 0271 | [Rebuild the shared provider transport after repeated unanswered failures](0271-provider-transport-rebuild.md) | Accepted for implementation (issue #234) | | 0269 | [Move capability documents between the global and project levels](0269-capability-level-transfer.md) | Accepted for implementation (amends ADR 0112) | | 0268 | Remove quotes, annotations, and side chats | Accepted (supersedes message-quotes-and-side-chats / response-annotations / floating-annotation-index) | diff --git a/docs/spec/03-runtime/01-ipc-protocol.md b/docs/spec/03-runtime/01-ipc-protocol.md index 2a25f74dba..c8b92a3d9a 100644 --- a/docs/spec/03-runtime/01-ipc-protocol.md +++ b/docs/spec/03-runtime/01-ipc-protocol.md @@ -735,8 +735,9 @@ context. Manual compaction never silently falls back. Provider `error` events may include bounded diagnostic fields in `AppError.details`: `phase` (`request` or `stream`), `providerStatus`, `providerCode`, `providerWaitMs`, `streamMs`, `retryAttempt`, and, for a -network failure, `networkCategory`, `networkCode`, `networkSyscall` and -`networkHost` plus the request correlation fields `requestMessages`, +network failure, `networkCategory`, `networkCode`, `networkSyscall`, +`networkHost` and `networkRoute` plus the request correlation fields +`requestMessages`, `requestBytes` and `compactionGeneration`. These fields are additive and redacted; they never carry credentials or an unrestricted provider response, and the request fields are counts and byte sizes only. A transient stream diff --git a/docs/spec/03-runtime/02-agent-runtime.md b/docs/spec/03-runtime/02-agent-runtime.md index 566096a2b3..50e1a72678 100644 --- a/docs/spec/03-runtime/02-agent-runtime.md +++ b/docs/spec/03-runtime/02-agent-runtime.md @@ -229,13 +229,23 @@ When the retry budget is exhausted, the final assistant error and lifecycle `AppError.details` when available: `phase` (`request` or `stream`), `providerStatus`, `providerCode`, `providerWaitMs`, `streamMs`, `retryAttempt`, the network diagnosis (`networkCategory`, `networkCode`, -`networkSyscall`, `networkHost`) and the request correlation fields +`networkSyscall`, `networkHost`, `networkRoute`) and the request correlation (`requestMessages`, `requestBytes`, `compactionGeneration`). For a persistent 429 or non-429 transient failure, `retryAttempt` is `10`. Credentials and unrestricted response bodies never enter the event or log. The active-turn status shows the remaining backoff and the retry budget as `Retrying in 0s · attempt 9/10` in English. +Each retry builds a new request, stream, and `AbortController`; the one piece of +state a retry shares is the process-wide undici dispatcher. When the same origin +fails twice in a row without any response and no fresh attempt reaches it, the +transport is rebuilt once (throttled to one rebuild every 30 seconds, never for +`dns`) before the next attempt, so the retry does not replay into a pool whose +connection is already dead. The rebuild installs the replacement before closing +the previous dispatcher and closes it gracefully, and it reproduces the +configured route, so another session's in-flight request finishes on the pool it +started on and a proxy is never silently dropped. + ### 5e. Silent-turn recovery A turn that ends with no tool call and no visible assistant text is invisible diff --git a/docs/spec/03-runtime/08-error-codes.md b/docs/spec/03-runtime/08-error-codes.md index 7a9e5e8aad..2f15adcd1a 100644 --- a/docs/spec/03-runtime/08-error-codes.md +++ b/docs/spec/03-runtime/08-error-codes.md @@ -309,6 +309,28 @@ omitted when it would repeat `networkCode`. Per-layer codes (`DNS_ERROR`, splits the layers without adding user-visible codes and locale strings for each of them. +The diagnosis is read from the live cause chain at the fetch boundary, not only +from the provider message. pi-ai flattens a rejected request into +`errorMessage`, so by the time classification runs the errno undici keeps in +`error.cause` is already gone and a bare `fetch failed` can only be reported as +`networkCategory: unknown`; the fetch wrapper still holds the original Error and +supplies the same validated fields from it. A captured cause also settles the +phase: the fault is reported as `phase: request` because no response ever +arrived, which is what distinguishes it from a stream that ended mid-response. +`networkRoute` (`direct`, `environment-proxy`, `http-proxy`, `socks5-proxy`) +names the hop the request was taking, so a failure at the proxy is readable +without guessing from an errno. + +When one origin fails this way repeatedly inside a turn — twice in a row, +without any response — the provider transport is rebuilt before the next attempt +instead of replaying into the same undici pool. The rebuild is process-wide and +deliberately bounded: one rebuild per streak, at most one every 30 seconds, and +never for a `dns` failure, which a fresh pool cannot change. The replacement is +installed before the previous dispatcher is closed, and the previous one is +closed gracefully, so a request another session already dispatched finishes on +the pool it started on. The route in effect is reproduced, never downgraded to a +direct connection. + ### Permission timeout UI/host timeout emits `PERMISSION_TIMEOUT` internally, tool result presented as denied (`TOOL_DENIED`) to agent. @@ -343,7 +365,8 @@ common credential/header values are redacted before event emission or persistence. When available, the details disclosure may also show bounded `phase`, `providerStatus`, `providerCode`, `providerWaitMs`, `streamMs`, `retryAttempt`, `networkCategory`, `networkCode`, `networkSyscall`, -`networkHost`, `requestMessages`, `requestBytes`, and `compactionGeneration` +`networkHost`, `networkRoute`, `requestMessages`, `requestBytes`, and +`compactionGeneration` fields. The request fields are counts and byte sizes only and the compaction field is the checkpoint generation counter; none of them carries message content. While a transient provider failure retries, the activity indicator's diff --git a/docs/zh-CN/spec/03-runtime/01-ipc-protocol.md b/docs/zh-CN/spec/03-runtime/01-ipc-protocol.md index 98c777c3ff..b1cfa33456 100644 --- a/docs/zh-CN/spec/03-runtime/01-ipc-protocol.md +++ b/docs/zh-CN/spec/03-runtime/01-ipc-protocol.md @@ -617,8 +617,8 @@ type AgentEvent = 提供程序 `error` 事件可能包括以下中的有限诊断字段: `AppError.details`:`phase`(`request` 或 `stream`)、`providerStatus`、 `providerCode`、`providerWaitMs`、`streamMs`、`retryAttempt`,以及网络故障 -时的 `networkCategory`、`networkCode`、`networkSyscall`、`networkHost` 和请求 -关联字段 `requestMessages`、`requestBytes`、`compactionGeneration`。这些字段 +时的 `networkCategory`、`networkCode`、`networkSyscall`、`networkHost`、 +`networkRoute` 和请求关联字段 `requestMessages`、`requestBytes`、`compactionGeneration`。这些字段 都是新增且经过编辑的;它们从不携带凭据或不受限制的提供商响应,请求字段 只有计数与字节大小。瞬时流故障可能会在内部重播 同一回合,没有终端 `error` 事件或重复的辅助消息。 diff --git a/docs/zh-CN/spec/03-runtime/02-agent-runtime.md b/docs/zh-CN/spec/03-runtime/02-agent-runtime.md index be40366c80..52ffe9104f 100644 --- a/docs/zh-CN/spec/03-runtime/02-agent-runtime.md +++ b/docs/zh-CN/spec/03-runtime/02-agent-runtime.md @@ -187,11 +187,17 @@ HTTP 429 处理是一个逻辑回合策略。此路径禁用了 pi-ai 的嵌套 提供程序故障在可用时于 `AppError.details` 中携带有界诊断: `phase`(`request` 或 `stream`)、`providerStatus`、`providerCode`、 `providerWaitMs`、`streamMs`、`retryAttempt`、网络诊断 -(`networkCategory`、`networkCode`、`networkSyscall`、`networkHost`)以及 -请求关联字段(`requestMessages`、`requestBytes`、`compactionGeneration`)。 +(`networkCategory`、`networkCode`、`networkSyscall`、`networkHost`、 +`networkRoute`)以及请求关联字段(`requestMessages`、`requestBytes`、 +`compactionGeneration`)。 对于持续的 429, `retryAttempt` 为 `5`;对于持续的非 429 瞬时故障,它为 `4`。凭据与不受限制的 -响应正文永远不会进入事件或日志。 +响应正文永远不会进入事件或日志。每次重试都会新建请求、流和 `AbortController`; +重试唯一共享的状态是进程级 undici dispatcher。当同一来源在一轮内连续两次没有 +任何响应、且新尝试仍无法到达它时,下一次尝试前会重建一次传输(每 30 秒最多一次, +`dns` 除外),避免重试继续复用连接已死的连接池。重建先安装替换、再优雅关闭旧 +的 dispatcher,并复现已配置的链路,因此其他会话正在进行的请求仍会在原连接池上 +完成,代理也不会被悄悄丢弃。 ### 5e。静默回合恢复 diff --git a/docs/zh-CN/spec/03-runtime/08-error-codes.md b/docs/zh-CN/spec/03-runtime/08-error-codes.md index d8a5665505..9ef6c88122 100644 --- a/docs/zh-CN/spec/03-runtime/08-error-codes.md +++ b/docs/zh-CN/spec/03-runtime/08-error-codes.md @@ -301,6 +301,22 @@ Node sidecar 将提供商 SDK 错误映射到: `TLS_ERROR`、`SOCKET_RESET` 等):分类字段已能区分这些层,而无需为每一层 增加用户可见的错误码与本地化文案。 +诊断来自 fetch 边界处的实时 cause 链,而不只是提供程序消息:pi-ai 会把被 +拒绝的请求摊平成 `errorMessage`,等到分类运行时 undici 存放在 `error.cause` +里的 errno 已经消失,裸 `fetch failed` 只能被记为 `networkCategory: unknown`; +而 fetch 包装层仍持有原始 Error,并从它给出同一组经过校验的字段。捕获到的 +cause 同时确定了阶段:没有任何响应到达时故障记为 `phase: request`,这正是它 +与「响应中途断流」的区别。`networkRoute`(`direct`、`environment-proxy`、 +`http-proxy`、`socks5-proxy`)指出请求实际走的链路,代理这一跳失败无需再从 +errno 猜测。 + +同一来源在一轮内连续两次这样失败(完全没有响应)时,下一次尝试前会重建提供 +程序传输,而不是继续复用同一个 undici 连接池。重建是进程级的、且有明确边界: +每个连续失败序列只重建一次,每 30 秒最多一次,且 `dns` 永不触发重建(新连接池 +无法改变名字解析结果)。替换在关闭旧 dispatcher 之前安装,旧的 dispatcher 采 +用优雅关闭,因此其他会话已派发的请求仍会在它原本使用的连接池上完成。生效的 +链路会被原样复现,绝不会悄悄降级为直连。 + ### 权限超时 UI/host 超时在内部发出 `PERMISSION_TIMEOUT`,工具结果向代理显示为拒绝 (`TOOL_DENIED`)。 @@ -332,7 +348,7 @@ UI/host 超时在内部发出 `PERMISSION_TIMEOUT`,工具结果向代理显示 公共 credential/header 值在事件发射或持久化之前进行编辑。 详细信息披露也可能显示有界的 `phase`、`providerStatus`、`providerCode`、 `providerWaitMs`、`streamMs`、`retryAttempt`、`networkCategory`、 -`networkCode`、`networkSyscall`、`networkHost`、`requestMessages`、 +`networkCode`、`networkSyscall`、`networkHost`、`networkRoute`、`requestMessages`、 `requestBytes` 和 `compactionGeneration` 字段。请求字段只有计数与字节大小, 压缩字段是检查点世代计数器,均不携带消息内容。当瞬时提供商故障正在重试时, 活动指示器的原因气泡会显示本地化摘要、稳定错误码,并在网络故障时显示传输层