diff --git a/CHANGELOG.md b/CHANGELOG.md index 9514711..6387c99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,9 @@ All notable changes to OLP land here. Per `CLAUDE.md` release_kit overlay, this ## Unreleased -(empty — Phase 2 entries land here once Phase 2 opens) +### D38 — maxConcurrent runtime enforcement (issue #1) + +- **Spawn lifecycle gate** — `hints.maxConcurrent` is now enforced at runtime per ADR 0002 Amendment 6: `lib/providers/index.mjs` exports a per-provider `tryAcquireSpawn` / `releaseSpawn` / `getActiveSpawnCount` semaphore; `server.mjs` gates both the buffered and streaming spawn call sites in `handleChatCompletions` with a try/finally release. Saturation surfaces as `ProviderError(CONCURRENCY_LIMIT)` which the fallback engine treats as a hard trigger (ADR 0004 Amendment 4) — the chain advances to the next hop instead of queueing. If the entire chain is saturated, the user receives a chain-exhausted error via the existing exhaustion path. Closes #1. Queue+timeout deferred (see ADR 0002 Amendment 6 § Design choice). Test count 431 → 447. ## v0.1.0 — 2026-05-24 diff --git a/docs/adr/0002-plugin-architecture.md b/docs/adr/0002-plugin-architecture.md index 58e19b0..2eff2d2 100644 --- a/docs/adr/0002-plugin-architecture.md +++ b/docs/adr/0002-plugin-architecture.md @@ -7,6 +7,30 @@ ## Amendments +### Amendment 6 — 2026-05-24: `maxConcurrent` runtime enforcement landed (D38, issue #1) + +- **Finding:** Amendment 1 (2026-05-23) ratified `maxSpawnTimeMs` into the Provider contract but explicitly noted that `hints.maxConcurrent` remained **declarative-only at v0.1** — type-validated at startup in `lib/providers/base.mjs` (`validateProvider` requires it to be a non-negative integer) but unenforced at runtime (no semaphore / in-flight counter / spawn queue in `server.mjs`). Cold-audit catch from D11 (commit `f659e29`): the diff-review reviewer grep-verified that the original ADR draft's claim "Enforced by the spawn-concurrency guard in `server.mjs`" was false. GitHub issue #1 was filed to track the gap. D38 closes that gap. +- **Change (D38):** + - Add a per-provider in-flight semaphore in `lib/providers/index.mjs` exporting three primitives plus a constant: + - `tryAcquireSpawn(providerName, maxConcurrent)` — atomic check-then-increment; returns `true` on success, `false` if at limit. Atomicity rests on the JS single-threaded invariant — the read and write are synchronous with NO `await` between them. A future async refactor MUST preserve this. + - `releaseSpawn(providerName)` — decrement; throws if the count would go negative (defensive bug guard for missing acquire / double release). + - `getActiveSpawnCount(providerName)` — returns current in-flight count; exported for diagnostics and tests (server.mjs uses it to populate the `activeSpawns` field on a synthesised `CONCURRENCY_LIMIT` error). `/health` integration deferred — when surfaced there it will land at `providers.status..activeSpawns`; not wired at D38. + - `DEFAULT_MAX_CONCURRENT_SPAWNS = 4` — defense-in-depth fallback when a plugin path bypasses `validateProvider` and passes undefined/null/NaN. The value matches the v0.1 plugin defaults (anthropic / codex / mistral all declare `hints.maxConcurrent: 4`). + - Wire the gate at both `provider.spawn(...)` call sites in `server.mjs handleChatCompletions`: + - **Buffered path** (inside `executeHopFn → collectAllChunks`): `tryAcquireSpawn` runs before `provider.spawn(...)`. On failure, synthesise `ProviderError(CONCURRENCY_LIMIT)` with `providerName` / `maxConcurrent` / `activeSpawns` fields for diagnostics and re-throw — the fallback engine treats it as a hard trigger (see ADR 0004 Amendment 4) and advances to the next chain hop. On success, the spawn drain loop runs inside a `try { … } finally { releaseSpawn(...) }` so the slot releases on every exit path (success, error, D16 SPAWN_FAILED salvage return, unexpected throw). + - **Streaming path** (single-hop real-SSE, `chain.length === 1` cache-miss branch): acquire happens BEFORE the streaming branch entry. If acquire fails, the branch is skipped and the request falls through to the buffered path — that path's own gate re-attempts acquire; a single-hop chain at maxConcurrent has no other hop to advance to, so the request surfaces a chain-exhausted error via `executeWithFallback`'s exhaustion path. If acquire succeeds, the existing streaming try/catch gains a `finally { releaseSpawn(streamProvider) }` so the slot releases on stop-chunk completion, generator exhaustion, abort, or any exception path. + - Add `CONCURRENCY_LIMIT` to `PROVIDER_ERROR_CODES` in `lib/providers/base.mjs` so the synthesised error type-checks with the existing closed enum. (Note: `CONCURRENCY_LIMIT` is synthesised by the orchestration layer, NOT thrown by provider plugins themselves — the code is in the enum for type consistency with the fallback engine's `HARD_TRIGGER_CODES` lookup.) + - Update the `maxConcurrent` description in § Decision (Provider contract hints) below — remove the "Declarative hint only at v0.1" caveat and add the implementation reference. +- **Update to § Decision § Provider contract hints (`maxConcurrent`):** replace the v0.1 caveat with: "`maxConcurrent` — integer; maximum simultaneous spawn count OLP will allow for this provider. The value is type-validated at startup (`lib/providers/base.mjs validateProvider`) and enforced at runtime by `tryAcquireSpawn` / `releaseSpawn` in `lib/providers/index.mjs`. Saturation surfaces as `ProviderError(CONCURRENCY_LIMIT)` (per `PROVIDER_ERROR_CODES`, `lib/providers/base.mjs`), which the fallback engine treats as a hard trigger per ADR 0004 Amendment 4 — the chain advances to the next hop. If the entire chain is saturated, the user receives a chain-exhausted error via the existing `executeWithFallback` exhaustion path." +- **Design choice — immediate-advancement vs. queue+timeout:** D38 implements immediate-advancement through the fallback chain. Rationale: + 1. The fallback chain exists precisely for this kind of overflow — saturation on the primary hop is a natural fit for the existing advancement mechanism. + 2. Queue+timeout introduces head-of-line blocking risk (a stuck/slow spawn blocks queued waiters) and adds a new timeout config surface (`hints.maxConcurrentWaitMs`?) that the contract does not currently have. + 3. Immediate-advancement gives fail-fast latency and matches the OLP multi-provider proxy philosophy (the user has spread their quota across providers explicitly so saturation should reach an alternate provider as fast as possible). + 4. Queue+timeout is **deferred to a future iteration** if real usage shows demand. Track via a follow-up issue if the design pressure surfaces. +- **Authority:** ALIGNMENT.md Rule 1 (Cite First) — internal authority is ADR 0002 (this ADR) + ADR 0004 (which adds CONCURRENCY_LIMIT to the hard-trigger taxonomy in its Amendment 4). No provider CLI doc cited because this change is internal to the orchestration layer; no provider plugin code changes (anthropic / codex / mistral already declare `hints.maxConcurrent` correctly per validateProvider). +- **Tests:** Suite 18 in `test-features.mjs` — 16 tests covering: `PROVIDER_ERROR_CODES` membership, `evaluateHardTriggers(CONCURRENCY_LIMIT)` returns true, semaphore unit behaviour (acquire / release / count / reset), saturation rejection, defensive coercion of non-integer maxConcurrent, double-release throws, HTTP-level concurrent-request peak-in-flight assertion (5 requests against maxConcurrent:2 → peak == 2), buffered-path counter release, streaming-path counter release, fallback advancement to secondary on saturated primary. +- **Procedural mechanism:** CC 开发铁律 v1.6 § 10.x (D38 issue #1 implementation; fresh-context opus reviewer to follow this implementation per Iron Rule 10). + ### Amendment 5 — 2026-05-24: Correct § Decision filesystem layout — `vibe.mjs` → `mistral.mjs` (D36 #5) - **Finding:** Issue #5 (D36) — § Decision filesystem layout (around line 47 of the original ADR) listed the Mistral provider plugin as `vibe.mjs` (named after the CLI binary `vibe`). The shipped file at `lib/providers/mistral.mjs` (D8) is named after the provider key, matching the established convention from the other two plugins: `anthropic.mjs` (provider key `anthropic`, CLI `claude`) and `codex.mjs` (provider key `openai`, CLI `codex`). The ADR's `vibe.mjs` entry was a drafting-time placeholder that did not get corrected when D8 landed `lib/providers/mistral.mjs`. @@ -94,7 +118,7 @@ Every provider plugin exports an object conforming to: - `hints: { requiresTTY, concurrentSpawnSafe, maxConcurrent, maxSpawnTimeMs, cacheable }` — fingerprint, concurrency, timeout, and cache hints: - `requiresTTY` — boolean; whether the provider CLI requires a TTY to produce non-interactive output (e.g., some CLIs suppress JSON output unless forced with a flag or a TTY is present). - `concurrentSpawnSafe` — boolean; whether the provider CLI is safe to spawn concurrently under the same auth context without rate-limit or session collisions. - - `maxConcurrent` — integer; maximum simultaneous spawn count OLP will allow for this provider. **Declarative hint only at v0.1**: the value is type-validated at startup (`lib/providers/base.mjs`) but no runtime enforcement (semaphore / in-flight counter / spawn queue) is wired in `server.mjs` yet. Tracking issue to be filed for a follow-up that lands the runtime guard. + - `maxConcurrent` — integer; maximum simultaneous spawn count OLP will allow for this provider. The value is type-validated at startup (`lib/providers/base.mjs validateProvider`) and **enforced at runtime** by `tryAcquireSpawn` / `releaseSpawn` in `lib/providers/index.mjs` (D38 — see Amendment 6). Saturation surfaces as `ProviderError(CONCURRENCY_LIMIT)`, which the fallback engine treats as a hard trigger per ADR 0004 Amendment 4 — the chain advances to the next hop. If the entire chain is saturated, the user receives a chain-exhausted error via the existing `executeWithFallback` exhaustion path. (Pre-D38 caveat removed; tracking issue #1 closed by Amendment 6.) - `maxSpawnTimeMs` — optional integer, milliseconds; maximum wall-clock time OLP allows for a single provider spawn before treating it as a hard fallback trigger. Defaults to `600000` (10 minutes) if absent. Enforcement lives inside each provider plugin's spawn drain loop (`_spawnAndStream`), which uses a `setTimeout` / `proc.kill` / `reject` pattern to throw `ProviderError(SPAWN_TIMEOUT)`; the fallback engine then treats this error as a hard trigger (ADR 0004 § Trigger taxonomy — Hard triggers bullet 4). The engine itself does not run the timer loop; it only acts on the thrown error. - `cacheable` — optional boolean, default `true`; if explicitly set to `false`, the provider opts out of OLP's response cache entirely. `executeHopFn` skips `cacheStore.getOrCompute` and calls `collectAllChunks` directly; no cache read or write occurs for any request to this provider. Omitting the field is equivalent to `cacheable: true`. See ADR 0005 § "Cache write conditions" item 3 and Amendment 3 above. (D23) diff --git a/docs/adr/0004-fallback-engine.md b/docs/adr/0004-fallback-engine.md index f1a9daf..a27e922 100644 --- a/docs/adr/0004-fallback-engine.md +++ b/docs/adr/0004-fallback-engine.md @@ -7,6 +7,23 @@ ## Amendments +### Amendment 4 — 2026-05-24: Add `CONCURRENCY_LIMIT` to v0.1 hard-trigger code taxonomy (D38, issue #1) + +- **Finding:** ADR 0002 Amendment 1 (2026-05-23) ratified `hints.maxConcurrent` into the Provider contract as **declarative-only at v0.1** — no runtime enforcement. GitHub issue #1 tracked the gap. D38 lands runtime enforcement (see ADR 0002 Amendment 6 for the implementation details and design rationale). Once a saturation event occurs, the orchestration layer must communicate "this hop is at capacity — advance the chain" to the fallback engine using a code that fits the existing hard-trigger taxonomy in `evaluateHardTriggers`. +- **Change (D38):** Add `CONCURRENCY_LIMIT` to both: + - `PROVIDER_ERROR_CODES` in `lib/providers/base.mjs` (closed enum used by `ProviderError`). + - `HARD_TRIGGER_CODES` in `lib/fallback/engine.mjs` — value `true` so `evaluateHardTriggers(ProviderError CONCURRENCY_LIMIT)` returns `true` and `classifyTrigger` returns `'hard'`. The chain advances to the next hop. The synthesised error carries diagnostic fields (`providerName`, `maxConcurrent`, `activeSpawns`) which surface in the existing `fallback_hard_trigger` log event via the `error.message` field. +- **v0.1 live hard-trigger codes after this amendment (5 codes):** `SPAWN_FAILED`, `CLI_NOT_FOUND`, `SPAWN_TIMEOUT`, `CONCURRENCY_LIMIT`, plus the explicit non-trigger `AUTH_MISSING:false`. Pre-D38 list (per Amendment 3) was 4 codes (`SPAWN_FAILED`, `CLI_NOT_FOUND`, `SPAWN_TIMEOUT` as triggers + `AUTH_MISSING:false`). +- **Synthesis vs. plugin-thrown:** Unlike the other live hard-trigger codes, `CONCURRENCY_LIMIT` is **NOT thrown by provider plugins themselves**. It is synthesised by `server.mjs handleChatCompletions` when `tryAcquireSpawn(provider, hints.maxConcurrent)` returns false. The code lives in `PROVIDER_ERROR_CODES` for type consistency with the closed enum that `HARD_TRIGGER_CODES` keys on; the orchestration layer is the only callsite that throws it. A future provider plugin that gains its own internal concurrency limit (e.g., a CLI that returns a specific exit code on rate-limit) could thrown this code too; the enum is forward-compatible. +- **Design choice — immediate-advancement vs. queue+timeout:** Re-stating from ADR 0002 Amendment 6 because this ADR governs the trigger taxonomy that surfaces the decision to the user: saturation is treated as a **hard trigger** (chain advances immediately) rather than as a **soft trigger** (would gate before spawn but would not advance after spawn attempt) or as a queueable condition (would block + timeout). The hard-trigger framing matches "the primary hop refused to serve this request; advance" semantics. Queue+timeout would require a NEW trigger category outside the existing taxonomy (hard / soft / deterministic-deferred / cost-aware-deferred) and is deferred per ADR 0002 Amendment 6 rationale. +- **First-chunk safety:** `tryAcquireSpawn` runs **before** `provider.spawn(...)` and before any bytes are written to the response. A `CONCURRENCY_LIMIT` rejection therefore satisfies the first-chunk rule trivially — zero bytes have been emitted to the client. Fallback is safe. +- **Authority:** ADR 0002 Amendment 6 (runtime enforcement implementation); GitHub issue #1 (tracking). +- **Tests:** Suite 18 in `test-features.mjs` — see ADR 0002 Amendment 6 § Tests for the full list. Specifically for this ADR: tests 18a (PROVIDER_ERROR_CODES membership), 18b (`evaluateHardTriggers(CONCURRENCY_LIMIT) === true`), 18c (AUTH_MISSING regression guard — D38 did not flip it), 18k (chain advances to fallback hop on saturated primary). +- **v1.x re-evaluation triggers:** + - If a future plugin gains a CLI-level concurrency response that should NOT be a hard trigger (e.g., "soft limit hit, retry after backoff") — file a follow-up to add a new code (e.g., `CONCURRENCY_BACKOFF`) rather than reclassifying `CONCURRENCY_LIMIT`. + - If queue+timeout becomes desirable (real usage shows fail-fast advancement is too aggressive for certain workloads), file an amendment to this ADR adding queue semantics as a NEW trigger category — do not reclassify CONCURRENCY_LIMIT into the existing taxonomy. +- **Procedural mechanism:** CC 开发铁律 v1.6 § 10.x (D38 issue #1 implementation; fresh-context opus reviewer to follow per Iron Rule 10). + ### Amendment 3 — 2026-05-24: Narrow v0.1 hard-trigger code taxonomy (D34 F7) - **Finding:** Round-6 cold-audit F7 (P2) — `PROVIDER_ERROR_CODES` in `lib/providers/base.mjs` and `HARD_TRIGGER_CODES` in `lib/fallback/engine.mjs` both listed `QUOTA_EXHAUSTED` and `RATE_LIMITED` as live hard-trigger codes. No v0.1 plugin emits either code. The Anthropic, Codex, and Mistral plugins all use `claude -p`, `codex exec --json`, and `vibe --prompt` respectively — none parse the underlying-API HTTP response status code or surface a structured quota/rate error; they only throw `SPAWN_FAILED`, `SPAWN_TIMEOUT`, `CLI_NOT_FOUND`, or `AUTH_MISSING`. The two `Hard triggers` bullets in § Trigger taxonomy ("HTTP 5xx from provider's underlying API" and "HTTP 4xx quota exhaustion") are therefore unreachable through the `ProviderError` code path at v0.1. @@ -15,7 +32,7 @@ - `QUOTA_EXHAUSTED` and `RATE_LIMITED` removed from `PROVIDER_ERROR_CODES` (base.mjs) and `HARD_TRIGGER_CODES` (engine.mjs). Dead code removal. - A comment block added in `evaluateHardTriggers` labeling the HTTP-status branches as "forward-compat reserved — v0.1 plugins never attach statusCode." - Test coverage: the two unit tests for `evaluateHardTriggers: ProviderError QUOTA_EXHAUSTED/RATE_LIMITED → fires` are removed (tombstoned with a removal comment). All other hard-trigger tests that used these codes as convenient test vectors are rewritten to use `SPAWN_FAILED` / `SPAWN_TIMEOUT`. -- **v0.1 live hard-trigger codes after this amendment:** `SPAWN_FAILED`, `CLI_NOT_FOUND`, `SPAWN_TIMEOUT`. `AUTH_MISSING` remains in the table as an explicit `false` entry (deliberate non-trigger per ADR 0004 § "No fallback for client-side errors" analogue). +- **v0.1 live hard-trigger codes after this amendment:** `SPAWN_FAILED`, `CLI_NOT_FOUND`, `SPAWN_TIMEOUT`. `AUTH_MISSING` remains in the table as an explicit `false` entry (deliberate non-trigger per ADR 0004 § "No fallback for client-side errors" analogue). **Subsequently extended by Amendment 4 (D38) — `CONCURRENCY_LIMIT` added as a 4th true entry; the v0.1 live-codes list as of D38 is `SPAWN_FAILED`, `CLI_NOT_FOUND`, `SPAWN_TIMEOUT`, `CONCURRENCY_LIMIT` plus the explicit `AUTH_MISSING:false` non-trigger entry.** - **v1.x re-activation path:** When a plugin gains HTTP-status parsing (e.g., an Anthropic plugin variant that makes direct Messages API calls rather than spawning `claude -p`), add the plugin-layer HTTP parsing, re-add `QUOTA_EXHAUSTED` and `RATE_LIMITED` to both tables, and amend this entry. The `evaluateHardTriggers` HTTP-status branches will then activate naturally with no further engine changes. - **Procedural mechanism:** CC 开发铁律 v1.6 § 10.x (Round-6 Cold Audit). diff --git a/lib/fallback/engine.mjs b/lib/fallback/engine.mjs index ad0cf3d..e68c20c 100644 --- a/lib/fallback/engine.mjs +++ b/lib/fallback/engine.mjs @@ -27,11 +27,16 @@ import { computeIRRequestHash } from '../cache/keys.mjs'; /** * Maps ProviderError codes to hard-trigger decisions. * - * v0.1 live codes (per ADR 0004 Amendment 3, D34 F7): - * - SPAWN_FAILED → hard trigger (provider CLI failed) - * - CLI_NOT_FOUND → hard trigger (binary missing) - * - AUTH_MISSING → NOT a hard trigger (user-config failure; user must fix) - * - SPAWN_TIMEOUT → hard trigger (per ADR 0004 § Trigger taxonomy bullet 4) + * v0.1 live codes (per ADR 0004 Amendment 3 (D34 F7) + Amendment 4 (D38)): + * - SPAWN_FAILED → hard trigger (provider CLI failed) + * - CLI_NOT_FOUND → hard trigger (binary missing) + * - AUTH_MISSING → NOT a hard trigger (user-config failure; user must fix) + * - SPAWN_TIMEOUT → hard trigger (per ADR 0004 § Trigger taxonomy bullet 4) + * - CONCURRENCY_LIMIT → hard trigger (D38 / issue #1, ADR 0004 Amendment 4): + * synthesized by server.mjs when a provider is at its + * hints.maxConcurrent in-flight limit. The chain + * advances immediately to the next hop instead of + * queueing — design rationale per ADR 0004 Amendment 4. * * QUOTA_EXHAUSTED and RATE_LIMITED removed (D34 F7 / ADR 0004 Amendment 3): * no v0.1 plugin parses underlying-API HTTP status codes, so these codes @@ -44,8 +49,9 @@ import { computeIRRequestHash } from '../cache/keys.mjs'; const HARD_TRIGGER_CODES = { SPAWN_FAILED: true, CLI_NOT_FOUND: true, - AUTH_MISSING: false, // user config problem — never fall over (ADR 0004 § Decision) - SPAWN_TIMEOUT: true, // ADR 0004 § Trigger taxonomy bullet 4: spawn timeout is a hard trigger + AUTH_MISSING: false, // user config problem — never fall over (ADR 0004 § Decision) + SPAWN_TIMEOUT: true, // ADR 0004 § Trigger taxonomy bullet 4: spawn timeout is a hard trigger + CONCURRENCY_LIMIT: true, // ADR 0004 Amendment 4 (D38, issue #1): saturation → advance chain }; /** diff --git a/lib/providers/base.mjs b/lib/providers/base.mjs index a33d4ae..bccbed6 100644 --- a/lib/providers/base.mjs +++ b/lib/providers/base.mjs @@ -141,8 +141,15 @@ export function validateProvider(p) { /** * Error codes surfaced by provider plugins. * - * v0.1 live codes (per ADR 0004 Amendment 3, D34 F7): - * SPAWN_FAILED, CLI_NOT_FOUND, AUTH_MISSING, SPAWN_TIMEOUT + * v0.1 live codes (per ADR 0004 Amendment 3, D34 F7 and Amendment 4, D38): + * SPAWN_FAILED, CLI_NOT_FOUND, AUTH_MISSING, SPAWN_TIMEOUT, CONCURRENCY_LIMIT + * + * CONCURRENCY_LIMIT (D38, issue #1): synthesized by the orchestration layer + * (NOT thrown by provider plugins themselves) when a spawn is attempted + * against a provider already at its hints.maxConcurrent limit. The fallback + * engine treats this as a hard trigger so the chain advances to the next hop + * rather than queueing. See ADR 0002 Amendment 6 (runtime enforcement) and + * ADR 0004 Amendment 4 (CONCURRENCY_LIMIT in hard-trigger taxonomy). * * QUOTA_EXHAUSTED and RATE_LIMITED were removed (D34 F7): no plugin parses * underlying-API HTTP status codes at v0.1, so these codes are never emitted. @@ -153,7 +160,8 @@ export const PROVIDER_ERROR_CODES = /** @type {const} */ ([ 'AUTH_MISSING', 'CLI_NOT_FOUND', 'SPAWN_FAILED', - 'SPAWN_TIMEOUT', // ADR 0004 § Trigger taxonomy bullet 4: spawn timeout is a hard trigger + 'SPAWN_TIMEOUT', // ADR 0004 § Trigger taxonomy bullet 4: spawn timeout is a hard trigger + 'CONCURRENCY_LIMIT', // ADR 0002 Amendment 6 / ADR 0004 Amendment 4 (D38, issue #1) ]); export class ProviderError extends Error { diff --git a/lib/providers/index.mjs b/lib/providers/index.mjs index d831216..3ae8196 100644 --- a/lib/providers/index.mjs +++ b/lib/providers/index.mjs @@ -209,3 +209,133 @@ export function getProviderByName(loadedProviders, name) { export function listAllProviderNames() { return STATIC_REGISTRY.map(p => p.name); } + +// ── Concurrency semaphore (D38, issue #1) ───────────────────────────────── +// +// Authority: ADR 0002 Amendment 6 (maxConcurrent runtime enforcement landed in D38) +// and ADR 0004 Amendment 4 (CONCURRENCY_LIMIT added to hard-trigger taxonomy). +// +// Per-provider in-flight spawn counter. The orchestration layer (server.mjs +// handleChatCompletions) calls tryAcquireSpawn() before provider.spawn() and +// releaseSpawn() after spawn lifecycle completion. On saturation, the caller +// synthesises a ProviderError(CONCURRENCY_LIMIT) which the fallback engine +// treats as a hard trigger — the chain advances to the next hop. If the +// entire chain is saturated, the user receives a chain-exhausted error via +// the existing executeWithFallback exhaustion path. +// +// Design decision (deliberate): immediate-advancement via fallback, NOT +// queue+timeout. Rationale (per D38 issue #1 design discussion): +// 1. The fallback chain exists precisely for this kind of overflow. +// 2. A queue introduces head-of-line blocking + a new timeout config surface. +// 3. Immediate-advancement gives fail-fast latency, matching the OLP +// multi-provider proxy philosophy. +// 4. Queue+timeout is deferred — track via a future issue if real usage +// shows need. +// +// **Atomicity invariant**: JavaScript is single-threaded; the +// read-then-write pair inside tryAcquireSpawn() executes synchronously with +// NO `await` between the check and the increment. This is the only reason +// the semaphore is correct without a Mutex. A future async refactor MUST +// preserve this — do NOT introduce an `await` between the limit check and +// the count update or the semaphore loses its mutual-exclusion guarantee +// (two callers could each read count=limit-1 before either increments). +// +// Module-level state: lives for the process lifetime; tests that need +// isolation should call __resetSpawnCounters() in their teardown. +// +// @type {Map} provider name → current in-flight spawn count +const _activeSpawns = new Map(); + +/** + * Default cap for tryAcquireSpawn when a plugin omits hints.maxConcurrent. + * + * validateProvider in base.mjs requires hints.maxConcurrent to be a + * non-negative integer at startup, so a missing value should not happen in + * production. This default is defense-in-depth for callers that pass a + * stripped-down provider stub (e.g., in tests) or future plugin paths that + * bypass validation. The value (4) matches the v0.1 plugin defaults + * (anthropic / codex / mistral all declare hints.maxConcurrent: 4). + */ +export const DEFAULT_MAX_CONCURRENT_SPAWNS = 4; + +/** + * Atomically attempts to reserve a spawn slot for `providerName`. + * + * If the current in-flight count is below `maxConcurrent`, increments the + * counter and returns true. Otherwise returns false WITHOUT incrementing — + * the caller is responsible for surfacing the saturation as a + * ProviderError(CONCURRENCY_LIMIT) for the fallback engine to consume. + * + * Atomicity: the check and the increment happen in a single synchronous + * block with no `await` in between. See the module-level invariant comment + * above for why this is sufficient. + * + * @param {string} providerName — provider key (e.g. 'anthropic') + * @param {number} [maxConcurrent=DEFAULT_MAX_CONCURRENT_SPAWNS] — limit from hints.maxConcurrent + * @returns {boolean} true if a slot was acquired, false if at limit + */ +export function tryAcquireSpawn(providerName, maxConcurrent = DEFAULT_MAX_CONCURRENT_SPAWNS) { + // Defensive: coerce undefined/null/non-integer to the default. validateProvider + // already enforces this at startup; this guards future plugin paths that + // bypass validation. + const limit = (typeof maxConcurrent === 'number' && Number.isInteger(maxConcurrent) && maxConcurrent >= 0) + ? maxConcurrent + : DEFAULT_MAX_CONCURRENT_SPAWNS; + + const current = _activeSpawns.get(providerName) ?? 0; + // Atomic check-then-increment (no `await` between read and write). + if (current >= limit) { + return false; + } + _activeSpawns.set(providerName, current + 1); + return true; +} + +/** + * Releases a spawn slot for `providerName`. Must be called exactly once per + * successful tryAcquireSpawn() call, regardless of whether the spawn succeeded + * or threw. The caller in server.mjs uses a try/finally pattern to guarantee + * the release fires on every exit path (success, error, abort, streaming end). + * + * Throws if the count would go negative — that indicates a bug (a release + * without a matching acquire, or a double-release). The throw is loud on + * purpose so the bug surfaces in tests rather than silently corrupting the + * counter for future requests. + * + * @param {string} providerName — provider key (e.g. 'anthropic') + * @throws {Error} if no slot is currently held for providerName + */ +export function releaseSpawn(providerName) { + const current = _activeSpawns.get(providerName) ?? 0; + if (current <= 0) { + throw new Error( + `releaseSpawn(${providerName}): counter would go negative — release without matching acquire (or double-release)`, + ); + } + const next = current - 1; + if (next === 0) { + _activeSpawns.delete(providerName); + } else { + _activeSpawns.set(providerName, next); + } +} + +/** + * Returns the current in-flight spawn count for `providerName`. Used by + * /health, diagnostics, and tests that need to assert peak concurrency. + * + * @param {string} providerName + * @returns {number} non-negative integer; 0 if no spawns in flight + */ +export function getActiveSpawnCount(providerName) { + return _activeSpawns.get(providerName) ?? 0; +} + +/** + * @internal — test seam: reset all in-flight spawn counters to zero. Used by + * test teardown to ensure a clean state across suites. Production code MUST + * NOT call this — it bypasses the acquire/release pairing invariant. + */ +export function __resetSpawnCounters() { + _activeSpawns.clear(); +} diff --git a/server.mjs b/server.mjs index 51bbf65..e55ef13 100644 --- a/server.mjs +++ b/server.mjs @@ -29,7 +29,16 @@ import { generateRequestId, SSE_DONE, } from './lib/ir/ir-to-openai.mjs'; -import { loadProviders, listAllProviderNames, getAliasMap, getModelCreated } from './lib/providers/index.mjs'; +import { + loadProviders, + listAllProviderNames, + getAliasMap, + getModelCreated, + tryAcquireSpawn, + releaseSpawn, + getActiveSpawnCount, + DEFAULT_MAX_CONCURRENT_SPAWNS, +} from './lib/providers/index.mjs'; import { ProviderError } from './lib/providers/base.mjs'; import { computeCacheKey, hasCacheControl, extractCacheControlMarkers } from './lib/cache/keys.mjs'; import { CacheStore } from './lib/cache/store.mjs'; @@ -503,43 +512,80 @@ async function handleChatCompletions(req, res) { // conditions item 1 requires no truncation). A non-enumerable __truncated marker // is set on the returned array so executeHopFn can evict the cache entry below. async function collectAllChunks() { + // D38 (issue #1): maxConcurrent runtime enforcement. + // Authority: ADR 0002 Amendment 6 + ADR 0004 Amendment 4. + // + // Try to acquire a spawn slot for this provider BEFORE invoking spawn(). + // If at limit, synthesise ProviderError(CONCURRENCY_LIMIT) which the + // fallback engine treats as a hard trigger — the chain advances to the + // next hop. validateProvider guarantees hints.maxConcurrent is present; + // DEFAULT_MAX_CONCURRENT_SPAWNS is a defense-in-depth fallback. + const maxConcurrent = hopProviderPlugin.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS; + if (!tryAcquireSpawn(hopProvider, maxConcurrent)) { + const concurrencyErr = new ProviderError( + `provider ${hopProvider} at maxConcurrent (${maxConcurrent}) — advancing to next hop`, + 'CONCURRENCY_LIMIT', + ); + concurrencyErr.providerName = hopProvider; + concurrencyErr.maxConcurrent = maxConcurrent; + // activeSpawns reflects the live counter at the rejection moment, + // queried directly. Since tryAcquireSpawn returned false the value + // equals maxConcurrent — read it explicitly for diagnostic clarity + // rather than echoing the limit (avoids future-reader confusion). + concurrencyErr.activeSpawns = getActiveSpawnCount(hopProvider); + throw concurrencyErr; + } + const chunks = []; + // try/finally: releaseSpawn MUST fire on every exit path — success + // (return at end), spawn throw (caught and re-thrown below), or the + // D16 truncation-salvage return. The finally is the only mechanism + // that guarantees release across all three. try { - for await (const irChunk of hopProviderPlugin.spawn(irReq, authContext)) { - // D16: check error chunks BEFORE pushing — preserves the invariant that - // chunks array contains only delta/stop chunks. Without this, the catch - // block's `chunks.length > 0` would mistake a single error chunk for - // "usable content streamed" (Case B) and synthesize a stop + return, - // sending an empty body to the client when the correct behavior is to - // re-throw and let the fallback engine advance the chain. - if (irChunk.type === 'error') { - throw new ProviderError( - irChunk.error ?? 'Provider emitted error chunk', - 'SPAWN_FAILED', - ); + try { + for await (const irChunk of hopProviderPlugin.spawn(irReq, authContext)) { + // D16: check error chunks BEFORE pushing — preserves the invariant that + // chunks array contains only delta/stop chunks. Without this, the catch + // block's `chunks.length > 0` would mistake a single error chunk for + // "usable content streamed" (Case B) and synthesize a stop + return, + // sending an empty body to the client when the correct behavior is to + // re-throw and let the fallback engine advance the chain. + if (irChunk.type === 'error') { + throw new ProviderError( + irChunk.error ?? 'Provider emitted error chunk', + 'SPAWN_FAILED', + ); + } + chunks.push(irChunk); + if (irChunk.type === 'stop') break; } - chunks.push(irChunk); - if (irChunk.type === 'stop') break; - } - } catch (spawnErr) { - if (spawnErr instanceof ProviderError && spawnErr.code === 'SPAWN_FAILED' && chunks.length > 0) { - // Case B (ADR 0004 Amendment 1): provider emitted usable chunks then exited - // non-zero. Synthesize a truncated stop and surface the partial response. - chunks.push({ type: 'stop', finish_reason: 'length' }); - logEvent('warn', 'spawn_failed_after_usable_chunks', { - chunks_count: chunks.length - 1, // exclude the synthesized stop - provider: hopProvider, - model: hopModel, - }); - // Mark as truncated so the caller can evict this entry from cache. - Object.defineProperty(chunks, '__truncated', { value: true, enumerable: false }); - return chunks; + } catch (spawnErr) { + if (spawnErr instanceof ProviderError && spawnErr.code === 'SPAWN_FAILED' && chunks.length > 0) { + // Case B (ADR 0004 Amendment 1): provider emitted usable chunks then exited + // non-zero. Synthesize a truncated stop and surface the partial response. + chunks.push({ type: 'stop', finish_reason: 'length' }); + logEvent('warn', 'spawn_failed_after_usable_chunks', { + chunks_count: chunks.length - 1, // exclude the synthesized stop + provider: hopProvider, + model: hopModel, + }); + // Mark as truncated so the caller can evict this entry from cache. + Object.defineProperty(chunks, '__truncated', { value: true, enumerable: false }); + return chunks; + } + // Case A (SPAWN_FAILED with no chunks) or any other error: re-throw. + // Fallback engine fires hard trigger and advances chain as before. + throw spawnErr; } - // Case A (SPAWN_FAILED with no chunks) or any other error: re-throw. - // Fallback engine fires hard trigger and advances chain as before. - throw spawnErr; + return chunks; + } finally { + // D38: spawn lifecycle ended (drain completed, stop chunk received, + // SPAWN_FAILED salvage returned, or any unexpected throw). Release + // the slot so the next caller can acquire it. Single-threaded JS + // guarantees no other caller has incremented this provider's count + // between our tryAcquireSpawn() above and this releaseSpawn(). + releaseSpawn(hopProvider); } - return chunks; } // D23: cacheable opt-out check (ADR 0002 Amendment 3 + ADR 0005 Amendment 3). @@ -626,7 +672,24 @@ async function handleChatCompletions(req, res) { // // On success: write chunks to res AND cache so subsequent identical requests // hit the burst-replay path. + // D38 (issue #1): pre-acquire a concurrency slot for the streaming path so + // saturation here behaves identically to saturation on the buffered path. + // If acquire fails, fall through (skip this branch) — the buffered path + // below also gates via tryAcquireSpawn and will surface a chain-exhausted + // error through executeWithFallback (correct behaviour: a single-hop chain + // at maxConcurrent has no other hop to advance to). + // + // Acquired here, released in the `finally` below. The gate intentionally + // lives BEFORE the streaming-branch entry check so the buffered fallthrough + // can re-attempt acquire from a clean slate. + let streamingAcquired = false; if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop) { + const candidatePlugin = loadedProviders.get(chain[0].provider); + const candidateMax = candidatePlugin?.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS; + streamingAcquired = candidatePlugin ? tryAcquireSpawn(chain[0].provider, candidateMax) : false; + } + + if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop && streamingAcquired) { const streamProvider = chain[0].provider; const streamModel = chain[0].model; const streamCacheKey = computeCacheKey(streamProvider, streamModel, ir); @@ -634,6 +697,8 @@ async function handleChatCompletions(req, res) { if (!streamPlugin) { // Provider disappeared between chain build and here (edge case). + // Release the slot we acquired above so the counter stays balanced. + releaseSpawn(streamProvider); return sendError(res, 503, `Provider ${streamProvider} is not enabled`, 'no_enabled_provider', olpErrorHeaders({ startMs, model: ir.model })); } @@ -783,6 +848,13 @@ async function handleChatCompletions(req, res) { res.end(); } } + } finally { + // D38 (issue #1): streaming spawn lifecycle ended — drain completed, + // stop chunk seen, generator exhausted without stop, or any catch + // path returned via res.end(). Release the slot acquired before + // entering the streaming branch. The finally fires on every JS exit + // path including the `return;` statements inside the try/catch body. + releaseSpawn(streamProvider); } return; } diff --git a/test-features.mjs b/test-features.mjs index b1d2fa5..b26da5e 100644 --- a/test-features.mjs +++ b/test-features.mjs @@ -28,7 +28,20 @@ import { SSE_DONE, } from './lib/ir/ir-to-openai.mjs'; import { validateProvider, ProviderError, withTimeout } from './lib/providers/base.mjs'; -import { loadProviders, getProviderForModel, getProviderByName, listAllProviderNames, getAliasMap, getModelCreated, REGISTRY_BOOTSTRAP_CREATED } from './lib/providers/index.mjs'; +import { + loadProviders, + getProviderForModel, + getProviderByName, + listAllProviderNames, + getAliasMap, + getModelCreated, + REGISTRY_BOOTSTRAP_CREATED, + tryAcquireSpawn, + releaseSpawn, + getActiveSpawnCount, + __resetSpawnCounters, + DEFAULT_MAX_CONCURRENT_SPAWNS, +} from './lib/providers/index.mjs'; import anthropic, { irToAnthropic, anthropicChunkToIR, @@ -8347,3 +8360,481 @@ describe('D33 round-5 cold-audit cleanup', () => { }); }); + +// ── Suite 18: D38 — maxConcurrent runtime enforcement (issue #1) ────────── +// +// Authority: +// - ADR 0002 Amendment 6 (D38): maxConcurrent runtime enforcement landed. +// - ADR 0004 Amendment 4 (D38): CONCURRENCY_LIMIT added to hard-trigger taxonomy. +// - GitHub issue #1 (maxConcurrent: declarative-only at v0.1). +// +// Tests: +// 18a: CONCURRENCY_LIMIT is in PROVIDER_ERROR_CODES. +// 18b: CONCURRENCY_LIMIT is a hard trigger (evaluateHardTriggers true). +// 18c: classifyTrigger returns 'hard' for CONCURRENCY_LIMIT. +// 18d: tryAcquireSpawn / releaseSpawn / getActiveSpawnCount unit behaviour. +// 18e: tryAcquireSpawn returns false at the limit. +// 18f: releaseSpawn throws when called without a matching acquire. +// 18g: DEFAULT_MAX_CONCURRENT_SPAWNS applies when maxConcurrent omitted. +// 18h: __resetSpawnCounters clears all in-flight counts. +// 18i: HTTP — 5 concurrent buffered requests against a maxConcurrent:2 mock → +// peak in-flight == 2; requests 3-5 hit chain-exhausted error. +// 18j: HTTP — counter releases after request completes (single-hop, sequential). +// 18k: Fallback — saturated primary (maxConcurrent:1) → secondary serves the 2nd. +// 18l: HTTP streaming — counter releases at end of stream, not at start. + +describe('D38 — maxConcurrent runtime enforcement (Suite 18)', () => { + + // ── 18a: PROVIDER_ERROR_CODES contains CONCURRENCY_LIMIT ────────────────── + + it('18a: CONCURRENCY_LIMIT is in PROVIDER_ERROR_CODES', () => { + assert.ok( + PROVIDER_ERROR_CODES.includes('CONCURRENCY_LIMIT'), + 'CONCURRENCY_LIMIT must be in PROVIDER_ERROR_CODES per ADR 0004 Amendment 4', + ); + }); + + // ── 18b: CONCURRENCY_LIMIT triggers hard fallback ───────────────────────── + + it('18b: evaluateHardTriggers(ProviderError CONCURRENCY_LIMIT) → true', () => { + const err = new ProviderError('provider stub at maxConcurrent (2)', 'CONCURRENCY_LIMIT'); + assert.equal( + evaluateHardTriggers(err), true, + 'CONCURRENCY_LIMIT must be classified as a hard trigger', + ); + }); + + // ── 18c: AUTH_MISSING regression guard ─────────────────────────────────── + // CONCURRENCY_LIMIT hard-trigger classification is covered by 18b; + // 18c guards that D38 did NOT accidentally flip AUTH_MISSING (a soft signal + // for v0.1 — see ADR 0004 Amendment 2) to hard alongside CONCURRENCY_LIMIT. + + it('18c: evaluateHardTriggers still false for AUTH_MISSING (regression guard)', () => { + // Regression guard: D38 didn't accidentally flip AUTH_MISSING to a hard trigger. + const err = new ProviderError('auth missing', 'AUTH_MISSING'); + assert.equal(evaluateHardTriggers(err), false); + }); + + // ── 18d: semaphore unit tests ───────────────────────────────────────────── + + describe('18d — tryAcquireSpawn / releaseSpawn / getActiveSpawnCount unit', () => { + + // Use a unique provider name to avoid colliding with other tests' state. + // __resetSpawnCounters() runs after each unit test to keep them isolated. + + it('18d.1: tryAcquireSpawn increments count from 0 → 1', () => { + __resetSpawnCounters(); + assert.equal(getActiveSpawnCount('unit-d1'), 0); + const ok = tryAcquireSpawn('unit-d1', 4); + assert.equal(ok, true, 'first acquire should succeed'); + assert.equal(getActiveSpawnCount('unit-d1'), 1); + __resetSpawnCounters(); + }); + + it('18d.2: releaseSpawn decrements count back to 0', () => { + __resetSpawnCounters(); + tryAcquireSpawn('unit-d2', 4); + assert.equal(getActiveSpawnCount('unit-d2'), 1); + releaseSpawn('unit-d2'); + assert.equal(getActiveSpawnCount('unit-d2'), 0); + __resetSpawnCounters(); + }); + + it('18d.3: each provider has independent counters', () => { + __resetSpawnCounters(); + tryAcquireSpawn('unit-d3-a', 4); + tryAcquireSpawn('unit-d3-a', 4); + tryAcquireSpawn('unit-d3-b', 4); + assert.equal(getActiveSpawnCount('unit-d3-a'), 2); + assert.equal(getActiveSpawnCount('unit-d3-b'), 1); + __resetSpawnCounters(); + }); + + it('18d.4: counter is removed from map when it reaches 0', () => { + __resetSpawnCounters(); + tryAcquireSpawn('unit-d4', 4); + releaseSpawn('unit-d4'); + // After release-to-zero, getActiveSpawnCount still returns 0 (Map.delete'd). + assert.equal(getActiveSpawnCount('unit-d4'), 0); + __resetSpawnCounters(); + }); + }); + + // ── 18e: tryAcquireSpawn returns false at the limit ─────────────────────── + + it('18e: tryAcquireSpawn returns false when count == maxConcurrent', () => { + __resetSpawnCounters(); + assert.equal(tryAcquireSpawn('unit-e', 2), true); + assert.equal(tryAcquireSpawn('unit-e', 2), true); + // At the limit now — third acquire must fail without incrementing. + assert.equal(tryAcquireSpawn('unit-e', 2), false, + 'acquire must fail when current count equals limit'); + // Counter must not have incremented on the failed acquire. + assert.equal(getActiveSpawnCount('unit-e'), 2, + 'failed acquire must not increment the counter'); + __resetSpawnCounters(); + }); + + // ── 18f: releaseSpawn without acquire throws ────────────────────────────── + + it('18f: releaseSpawn without matching acquire throws', () => { + __resetSpawnCounters(); + assert.throws( + () => releaseSpawn('unit-f-never-acquired'), + /releaseSpawn.*counter would go negative/, + 'release without acquire must throw to surface the bug loudly', + ); + __resetSpawnCounters(); + }); + + // ── 18g: DEFAULT_MAX_CONCURRENT_SPAWNS fallback ─────────────────────────── + + it('18g: DEFAULT_MAX_CONCURRENT_SPAWNS applies when maxConcurrent argument omitted', () => { + __resetSpawnCounters(); + assert.ok(Number.isInteger(DEFAULT_MAX_CONCURRENT_SPAWNS) && DEFAULT_MAX_CONCURRENT_SPAWNS > 0, + 'DEFAULT_MAX_CONCURRENT_SPAWNS must be a positive integer'); + // Acquire up to the default limit without arg — all should succeed. + for (let i = 0; i < DEFAULT_MAX_CONCURRENT_SPAWNS; i++) { + assert.equal(tryAcquireSpawn('unit-g'), true, `acquire #${i + 1} should succeed`); + } + // Next acquire (still no arg) should fail at the default boundary. + assert.equal(tryAcquireSpawn('unit-g'), false, + `acquire #${DEFAULT_MAX_CONCURRENT_SPAWNS + 1} should fail at default limit`); + __resetSpawnCounters(); + }); + + it('18g.2: non-integer maxConcurrent coerces to DEFAULT_MAX_CONCURRENT_SPAWNS', () => { + __resetSpawnCounters(); + // Defensive coercion: a plugin path that bypasses validateProvider could + // produce undefined/null/NaN. The gate must not blow up; it must use the default. + assert.equal(tryAcquireSpawn('unit-g2', undefined), true); + assert.equal(tryAcquireSpawn('unit-g2', null), true); + assert.equal(tryAcquireSpawn('unit-g2', NaN), true); + assert.equal(tryAcquireSpawn('unit-g2', 'four'), true); + // After 4 acquires (matching default), 5th should fail. + assert.equal(getActiveSpawnCount('unit-g2'), 4); + assert.equal(tryAcquireSpawn('unit-g2', undefined), false); + __resetSpawnCounters(); + }); + + // ── 18h: __resetSpawnCounters clears state ──────────────────────────────── + + it('18h: __resetSpawnCounters clears all in-flight counts', () => { + tryAcquireSpawn('unit-h-1', 4); + tryAcquireSpawn('unit-h-2', 4); + assert.equal(getActiveSpawnCount('unit-h-1'), 1); + assert.equal(getActiveSpawnCount('unit-h-2'), 1); + __resetSpawnCounters(); + assert.equal(getActiveSpawnCount('unit-h-1'), 0); + assert.equal(getActiveSpawnCount('unit-h-2'), 0); + }); + + // ──────────────────────────────────────────────────────────────────────────── + // HTTP integration tests + // ──────────────────────────────────────────────────────────────────────────── + // + // Strategy: install a controllable mock spawn on the anthropic plugin that + // exposes a Promise-based "release gate" — the mock yields no data until + // the test explicitly resolves the gate. This lets us hold multiple spawns + // in-flight at the same time deterministically (no timing races). + // + // For the maxConcurrent:N tests we override anthropic.hints.maxConcurrent + // for the duration of the test, then restore. ProviderError synthesised by + // the gate carries CONCURRENCY_LIMIT and the fallback engine treats it as + // a hard trigger. With a single-hop chain the chain exhausts and the + // client receives a 502/non-200. + + describe('18 HTTP integration', () => { + let server18; + let port18; + let savedToken18; + let savedMaxConcurrent; + + before(async () => { + savedToken18 = process.env.CLAUDE_CODE_OAUTH_TOKEN; + process.env.CLAUDE_CODE_OAUTH_TOKEN = 'fake-token-suite-18'; + __setProvidersEnabled({ anthropic: true }); + __resetSpawnCounters(); + + const { createOlpServer: s18, __clearCache: cc18 } = await import('./server.mjs'); + cc18(); + server18 = s18(); + await new Promise((resolve, reject) => { + server18.listen(0, '127.0.0.1', resolve); + server18.once('error', reject); + }); + port18 = server18.address().port; + }); + + after(async () => { + __resetProvidersEnabled(); + __resetSpawnImpl(); + __resetSpawnCounters(); + if (savedToken18 !== undefined) { + process.env.CLAUDE_CODE_OAUTH_TOKEN = savedToken18; + } else { + delete process.env.CLAUDE_CODE_OAUTH_TOKEN; + } + if (!server18) return; + return new Promise(r => server18.close(r)); + }); + + /** + * Makes a controllable mock spawn that completes only when the given + * gate Promise resolves. Records peak in-flight via `peakRef`. + */ + function makeGatedMockSpawn(gatePromise, content, peakRef, activeRef) { + return function gatedSpawn(_bin, _args, _opts) { + const proc = new EventEmitter(); + proc.stdout = new EventEmitter(); + proc.stderr = new EventEmitter(); + proc.stdin = { + write: () => {}, + end: async () => { + // Track this spawn as in-flight from the moment stdin closes. + activeRef.count++; + if (activeRef.count > peakRef.peak) peakRef.peak = activeRef.count; + // Wait for the gate to resolve before emitting any data. + await gatePromise; + // Decrement on exit. + setImmediate(() => { + proc.stdout.emit('data', Buffer.from(content)); + proc.stdout.emit('end'); + proc.stderr.emit('end'); + proc.emit('close', 0, null); + activeRef.count--; + }); + }, + }; + proc.killed = false; + proc.kill = () => {}; + return proc; + }; + } + + // ── 18i: 5 concurrent requests, maxConcurrent:2 ──────────────────────── + + it('18i: 5 concurrent buffered requests with maxConcurrent:2 → peak in-flight == 2', async () => { + // Override anthropic's maxConcurrent for this test. + savedMaxConcurrent = anthropic.hints.maxConcurrent; + anthropic.hints.maxConcurrent = 2; + + const { __clearCache: cc18i } = await import('./server.mjs'); + cc18i(); + + // Gate that we'll resolve to let mock spawns complete. + let releaseGate; + const gatePromise = new Promise(resolve => { releaseGate = resolve; }); + const peakRef = { peak: 0 }; + const activeRef = { count: 0 }; + __setSpawnImpl(makeGatedMockSpawn(gatePromise, 'concurrency-test-response', peakRef, activeRef)); + + try { + // Fire 5 concurrent buffered requests with DIFFERENT message content + // so cache lookups don't collapse them. + const reqs = []; + for (let i = 0; i < 5; i++) { + reqs.push(fetch({ + port: port18, + method: 'POST', + path: '/v1/chat/completions', + body: { + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: `concurrency-test-msg-${i}-${Date.now()}` }], + }, + })); + } + + // Give the server a moment to receive all requests and attempt acquires + // (the requests that succeed acquire will wait on gatePromise; the + // ones that fail acquire will return immediately with an error). + // The wait must be long enough for all 5 to either enter the spawn or + // be rejected by the gate, but short enough to keep the test fast. + await new Promise(r => setTimeout(r, 100)); + + // Peak in-flight count must equal exactly 2 (== maxConcurrent). + assert.equal(peakRef.peak, 2, + `peak in-flight must be exactly 2 (== maxConcurrent), observed ${peakRef.peak}`); + // Also assert via the exported getActiveSpawnCount helper (covers a + // different path — the module-level _activeSpawns map vs the mock-side + // counter). Both must agree. + assert.equal(getActiveSpawnCount('anthropic'), 2, + 'module-level active-spawn counter must report 2 in-flight'); + + // Release the gate so the 2 in-flight spawns can complete. + releaseGate(); + + // Wait for all 5 requests to settle. + const results = await Promise.all(reqs); + + // 2 successes (requests that acquired), 3 errors (CONCURRENCY_LIMIT). + const successes = results.filter(r => r.status === 200); + const failures = results.filter(r => r.status !== 200); + assert.equal(successes.length, 2, + `expected exactly 2 successful responses, got ${successes.length} (statuses: ${results.map(r => r.status).join(', ')})`); + assert.equal(failures.length, 3, + `expected exactly 3 failed responses (chain-exhausted on CONCURRENCY_LIMIT), got ${failures.length}`); + } finally { + // Restore original maxConcurrent. + anthropic.hints.maxConcurrent = savedMaxConcurrent; + __resetSpawnImpl(); + __resetSpawnCounters(); + } + }); + + // ── 18j: counter releases after request completes ───────────────────── + + it('18j: counter releases to 0 after buffered request completes', async () => { + savedMaxConcurrent = anthropic.hints.maxConcurrent; + anthropic.hints.maxConcurrent = 2; + const { __clearCache: cc18j } = await import('./server.mjs'); + cc18j(); + + // Plain mock that completes immediately. + __setSpawnImpl(makeMockSpawn(['release-test-response'])); + + try { + // Fire request 1 and wait for completion. + const r1 = await fetch({ + port: port18, + method: 'POST', + path: '/v1/chat/completions', + body: { + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: `release-test-msg-1-${Date.now()}` }], + }, + }); + assert.equal(r1.status, 200); + // Counter must be back to 0 after the request completes. + assert.equal(getActiveSpawnCount('anthropic'), 0, + 'counter must release to 0 after request completes'); + + // Fire request 2 (different content for cache miss) — must also succeed, + // proving the slot is reusable. + const r2 = await fetch({ + port: port18, + method: 'POST', + path: '/v1/chat/completions', + body: { + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: `release-test-msg-2-${Date.now()}` }], + }, + }); + assert.equal(r2.status, 200); + assert.equal(getActiveSpawnCount('anthropic'), 0, + 'counter must release after second request too'); + } finally { + anthropic.hints.maxConcurrent = savedMaxConcurrent; + __resetSpawnImpl(); + __resetSpawnCounters(); + } + }); + + // ── 18l: streaming counter release ──────────────────────────────────── + + it('18l: streaming request — counter releases at end of stream', async () => { + savedMaxConcurrent = anthropic.hints.maxConcurrent; + anthropic.hints.maxConcurrent = 2; + const { __clearCache: cc18l } = await import('./server.mjs'); + cc18l(); + + __setSpawnImpl(makeMockSpawn(['stream-release-content'])); + + try { + // Make a streaming request and consume the full stream. + const port = port18; + const body = await new Promise((resolve, reject) => { + const req = httpRequest({ + hostname: '127.0.0.1', + port, + method: 'POST', + path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + }, res => { + let data = ''; + res.on('data', c => { data += c; }); + res.on('end', () => resolve(data)); + res.on('error', reject); + }); + req.on('error', reject); + req.write(JSON.stringify({ + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: `stream-release-${Date.now()}` }], + stream: true, + })); + req.end(); + }); + assert.ok(body.includes('data:'), 'streaming body must contain SSE data lines'); + + // After the stream completes (res.on('end') resolved), the counter must + // be back to 0. Tests the streaming-branch finally{} fired. + assert.equal(getActiveSpawnCount('anthropic'), 0, + 'counter must release after streaming response completes'); + } finally { + anthropic.hints.maxConcurrent = savedMaxConcurrent; + __resetSpawnImpl(); + __resetSpawnCounters(); + } + }); + }); + + // ── 18k: fallback test — saturated primary advances to secondary ────────── + // + // Uses a 2-hop chain executed via executeWithFallback (unit-level, not HTTP). + // Primary has maxConcurrent:1 and is already saturated by a synthetic + // pre-acquire; second request must advance to the fallback hop. + + describe('18k — fallback advances on saturation', () => { + after(() => __resetSpawnCounters()); + + it('18k: saturated primary (maxConcurrent:1) → fallback hop serves the request', async () => { + // Pre-saturate the 'anthropic' counter (1/1) so the next acquire fails. + __resetSpawnCounters(); + tryAcquireSpawn('anthropic', 1); + assert.equal(getActiveSpawnCount('anthropic'), 1); + + // Build a 2-hop chain: anthropic (saturated) → openai (will serve). + // executeHopFn for this test calls tryAcquireSpawn itself, mirroring + // the production gate semantics. On failure: throws ProviderError + // CONCURRENCY_LIMIT; fallback engine advances to next hop. + const chain = [ + { provider: 'anthropic', model: 'claude-sonnet-4-6' }, + { provider: 'openai', model: 'gpt-5.5' }, + ]; + const ir = makeIR({ model: 'claude-sonnet-4-6', stream: false }); + + async function testHopFn(hopProvider, hopModel, _ir) { + const max = hopProvider === 'anthropic' ? 1 : 4; + if (!tryAcquireSpawn(hopProvider, max)) { + const err = new ProviderError( + `provider ${hopProvider} at maxConcurrent (${max})`, + 'CONCURRENCY_LIMIT', + ); + err.providerName = hopProvider; + err.maxConcurrent = max; + err.activeSpawns = max; + throw err; + } + try { + // Yield a single stop chunk so the hop "succeeds". + return [{ type: 'delta', content: `served-by-${hopProvider}` }, { type: 'stop', finish_reason: 'stop' }]; + } finally { + releaseSpawn(hopProvider); + } + } + + const result = await executeWithFallback(chain, ir, testHopFn, { logEvent: () => {} }); + assert.ok(result.chunks !== null, 'fallback should have produced chunks'); + assert.equal(result.fallbackHops, 1, 'must have advanced to second hop'); + assert.equal(result.providerUsed, 'openai', + 'secondary (openai) must serve when primary (anthropic) is saturated'); + const content = result.chunks.filter(c => c.type === 'delta').map(c => c.content).join(''); + assert.ok(content.includes('served-by-openai'), 'content must come from fallback hop'); + + // Cleanup: release the synthetic pre-saturation slot. + releaseSpawn('anthropic'); + assert.equal(getActiveSpawnCount('anthropic'), 0); + }); + }); + +});