From 3a2a88d6de90c15a4573eb733f245dc73c8b2424 Mon Sep 17 00:00:00 2001 From: integ Date: Sat, 12 Sep 2026 23:03:07 -0500 Subject: [PATCH] feat(oauth): keep parallel conversations cached instead of dropping them at a fixed limit The pools that hold a reusable ChatGPT connection per live conversation were capped at 48 nursery and 64 established heads, and a cap that bound silently discarded a conversation whose next turn then resent its whole history uncached. Every number was a guess per machine and per workload; head reuse in a 27.6-hour ledger was identical at every cap from 8 to unlimited. The default is now unbounded; the idle TTLs are the retention policy. Running out of file descriptors on a connection open is detected, from the socket's EMFILE/ENFILE code or, because a hostname dial fails inside getaddrinfo first and surfaces as ENOTFOUND, from a one-descriptor probe on any other pre-open failure, and handled as load shedding: every idle pooled head is terminated (not closed, so the descriptor is freed synchronously), the request takes the existing one-shot transport retry against the freed descriptors, and the user is told once per process, on the parent-notice channel, which limit was hit, how many pooled connections were registered, and the remedy. Busy heads, open sockets and isolated sockets are never shed; when nothing is idle the request fails once with the same actionable message. CLODEX_WS_MAX_CONNECTIONS and CLODEX_WS_MAX_NURSERY_CONNECTIONS remain as an optional cap on the idle pool and now accept any positive integer. Closes #222 --- .claude/docs/oauth-continuation.md | 84 ++++-- src/oauth/responses-websocket.ts | 249 ++++++++++++---- tests/responses-websocket.test.ts | 454 ++++++++++++++++++++++++++++- 3 files changed, 706 insertions(+), 81 deletions(-) diff --git a/.claude/docs/oauth-continuation.md b/.claude/docs/oauth-continuation.md index ed014bd1..9fab9bae 100644 --- a/.claude/docs/oauth-continuation.md +++ b/.claude/docs/oauth-continuation.md @@ -102,11 +102,12 @@ not divide. **Count each response once**: 254 request ids here carry more than o (retries), so joining usage per decision double-counts the successful attempt and inflated these very numbers by ~2.3M tokens. -**Connection pools are process-wide, not per-partition:** `maxConnections` (established, default 64) -and `maxNurseryConnections` (default 48). A head starts in the nursery and is promoted when selected -for its first continuation, before that continuation is known to succeed — so a workload whose -concurrent subagents inherit the parent's Claude session id, and therefore share one partition, can -lose heads before their next turn and with them the continuation. +**Connection pools are process-wide, not per-partition:** `maxConnections` (established) and +`maxNurseryConnections`, both unbounded by default; the idle TTLs bound retention and descriptor +exhaustion is handled, see "The pools are UNBOUNDED by default" below. A head starts in the nursery and is promoted when selected for +its first continuation, before that continuation is known to succeed. Under an env-set cap, a workload whose concurrent +subagents inherit the parent's Claude session id, and therefore share one partition, can lose heads +before their next turn and with them the continuation. **Keeping a fan-out's chains alive trades throwaway sockets for retained heads.** The mismatching turn still opens its own socket; only a LATER turn of that conversation can reuse it. Of the 4,934 @@ -137,7 +138,7 @@ resuming that conversation costs a fresh upgrade and a full-context resend. Cons through the transport (nine sockets before this change, ten after), so the mechanism is established; its frequency in ordinary traffic is not, and the cap replay below finds no eviction-caused loss across the ledger's 27.6 hours: the ten real cap evictions displaced heads that had been idle 217-284 -seconds, already at the nursery TTL. The caps were nonetheless raised, on headroom over observed +seconds, approaching the nursery TTL. The caps were nonetheless raised, on headroom over observed concurrency rather than on any observed loss — see the sizing discussion below. **On a fresh pool, and for a fan-out whose members have distinguishable opening turns, it opened @@ -197,6 +198,49 @@ committed path, which has the same property. A stale memo could only mis-decide don't-isolate — the committed history that drives `previous_response_id` is recomputed from `originalPayload` and never reads it. +**The pools are UNBOUNDED by default; descriptor exhaustion is detected and handled as load +shedding.** `RESPONSES_WS_MAX_CONNECTIONS` and `RESPONSES_WS_MAX_NURSERY_CONNECTIONS` are both +`Infinity`, so `evictOldestIdleGeneration` never fires unless a finite cap is set — +`CLODEX_WS_MAX_CONNECTIONS` / `CLODEX_WS_MAX_NURSERY_CONNECTIONS` (any positive integer; malformed +values are logged and ignored) or the programmatic option, which outranks the environment. Such a +cap bounds the IDLE pool only: busy heads still exceed it and isolated sockets are never counted. +The `ws_head_decision` fields `maxConnections` / `maxNurseryConnections` read `null` when unbounded. +The reasoning, from the sizing work below: every numeric cap was a guess per machine and per +workload, a cap that bound cost a reusable conversation a full uncached resend, and head reuse in +the 27.6-hour ledger was identical at every cap from 8 to unlimited. Retention is bounded by the +idle and hard TTLs, and its ordinary cost is memory (~0.73 MiB per head measured; the pacer's 60 +dials/min times the TTLs bounds idle occupancy at roughly 300 nursery / 1,800 established heads, +against an observed organic peak of 28). Descriptors are the backstop, not the usual bound: Node +raises `RLIMIT_NOFILE`'s soft limit to the hard limit at startup (a stock macOS shell reports a 256 +SOFT limit; Node saw 245,749), so only a service or container with a clamped HARD limit reaches +exhaustion. Note that plain `ulimit -n N` in bash/zsh clamps both, which is why that is the +reproduction and not the counterexample. + +When it does, `descriptorExhaustionCode` in `createConnection`'s `error` handler decides whether a +failed dial was descriptor exhaustion: `EMFILE`/`ENFILE` as the socket's error code, or — because +the shipped route is a hostname, and a full descriptor table fails inside `getaddrinfo` first, which +Node reports as `ENOTFOUND` with no cause — any other socket-open error whose one-descriptor probe +(`fs.openSync(os.devNull)`, closed at once) throws `EMFILE`/`ENFILE`. Reproduced on macOS and Linux +under a hard `ulimit -n 40`; a real `ENOTFOUND` with descriptors available (1,338 in the local +ledgers) stays on the ordinary path. Then `shedIdleConnectionsForDescriptors` **terminates** every +idle pooled head (not `close()`: a close handshake holds the descriptor until the peer answers or +ws's 30 s timer fires; `terminate()` destroys the socket and Node closes the descriptor +synchronously inside `uv_close`), oldest first, busy heads and isolated sockets untouched; the +request then takes the ordinary one-shot transport retry, whose replacement dials against the freed +descriptors. If that retry is starved too there was nothing idle to shed, and the request fails with +a message naming the limit and the remedy — bounded by the single retry, never a loop. Each +occurrence records a `ws_descriptor_exhaustion` diagnostic (`code`, `detectedBy: error_code | +descriptor_probe`, `socketErrorCode`, `heldConnections` = pooled entries registered other than the +failing dial, `shedConnections`); the shed heads are NOT in any decision's `evictions` array, +because the shed happens in the socket error handler rather than at a head decision. The user is +told **once per process**, on the parent-notice channel (the muted stderr under `clodex claude` +would swallow it): which limit, how many pooled connections were registered and shed, and the remedy +— `ulimit -n` in the launching shell or the service limit for `EMFILE`; for `ENFILE`, the kernel +file table, which no per-process knob raises. Nothing here reads heap pressure. + +The rest of this section is the sizing history that led there. It remains accurate about how the +caps behave when an env override sets one, and about how NOT to reason from a ledger replay. + **Cap enforcement touches BOTH pools, at two different moments.** Creating a retained head calls `evictOldestIdleGeneration('nursery', maxNurseryConnections, 'nursery_lru_cap')` first; when that head is later selected for its first continuation, `continueOnHead` calls @@ -207,7 +251,8 @@ which could be a large long-lived conversation that then resends full context. W `established_lru_cap` alongside `nursery_lru_cap`. **Pool caps are sized from peak OCCUPANCY, and a replay that reasons from eviction victim ages will -mislead you.** The two shipped caps rest on very different evidence and should be changed separately. +mislead you.** The two caps that shipped before the default became unbounded (64 established / 48 +nursery) rested on very different evidence. The established cap went 32 -> 64 on demand: this ledger's established gauge peaked at 28 against the old cap of 32, in ORGANIC traffic, and replaying it with the turns it used to isolate keeping heads of @@ -234,7 +279,7 @@ with the cap and saturates at the nursery TTL, so the criterion reduces to arriv and mostly encodes whatever retry storm dominates the sample. **What the ledger does establish, from its own gauges rather than a model:** 10 real `nursery_lru_cap` -evictions and zero `established_lru_cap` ones; the nursery victims had been idle 217-284s, already at +evictions and zero `established_lru_cap` ones; the nursery victims had been idle 217-284s, approaching the 5-minute nursery TTL, so no recorded cap eviction cost a reusable conversation. Head reuse was cap-invariant across every setting replayed, including unlimited. The case for raising a cap is headroom over concurrency, never an observed loss. @@ -278,7 +323,7 @@ identifying the cause needed `--ws-diagnostics` and a JSONL trawl. Watch `idle_m count: an eviction whose victim had been idle for minutes cost nothing, and one at a few seconds is the signal that a cap is too small. -**An empty slot is free, so treat the caps as safety valves rather than tuning knobs.** They are read +**An empty slot is free, which is why the default is no cap at all.** The caps are read only by the `>=` comparison in `evictOldestIdleGeneration` and echoed into diagnostics; nothing is preallocated and the registry is a `Map` of `Set`s sized by live entries, so unused capacity costs zero bytes and zero cycles, and eviction's sort is over actual entries. The caps also do not govern @@ -287,20 +332,15 @@ hitting one is the expensive event: a reusable conversation is discarded and its uncached prompt plus a fresh upgrade. The pool size a workload actually needs is a property of how many agents are running, and churning connections underneath that number costs more than holding them. -What a larger cap does raise is the ceiling on sockets HELD at once — 40 idle heads to 112 — each -holding its conversation plus the canonical copy memoized for prefix comparison (roughly twice the -context; 16 in-flight heads measured 11.7MB), bounded in practice by the 5- and 30-minute idle TTLs. -**File descriptors are the real ceiling and the reason not to go much higher.** 112 is under half the -256-descriptor soft limit a stock macOS shell commonly carries, and there is no `EMFILE` handling on -this path, so exceeding a user's limit is an unhandled failure rather than a degraded mode. Check that -before raising these again. The edge's own per-account connection limit is separate and not known; the -44 upgrade rejections in this ledger are the only evidence about it. -Override via `CLODEX_WS_MAX_CONNECTIONS` / -`CLODEX_WS_MAX_NURSERY_CONNECTIONS` (integer 1–1024; malformed values are logged and ignored). An -explicit programmatic option outranks the environment so tests are never perturbed. Eviction reasons +What retention does raise is the number of heads HELD at once, each holding its conversation plus, +once prefix comparison has memoized one, a canonical copy (16 in-flight heads measured 11.7MB; +~0.73 MiB per head), bounded in practice by the 5- and 30-minute idle TTLs. Memory is the ordinary +ceiling; descriptors are the backstop, handled as described above, and the reason a numeric cap is +no longer needed to stay under either. The edge's own per-account connection limit is separate and +not known; the 44 upgrade rejections in this ledger are the only evidence about it. Eviction reasons (`nursery_lru_cap`, `established_lru_cap`, `idle_ttl`, `nursery_idle_ttl`, `hard_ttl`) appear in the -`evictions` array on every `ws_head_decision` diagnostic — sustained `*_lru_cap` counts mean a cap -is too small. +`evictions` array on every `ws_head_decision` diagnostic — a `*_lru_cap` entry can only appear under +a finite cap. ### Pacing new connections diff --git a/src/oauth/responses-websocket.ts b/src/oauth/responses-websocket.ts index 405704f7..fbff35c1 100644 --- a/src/oauth/responses-websocket.ts +++ b/src/oauth/responses-websocket.ts @@ -7,6 +7,8 @@ // only after proving the next translated conversation appends to the chain head. import { createHash } from 'node:crypto'; +import { closeSync, openSync } from 'node:fs'; +import { devNull } from 'node:os'; import { AsyncLocalStorage } from 'node:async_hooks'; import type { FetchFunction } from '@ai-sdk/provider-utils'; import type { RawData, WebSocket as WsWebSocket } from 'ws'; @@ -36,48 +38,32 @@ export const RESPONSES_WS_HARD_TTL_MS = 55 * 60_000; export const RESPONSES_WS_IDLE_TTL_MS = 30 * 60_000; export const RESPONSES_WS_NURSERY_IDLE_TTL_MS = 5 * 60_000; /** - * Pool caps. An unused slot costs nothing — the caps are read only by the `>=` - * comparison in `evictOldestIdleGeneration`, nothing is preallocated, and the - * registry is sized by live entries — while hitting a cap discards a reusable - * conversation whose next turn then pays a full uncached prompt plus a fresh - * upgrade. The asymmetry is the whole argument: size these above demand, because - * unused capacity is free and a cap that binds is not. + * Pool caps: UNBOUNDED by default. The idle TTLs below are the retention + * policy; nothing else shrinks the pools unless the machine runs out of file + * descriptors, and that case is detected and handled as load shedding (see + * `shedIdleConnectionsForDescriptors`) rather than as a failure. * - * The two numbers rest on different strengths of evidence. Say so before changing - * either. + * A numeric cap was tried first (8/32, then 48/64) and every value was wrong for + * somebody: it had to be guessed per machine and per workload, and getting it + * wrong degraded SILENTLY — a cap eviction discards a reusable conversation whose + * next turn then resends its whole history uncached. The evidence that made the + * caps removable: over a 27.6-hour local ledger, head reuse was identical at + * every cap from 8 to unlimited, and the ten real cap evictions displaced heads + * idle 217-284s, approaching the 5-minute nursery TTL. * - * `maxConnections` (established) 32 -> 64 is demand-driven. In a 27.6-hour local - * ledger the established gauge peaked at 28 against the old cap of 32 — 88% of it — - * in ORGANIC traffic, and replaying that ledger with the turns it used to isolate - * keeping heads of their own puts the peak at 46. 32 was genuinely tight. - * - * `maxNurseryConnections` 8 -> 48 is a deliberately generous safety valve, NOT a - * measured requirement. Organic nursery occupancy in that ledger was 1-4 for 22 of - * 24 hours; a 16-conversation lab fan-out reached 11; the only readings near 24 came - * from a single hour of upstream auth failures and its aftermath. 48 covers a - * fan-out several times larger than anything observed, and is chosen because - * overshoot is free rather than because demand was seen at that level. - * - * DO NOT re-derive these from eviction victim ages, and distrust any replay that - * says you should. An earlier attempt did and was wrong three ways: it fed every - * head decision into the pool including ~3,880 `parallel_isolated` ones that are - * never registered in a pool at all; it never tore down heads whose request FAILED, - * though `failContext` ends in `deleteEntry`, so 84% of its eviction "victims" could - * not exist; and it compared idle time against request-start-to-request-start gaps, - * which include generation time. The check that catches all of it: that model - * predicted 231 nursery cap evictions at the cap this ledger actually ran, which - * recorded 10. Measured from completion, the real idle horizon is p50 0.1s / p90 - * 0.8s / p99 22.2s, and the 10 real cap evictions displaced heads idle 217-284s — - * at the 5-minute nursery TTL, costing nothing. - * - * Neither cap is a hard ceiling: only IDLE entries are evictable, so a generation - * can exceed its cap while every head is busy, and isolated sockets are never - * registered and never counted. The cost that scales with these numbers is memory — - * each retained head holds its conversation plus a canonical copy for prefix - * comparison — plus held descriptors, and there is no EMFILE handling on this path. + * Neither cap was a hard ceiling anyway: only IDLE entries are evictable, so a + * generation exceeded its cap while every head was busy, and isolated sockets + * are never registered and never counted. The ordinary cost of retention is + * memory — a retained head holds its conversation, plus a canonical copy once + * prefix comparison has memoized one (~0.73 MiB per head measured) — bounded by + * the pacer (60 dials/min) times the TTLs. Descriptors are the backstop, not the + * usual bound: Node raises the soft limit to the hard limit at startup, so only + * a service or container with a clamped HARD limit reaches `EMFILE`. + * `CLODEX_WS_MAX_CONNECTIONS` / `CLODEX_WS_MAX_NURSERY_CONNECTIONS` stay as an + * optional cap on the idle pool for anyone who wants one. */ -export const RESPONSES_WS_MAX_CONNECTIONS = 64; -export const RESPONSES_WS_MAX_NURSERY_CONNECTIONS = 48; +export const RESPONSES_WS_MAX_CONNECTIONS = Number.POSITIVE_INFINITY; +export const RESPONSES_WS_MAX_NURSERY_CONNECTIONS = Number.POSITIVE_INFINITY; export interface ResponsesWebSocketFetchOptions { providerId?: string; @@ -230,9 +216,9 @@ interface ConnectionEntry { // once: rewinds/branches, hidden title-generation requests, and stop hooks can // all share its model/effort/cache key. Retain each head and select by exact // conversation prefix instead of letting the newest branch replace the rest. -// New heads live in a separately capped nursery LRU until their first reuse; -// established heads therefore never consume nursery capacity, and one-shot -// nursery traffic never consumes the established LRU's reserved slots. +// New heads live in a nursery generation until their first reuse, with its own +// (shorter) idle TTL and, under an env cap, its own LRU — so one-shot nursery +// traffic never displaces established heads. const connections = new Map>(); let nextConnectionDebugId = 1; @@ -1716,6 +1702,135 @@ function evictOldestIdleGeneration( return evictions; } +/** + * The process ran out of file descriptors while opening a socket. `EMFILE` is + * the per-process limit, `ENFILE` the system-wide file table. A numeric-address + * dial reports either as the socket's `error` code — but the shipped route is a + * HOSTNAME, and a full descriptor table fails inside `getaddrinfo` first, which + * Node reports as `ENOTFOUND` with no cause. So the error code alone is not the + * detector: on any other socket-open error, `descriptorExhaustionCode` probes + * the process directly by opening one descriptor. Reproduced on macOS and + * Linux under a hard `ulimit -n 40`: `dns.lookup('chatgpt.com')` -> ENOTFOUND, + * `net.connect(port, '127.0.0.1')` -> EMFILE, and freeing one descriptor makes + * the same lookup succeed. + */ +const DESCRIPTOR_EXHAUSTION_CODES = new Set(['EMFILE', 'ENFILE']); + +function isDescriptorExhaustion(code: unknown): code is string { + return typeof code === 'string' && DESCRIPTOR_EXHAUSTION_CODES.has(code); +} + +/** + * The exhaustion code behind a socket-open failure, or undefined when the + * process can still open a descriptor. The probe costs one open/close of the + * null device and runs only on the failure path. + */ +function descriptorExhaustionCode(error: Error): string | undefined { + const code = (error as NodeJS.ErrnoException).code; + if (isDescriptorExhaustion(code)) return code; + try { + closeSync(openSync(devNull, 'r')); + return undefined; + } catch (probe) { + const probeCode = (probe as NodeJS.ErrnoException).code; + return isDescriptorExhaustion(probeCode) ? probeCode : undefined; + } +} + +let descriptorExhaustionNoticed = false; + +export function resetDescriptorExhaustionNoticeForTests(): void { + descriptorExhaustionNoticed = false; +} + +/** + * Descriptor exhaustion, handled as load-shedding rather than as a failure. + * + * With no numeric pool cap, the descriptor limit is where the pool stops + * growing — so hitting it is exactly the condition a cap eviction used to + * stand in for, now signalled by the machine instead of guessed. Every idle + * pooled head is closed, oldest first, so the transport retry that follows + * (`retryTransportFailure`, one attempt with the full context) opens its + * replacement against freed descriptors. Busy heads and isolated sockets are + * untouched: they carry a response somebody is waiting on, and closing them + * trades one failure for another. When nothing is idle there is nothing to + * shed, the retry reports the same exhaustion, and the request fails with a + * message that names the limit — bounded by the single retry, never a loop. + * + * Victims are TERMINATED, not closed. `close()` starts the WebSocket closing + * handshake and the descriptor stays open until the peer answers (or ws's + * 30-second close timeout fires); `terminate()` destroys the underlying socket, + * and Node closes the descriptor synchronously inside `uv_close`, so the + * replacement dialled in the same tick can take it. + * + * The user is told ONCE per process, on the parent-notice channel — the muted + * stderr under `clodex claude` would swallow it (see src/parent-notice.ts) — + * in terms they can act on: which limit, how many pooled connections were + * registered, and the knob. Later occurrences go to the debug log and the + * diagnostic ledger. + */ +function shedIdleConnectionsForDescriptors( + failing: ConnectionEntry, + ctx: RequestContext, + code: string, + socketErrorCode: string | undefined, +): void { + // Pooled entries other than the one whose dial just failed. Isolated sockets + // are never registered, so they are neither counted nor shed. + const registered = connectionEntries().filter(entry => entry !== failing); + const idle = registered + .filter(entry => !entry.inFlight && entry.generation !== 'isolated') + .sort((left, right) => left.lastUsedAt - right.lastUsedAt); + for (const victim of idle) { + const idleMs = Math.max(0, victim.options.now() - victim.lastUsedAt); + victim.debug( + `shedding idle ${victim.generation} connection after ${code}: ` + + `connection=${victim.debugId} idle_ms=${idleMs} reason=descriptor_exhaustion`, + ); + victim.inFlight = false; + victim.current = undefined; + unregisterEntry(victim); + try { victim.socket.terminate(); } catch { /* ignore */ } + } + failing.debug( + `${code} opening connection=${failing.debugId}: registered=${registered.length} shed=${idle.length}` + + (socketErrorCode && socketErrorCode !== code ? ` reported_as=${socketErrorCode}` : ''), + ); + emitContextDiagnostic(failing, ctx, { + event: 'ws_descriptor_exhaustion', + code, + detectedBy: socketErrorCode === code ? 'error_code' : 'descriptor_probe', + socketErrorCode: boundedDiagnosticIdentifier(socketErrorCode), + heldConnections: registered.length, + shedConnections: idle.length, + }); + if (descriptorExhaustionNoticed) return; + descriptorExhaustionNoticed = true; + emitParentNotice( + `clodex: warning: ${descriptorLimitName(code)} was reached (${code}) while opening a ChatGPT connection. ` + + `clodex had ${registered.length} pooled connection(s) registered and closed ${idle.length} idle one(s) to recover; ` + + `each parallel conversation keeps one open. ${descriptorLimitRemedy(code)} ` + + 'Further open-file warnings suppressed.', + ); +} + +function descriptorLimitName(code: string): string { + return code === 'ENFILE' ? "the system-wide open-file limit" : "this process's open-file limit"; +} + +// ENFILE is the kernel's file table, which no per-process knob raises. +function descriptorLimitRemedy(code: string): string { + return code === 'ENFILE' + ? 'Close other programs holding many files, or raise the system-wide file limit, if this recurs.' + : 'Raise it with `ulimit -n` in the shell that starts clodex (or the service limit for a ' + + 'launchd/systemd-managed server) if this recurs.'; +} + +function descriptorExhaustionMessage(code: string, registered: number): string { + return `${descriptorLimitName(code)} was reached (${code}) while opening a ChatGPT connection ` + + `with ${registered} pooled connection(s) registered; ${descriptorLimitRemedy(code)}`; +} + function isModelDataEvent(type: string | undefined): boolean { return Boolean(type && ( type.includes('.delta') @@ -2220,13 +2335,26 @@ function createConnection( socket.on('error', (error: Error) => { const ctx = entry.current; if (ctx) { + const socketErrorCode = (error as NodeJS.ErrnoException).code; const details = { source: 'socket_error', socketErrorName: boundedDiagnosticIdentifier(error.name), - socketErrorCode: boundedDiagnosticIdentifier((error as NodeJS.ErrnoException).code), + socketErrorCode: boundedDiagnosticIdentifier(socketErrorCode), ...diagnosticTextFingerprint('errorMessage', error.message), }; - handleTransportFailure(entry, ctx, error.message, details); + // Shed BEFORE the transport retry below, so the replacement it dials + // finds descriptors free. The retry itself is the ordinary one-shot path. + // Only a dial can be starved of a descriptor: a socket that is already + // open holds its own, and an error there is not descriptor pressure this + // request can recover from by shedding (no replay after output either). + let message = error.message; + const exhaustion = entry.open ? undefined : descriptorExhaustionCode(error); + if (exhaustion) { + const registered = connectionEntries().filter(other => other !== entry).length; + message = descriptorExhaustionMessage(exhaustion, registered); + shedIdleConnectionsForDescriptors(entry, ctx, exhaustion, socketErrorCode); + } + handleTransportFailure(entry, ctx, message, details); } else deleteEntry(entry); }); socket.on('close', (code: number, reason: Buffer) => { @@ -2248,29 +2376,34 @@ function createConnection( return entry; } +function diagnosticCap(cap: number): number | null { + return Number.isFinite(cap) ? cap : null; +} + /** - * Build a fetch transport backed by persistent, session-aware Responses sockets. - * Each returned Response still represents exactly one AI SDK request. - */ -/** - * Reads a connection-pool cap from the environment. + * Reads a connection-pool cap from the environment — an optional cap on the + * idle pool, now that the shipped default is unbounded. * - * Both pools are process-wide, so a workload that fans out into many concurrent - * subagent conversations can evict heads before their next turn arrives. An - * explicit option still wins, so tests are never perturbed by a stray variable. - * A malformed value is reported and ignored rather than silently reinterpreted. + * Both pools are process-wide, so a bound set here evicts heads across every + * conversation the process serves. An explicit option still wins, so tests are + * never perturbed by a stray variable. A malformed value is reported and + * ignored rather than silently reinterpreted. */ function envConnectionCap(name: string, log?: (message: string) => void): number | undefined { const raw = process.env[name]; if (raw === undefined || raw.trim() === '') return undefined; const value = Number(raw.trim()); - if (!Number.isInteger(value) || value < 1 || value > 1024) { - try { log?.(`ws: ignoring ${name}=${raw} (expected an integer between 1 and 1024)`); } catch { /* ignore */ } + if (!Number.isInteger(value) || value < 1) { + try { log?.(`ws: ignoring ${name}=${raw} (expected a positive integer)`); } catch { /* ignore */ } return undefined; } return value; } +/** + * Build a fetch transport backed by persistent, session-aware Responses sockets. + * Each returned Response still represents exactly one AI SDK request. + */ export function createResponsesWebSocketFetch( wsUrl: string, log?: (message: string) => void, @@ -2715,8 +2848,10 @@ export function createResponsesWebSocketFetch( activeConnectionCount: connectionCount(), nurseryConnectionCount: connectionCountByGeneration('nursery'), establishedConnectionCount: connectionCountByGeneration('established'), - maxConnections: resolvedOptions.maxConnections, - maxNurseryConnections: resolvedOptions.maxNurseryConnections, + // `null` is unbounded, the shipped default. A number is a finite cap on + // the idle pool from an env or programmatic override. + maxConnections: diagnosticCap(resolvedOptions.maxConnections), + maxNurseryConnections: diagnosticCap(resolvedOptions.maxNurseryConnections), selectedConnectionId: selected?.debugId, selectedGeneration: selected?.generation, continuationMatchMode: selectedMatch?.mode, diff --git a/tests/responses-websocket.test.ts b/tests/responses-websocket.test.ts index 5027b7a5..7d97aa04 100644 --- a/tests/responses-websocket.test.ts +++ b/tests/responses-websocket.test.ts @@ -11,6 +11,7 @@ class FakeWebSocket extends EventEmitter { options: { headers?: Record }; send = vi.fn(); close = vi.fn(); + terminate = vi.fn(); constructor(url: string, options: { headers?: Record }) { super(); this.url = url; @@ -21,9 +22,26 @@ class FakeWebSocket extends EventEmitter { vi.mock('ws', () => ({ WebSocket: FakeWebSocket, default: FakeWebSocket })); +// The descriptor probe opens the null device; tests make that throw to stand in +// for a full descriptor table without lowering the test runner's own limit. +const descriptorProbe = vi.hoisted(() => ({ failWith: undefined as string | undefined })); +vi.mock('node:fs', async importOriginal => { + const actual = await importOriginal(); + return { + ...actual, + openSync: ((...args: Parameters) => { + if (descriptorProbe.failWith) { + throw Object.assign(new Error(`${descriptorProbe.failWith}: too many open files`), { code: descriptorProbe.failWith }); + } + return actual.openSync(...args); + }) as typeof actual.openSync, + }; +}); + import { installParentNoticeSink } from '../src/parent-notice.js'; import { createResponsesWebSocketFetch, + resetDescriptorExhaustionNoticeForTests, resetReasoningGapWarningsForTests, resetToolArgumentGapWarningsForTests, resetResponsesWebSocketConnectionsForTests, @@ -5094,10 +5112,11 @@ describe('createResponsesWebSocketFetch', () => { (diagnostics.filter(event => event.event === 'ws_head_decision').at(-1)!.evictions ?? []) as Record[]; - it('reports the shipped pool caps when nothing overrides them', async () => { + it('reports the shipped pools as unbounded when nothing overrides them', async () => { // The defaults are the deliverable of the sizing change, and they reach behaviour // only through option resolution — so read them back off a decision made by a // fetch constructed the way production constructs one, not off the constants. + // `null` is the unbounded default: the descriptor limit is the ceiling. // The env overrides must be cleared: a developer who exports them for their own // server would otherwise have this test confirm THEIR caps as the shipped ones. const saved = [ @@ -5122,7 +5141,7 @@ describe('createResponsesWebSocketFetch', () => { emitTextResponse(lastSocket(), 'resp_default_caps', 'ok'); await readAll(response); expect(diagnostics.filter(event => event.event === 'ws_head_decision').at(-1)) - .toMatchObject({ maxConnections: 64, maxNurseryConnections: 48 }); + .toMatchObject({ maxConnections: null, maxNurseryConnections: null }); } finally { if (saved[0] !== undefined) process.env.CLODEX_WS_MAX_CONNECTIONS = saved[0]; if (saved[1] !== undefined) process.env.CLODEX_WS_MAX_NURSERY_CONNECTIONS = saved[1]; @@ -6747,3 +6766,434 @@ describe('new-connection pacing', () => { } }); }); + +describe('descriptor exhaustion', () => { + beforeEach(() => { + resetResponsesWebSocketConnectionsForTests(); + resetDescriptorExhaustionNoticeForTests(); + descriptorProbe.failWith = undefined; + fakeSockets.length = 0; + }); + + /** Admits every request at once; the shared pacer would queue a large fan-out. */ + const instantPacer = () => ({ admit: vi.fn(async () => ({ kind: 'admitted' as const, waitedMs: 0 })) }); + + const rootPayload = (text: string, sessionId: string) => sessionPayload( + [{ role: 'user', content: [{ type: 'input_text', text }] }], + { prompt_cache_key: sessionId }, + ); + + /** Opens one head per session id and completes its turn, leaving it idle. */ + async function openIdleHeads( + wsFetch: ReturnType, + sessionIds: string[], + ): Promise { + const sockets: FakeWebSocket[] = []; + for (const sessionId of sessionIds) { + const response = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload(`root ${sessionId}`, sessionId)), + }); + const socket = lastSocket(); + socket.emit('open'); + emitTextResponse(socket, `resp_${sessionId}`, 'ok'); + await readAll(response); + sockets.push(socket); + } + return sockets; + } + + it('keeps more heads than the old caps allowed, with no cap eviction', async () => { + // 48 was the nursery cap this replaces; every one of these heads is a + // conversation whose next turn would otherwise resend its history uncached. + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-unbounded', + pacer: instantPacer(), + onDiagnostic: event => diagnostics.push(event), + }); + const sessionIds = Array.from({ length: 60 }, (_, index) => `session-${index}`); + const sockets = await openIdleHeads(wsFetch, sessionIds); + expect(sockets.every(socket => !socket.close.mock.calls.length)).toBe(true); + const evictions = diagnostics + .filter(event => event.event === 'ws_head_decision') + .flatMap(event => (event.evictions ?? []) as Record[]); + expect(evictions).toEqual([]); + // A decision is recorded before its own head registers, so the last one + // counts the 59 heads already held. + expect(diagnostics.at(-1)).toMatchObject({ + event: 'ws_head_decision', + nurseryConnectionCount: 59, + maxNurseryConnections: null, + }); + }); + + it('sheds idle heads on EMFILE, retries on freed descriptors, and tells the user once', async () => { + const notices: string[] = []; + const release = installParentNoticeSink(line => notices.push(line)); + try { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-emfile', + pacer: instantPacer(), + onDiagnostic: event => diagnostics.push(event), + }); + const [older, newer] = await openIdleHeads(wsFetch, ['older', 'newer']); + + const response = await withResponsesWebSocketDiagnosticContext( + { requestId: 'req-emfile' }, + () => wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('third conversation', 'third')), + }), + ); + const starved = lastSocket(); + expect(fakeSockets).toHaveLength(3); + starved.emit('error', Object.assign(new Error('connect EMFILE 1.2.3.4:443'), { code: 'EMFILE' })); + + // Idle heads are torn down hard — a graceful close keeps the descriptor + // until the peer answers — and the replacement is dialled right away. + expect(older!.terminate).toHaveBeenCalledOnce(); + expect(newer!.terminate).toHaveBeenCalledOnce(); + expect(older!.close).not.toHaveBeenCalled(); + expect(fakeSockets).toHaveLength(4); + const replacement = lastSocket(); + replacement.emit('open'); + emitTextResponse(replacement, 'resp_third', 'recovered'); + expect(await readAll(response)).toContain('recovered'); + + expect(diagnostics).toContainEqual(expect.objectContaining({ + event: 'ws_descriptor_exhaustion', + requestId: 'req-emfile', + code: 'EMFILE', + detectedBy: 'error_code', + heldConnections: 2, + shedConnections: 2, + })); + expect(diagnostics).toContainEqual(expect.objectContaining({ + event: 'ws_transport_retry', outcome: 'recovered', requestId: 'req-emfile', + })); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain("this process's open-file limit was reached (EMFILE)"); + expect(notices[0]).toContain('had 2 pooled connection(s) registered'); + expect(notices[0]).toContain('closed 2 idle'); + expect(notices[0]).toContain('ulimit -n'); + + // A later exhaustion in the same process is recorded, not re-announced. + const again = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('fourth conversation', 'fourth')), + }); + // The shed heads are gone from the pool: this decision sees only the + // recovered third head. + expect(diagnostics.filter(event => event.event === 'ws_head_decision').at(-1)) + .toMatchObject({ activeConnectionCount: 1 }); + lastSocket().emit('error', Object.assign(new Error('connect ENFILE'), { code: 'ENFILE' })); + const secondReplacement = lastSocket(); + secondReplacement.emit('open'); + emitTextResponse(secondReplacement, 'resp_fourth', 'again'); + expect(await readAll(again)).toContain('again'); + expect(diagnostics.filter(event => event.event === 'ws_descriptor_exhaustion')) + .toHaveLength(2); + expect(notices).toHaveLength(1); + } finally { + release(); + } + }); + + it('never sheds a head that is carrying a response', async () => { + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-emfile-busy', + pacer: instantPacer(), + }); + const busyResponse = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('busy conversation', 'busy')), + }); + const busySocket = lastSocket(); + busySocket.emit('open'); + // Its turn is still streaming: nothing has completed on this socket. + const [idle] = await openIdleHeads(wsFetch, ['idle']); + + const starvedResponse = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('starved conversation', 'starved')), + }); + lastSocket().emit('error', Object.assign(new Error('connect EMFILE'), { code: 'EMFILE' })); + + expect(idle!.terminate).toHaveBeenCalledOnce(); + expect(busySocket.terminate).not.toHaveBeenCalled(); + expect(busySocket.close).not.toHaveBeenCalled(); + + const replacement = lastSocket(); + replacement.emit('open'); + emitTextResponse(replacement, 'resp_starved', 'starved ok'); + expect(await readAll(starvedResponse)).toContain('starved ok'); + emitTextResponse(busySocket, 'resp_busy', 'busy ok'); + expect(await readAll(busyResponse)).toContain('busy ok'); + }); + + it('fails with an actionable message when the retry is starved too, and does not loop', async () => { + const notices: string[] = []; + const release = installParentNoticeSink(line => notices.push(line)); + try { + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-emfile-exhausted', + pacer: instantPacer(), + }); + const response = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('only conversation', 'only')), + }); + lastSocket().emit('error', Object.assign(new Error('connect EMFILE'), { code: 'EMFILE' })); + expect(fakeSockets).toHaveLength(2); + lastSocket().emit('error', Object.assign(new Error('connect EMFILE'), { code: 'EMFILE' })); + expect(fakeSockets).toHaveLength(2); + + const body = await readAll(response); + expect(body).toContain('open-file limit was reached (EMFILE)'); + expect(body).toContain('0 pooled connection(s) registered'); + expect(body).toContain('ulimit -n'); + expect(body).not.toContain('connect EMFILE'); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain('closed 0 idle'); + } finally { + release(); + } + }); + + it('leaves other socket errors on the ordinary retry path', async () => { + const notices: string[] = []; + const release = installParentNoticeSink(line => notices.push(line)); + try { + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-not-emfile', + pacer: instantPacer(), + }); + const [idle] = await openIdleHeads(wsFetch, ['idle']); + const response = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('reset conversation', 'reset')), + }); + lastSocket().emit('error', Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' })); + expect(idle!.terminate).not.toHaveBeenCalled(); + expect(notices).toEqual([]); + const replacement = lastSocket(); + replacement.emit('open'); + emitTextResponse(replacement, 'resp_reset', 'ok'); + expect(await readAll(response)).toContain('ok'); + } finally { + release(); + } + }); + + it('sheds established heads too, oldest first', async () => { + // Established heads are the dominant idle population a full descriptor + // table finds (organic peak 28 vs 1-4 nursery), and promotion happens only + // on a continuation — so this stages a real second turn. + let clock = 1_000_000; + const now = () => clock; + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-emfile-established', + pacer: instantPacer(), + now, + onDiagnostic: event => diagnostics.push(event), + }); + const firstInput = [{ role: 'user', content: [{ type: 'input_text', text: 'turn one' }] }]; + const first = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(sessionPayload(firstInput, { prompt_cache_key: 'established' })), + }); + const established = lastSocket(); + established.emit('open'); + emitTextResponse(established, 'resp_e1', 'one'); + await readAll(first); + + clock += 1_000; + const second = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(sessionPayload([ + ...firstInput, + { role: 'assistant', content: [{ type: 'output_text', text: 'one' }] }, + { role: 'user', content: [{ type: 'input_text', text: 'turn two' }] }, + ], { prompt_cache_key: 'established' })), + }); + expect(fakeSockets).toHaveLength(1); + expect(JSON.parse(established.send.mock.calls[1]![0] as string).previous_response_id).toBe('resp_e1'); + // Promotion happens when the head is selected, so the decision already + // reports the generation the shed will find. + expect(diagnostics.filter(event => event.event === 'ws_head_decision').at(-1)) + .toMatchObject({ decision: 'continuation', selectedGeneration: 'established' }); + emitTextResponse(established, 'resp_e2', 'two'); + await readAll(second); + + // A younger nursery head, used more recently than the established one. + clock += 1_000; + const [nursery] = await openIdleHeads(wsFetch, ['younger']); + + clock += 1_000; + const starved = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('starved', 'starved')), + }); + expect(diagnostics.filter(event => event.event === 'ws_head_decision').at(-1)) + .toMatchObject({ establishedConnectionCount: 1, nurseryConnectionCount: 1 }); + lastSocket().emit('error', Object.assign(new Error('connect EMFILE'), { code: 'EMFILE' })); + + expect(established.terminate).toHaveBeenCalledOnce(); + expect(nursery!.terminate).toHaveBeenCalledOnce(); + expect(established.close).not.toHaveBeenCalled(); + // Oldest first: the established head was last used before the nursery one. + expect(established.terminate.mock.invocationCallOrder[0]!) + .toBeLessThan(nursery!.terminate.mock.invocationCallOrder[0]!); + expect(diagnostics).toContainEqual(expect.objectContaining({ + event: 'ws_descriptor_exhaustion', heldConnections: 2, shedConnections: 2, + })); + const replacement = lastSocket(); + replacement.emit('open'); + emitTextResponse(replacement, 'resp_starved', 'recovered'); + expect(await readAll(starved)).toContain('recovered'); + }); + + it('detects exhaustion behind a hostname lookup failure by probing a descriptor', async () => { + // The shipped route is a hostname, and a full descriptor table fails inside + // getaddrinfo — the socket reports ENOTFOUND, not EMFILE. + const notices: string[] = []; + const release = installParentNoticeSink(line => notices.push(line)); + try { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-enotfound', + pacer: instantPacer(), + onDiagnostic: event => diagnostics.push(event), + }); + const [idle] = await openIdleHeads(wsFetch, ['idle']); + const response = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('starved', 'starved')), + }); + descriptorProbe.failWith = 'EMFILE'; + lastSocket().emit('error', Object.assign(new Error('getaddrinfo ENOTFOUND chatgpt.com'), { code: 'ENOTFOUND' })); + descriptorProbe.failWith = undefined; + + expect(idle!.terminate).toHaveBeenCalledOnce(); + expect(diagnostics).toContainEqual(expect.objectContaining({ + event: 'ws_descriptor_exhaustion', + code: 'EMFILE', + detectedBy: 'descriptor_probe', + socketErrorCode: 'ENOTFOUND', + heldConnections: 1, + shedConnections: 1, + })); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain('(EMFILE)'); + const replacement = lastSocket(); + replacement.emit('open'); + emitTextResponse(replacement, 'resp_starved', 'recovered'); + expect(await readAll(response)).toContain('recovered'); + } finally { + release(); + } + }); + + it('leaves a genuine lookup failure alone when descriptors are available', async () => { + const notices: string[] = []; + const release = installParentNoticeSink(line => notices.push(line)); + try { + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-real-enotfound', + pacer: instantPacer(), + }); + const [idle] = await openIdleHeads(wsFetch, ['idle']); + const response = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('offline', 'offline')), + }); + lastSocket().emit('error', Object.assign(new Error('getaddrinfo ENOTFOUND chatgpt.com'), { code: 'ENOTFOUND' })); + lastSocket().emit('error', Object.assign(new Error('getaddrinfo ENOTFOUND chatgpt.com'), { code: 'ENOTFOUND' })); + expect(idle!.terminate).not.toHaveBeenCalled(); + expect(notices).toEqual([]); + expect(await readAll(response)).toContain('ENOTFOUND'); + } finally { + release(); + } + }); + + it('names the system-wide limit on ENFILE and does not prescribe ulimit', async () => { + const notices: string[] = []; + const release = installParentNoticeSink(line => notices.push(line)); + try { + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-enfile', + pacer: instantPacer(), + }); + const response = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('only', 'only')), + }); + lastSocket().emit('error', Object.assign(new Error('connect ENFILE'), { code: 'ENFILE' })); + lastSocket().emit('error', Object.assign(new Error('connect ENFILE'), { code: 'ENFILE' })); + const body = await readAll(response); + expect(body).toContain('system-wide open-file limit was reached (ENFILE)'); + expect(body).not.toContain('ulimit'); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain('system-wide open-file limit was reached (ENFILE)'); + expect(notices[0]).not.toContain('ulimit'); + } finally { + release(); + } + }); + + it('honours an env cap above the old 1024 ceiling instead of ignoring it', async () => { + process.env.CLODEX_WS_MAX_NURSERY_CONNECTIONS = '2000'; + try { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-big-cap', + onDiagnostic: event => diagnostics.push(event), + }); + await openIdleHeads(wsFetch, ['one']); + expect(diagnostics.filter(event => event.event === 'ws_head_decision').at(-1)) + .toMatchObject({ maxNurseryConnections: 2000 }); + } finally { + delete process.env.CLODEX_WS_MAX_NURSERY_CONNECTIONS; + } + }); + + it('does not shed on an error from a socket that is already open', async () => { + // An open socket holds its own descriptor; a mid-stream failure there is not + // a starved dial, and after output there is no retry to benefit from a shed. + const notices: string[] = []; + const release = installParentNoticeSink(line => notices.push(line)); + try { + const wsFetch = createResponsesWebSocketFetch(WS_URL, undefined, { + accountId: 'acct-open-error', + pacer: instantPacer(), + }); + const [idle] = await openIdleHeads(wsFetch, ['idle']); + const response = await wsFetch('https://x', { + method: 'POST', headers: {}, + body: JSON.stringify(rootPayload('streaming', 'streaming')), + }); + const streaming = lastSocket(); + streaming.emit('open'); + streaming.emit('message', Buffer.from(JSON.stringify({ type: 'response.created', response: { id: 'resp_s' } }))); + streaming.emit('message', Buffer.from(JSON.stringify({ + type: 'response.output_item.added', output_index: 0, item: { type: 'message', id: 'msg_s' }, + }))); + streaming.emit('message', Buffer.from(JSON.stringify({ + type: 'response.output_text.delta', item_id: 'msg_s', delta: 'partial', + }))); + descriptorProbe.failWith = 'EMFILE'; + streaming.emit('error', Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' })); + descriptorProbe.failWith = undefined; + expect(idle!.terminate).not.toHaveBeenCalled(); + expect(notices).toEqual([]); + expect(fakeSockets).toHaveLength(2); + expect(await readAll(response)).toContain('ECONNRESET'); + } finally { + release(); + } + }); +});