From 26b09af5a801f730da0149447303cd589b6b9b23 Mon Sep 17 00:00:00 2001 From: dtzp555 Date: Tue, 7 Jul 2026 22:41:12 +1000 Subject: [PATCH] fix: honor runtime-lowered concurrency limit + cancel queued waiters on disconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes three findings from an independent concurrency audit of the -p/stream-json wait-queue (lib/tui/semaphore.mjs, reused by server.mjs as `claudeSemaphore`) and its acquireClaudeSlot()/callClaudeTui() callers in server.mjs: F1 (MEDIUM) — release() handed a freed slot straight to the next queued waiter without re-checking `this.limit`, so a PATCH /settings maxConcurrent decrease was silently ignored until every already-inflight task happened to finish on its own. release() now only re-grants when post-decrement inflight is still under the current limit, and a new setLimit() wakes queued waiters immediately when the limit is raised instead of only on the next incidental release(). F2 (MEDIUM) — a request queued behind the concurrency limit had no link to its HTTP connection, so a client that disconnected while still queued would still get a claude process spawned for it once a slot freed — burning subscription quota for a dead socket. acquire() now accepts an optional AbortSignal; server.mjs derives one from the client's res "close" event (closeSignalFor) and passes it into claudeSemaphore.acquire() / tuiSemaphore.acquire() while queued. On abort the waiter is spliced out of the queue (not just flagged), so `queued` accounting stays exact; the same "close" signal is wired into acquireClaudeSlot() (-p path, non-streaming + streaming + singleflight-wrapped) and callClaudeTui() (TUI path). If the response is already destroyed by the time we try to queue, we reject immediately without ever entering the queue. F8 (cosmetic) — acquireClaudeSlot() set `stats.queued = claudeSemaphore.queued + 1` BEFORE calling acquire(), over-reporting /health's queued count by 1 whenever a slot was granted immediately (the common, non-queued case). acquire() already updates its internal queue synchronously before returning a Promise, so reading claudeSemaphore.queued right AFTER calling it (instead of guessing "+1" before) is exact. No /health field was added, removed, or renamed. ALIGNMENT.md: this PR touches request-handler code (callClaude, callClaudeStreaming, callClaudeTui, acquireClaudeSlot) but is local concurrency-control/queue-accounting infrastructure with no cli.js wire analogue — it does not add, rename, or change any endpoint, header, request field, or response field, and does not touch the /v1/messages forwarding path or the OAuth bearer machinery (the two Class A surfaces this repo governs). The /health response shape is unchanged (same field set, same nesting; only the *value* of the pre-existing `stats.queued` field is corrected). Per CLAUDE.md hard-requirement #1, a cli.js citation is therefore declared ABSENT: there is no corresponding cli.js operation to cite because this is not a cli.js-mirror (Class A) change and not a Class B endpoint-contract change either. Tests: added 6 unit tests to test-features.mjs against the shared TuiSemaphore (lowering the limit mid-load does not over-admit; raising the limit wakes queued waiters up to the new headroom, FIFO; a queued waiter cancelled via AbortSignal is spliced out and never later acquires; an already-aborted signal never touches the queue; cancelling one of several queued waiters preserves FIFO for the rest). 238 pre-existing tests remain green; suite is now 244/244. Verification: `node --check server.mjs && node --check lib/tui/semaphore.mjs && npm test` — 244 passed, 0 failed. Co-Authored-By: Claude Fable 5 --- lib/tui/semaphore.mjs | 72 +++++++++++++--- server.mjs | 178 ++++++++++++++++++++++++++++++---------- test-features.mjs | 186 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 382 insertions(+), 54 deletions(-) diff --git a/lib/tui/semaphore.mjs b/lib/tui/semaphore.mjs index 56ddab7..c84b3cc 100644 --- a/lib/tui/semaphore.mjs +++ b/lib/tui/semaphore.mjs @@ -20,6 +20,15 @@ // // Pure + importable so test-features.mjs can assert the bound directly (no server boot). +// Thrown by acquire() when the caller-supplied AbortSignal fires before a slot was granted +// (audit finding F2 — a client that disconnects while queued must never receive a slot; the +// queue entry is spliced out, not just flagged, so `queued` accounting stays exact). Distinct +// `name` lets callers (server.mjs acquireClaudeSlot) tell "client went away" apart from +// "queue is full" without string-matching the message. +export class SemaphoreAbortError extends Error { + constructor(message) { super(message); this.name = "SemaphoreAbortError"; } +} + export class TuiSemaphore { // limit: max concurrent slots. maxQueue: max waiters before run() rejects with backpressure. constructor(limit, { maxQueue } = {}) { @@ -34,9 +43,30 @@ export class TuiSemaphore { get inflight() { return this._inflight; } get queued() { return this._waiters.length; } + // Runtime-adjust the concurrency limit (audit finding F1 — a PATCH /settings maxConcurrent + // change must actually take effect, not just be ignored until every currently-inflight task + // happens to finish). Lowering the limit is handled lazily by release() (see below) — it + // simply stops re-granting until inflight drains under the new, lower limit. Raising the + // limit has immediate headroom, so we wake as many queued waiters as now fit. + setLimit(limit) { + this.limit = Math.max(1, parseInt(limit, 10) || 1); + while (this._inflight < this.limit && this._waiters.length > 0) { + const next = this._waiters.shift(); + this._inflight++; + next(); + } + } + // Acquire a slot. Resolves once a slot is free (immediately if under the limit, otherwise // when an in-flight task releases). Rejects synchronously-ish if the wait queue is full. - acquire() { + // `signal` (optional AbortSignal, F2) lets the caller cancel a QUEUED wait — e.g. wired to + // a client's socket "close" event so a request that disconnects before a slot is granted + // is removed from the queue instead of eventually being handed a slot for a dead socket. + // If `signal` is already aborted, reject immediately without ever touching the queue. + acquire(signal) { + if (signal?.aborted) { + return Promise.reject(new SemaphoreAbortError("acquire aborted before requesting a slot")); + } if (this._inflight < this.limit) { this._inflight++; return Promise.resolve(); @@ -46,24 +76,44 @@ export class TuiSemaphore { `tui_queue_full: TUI concurrency limit (${this.limit}) reached and wait queue ` + `(${this.maxQueue}) is full`)); } - return new Promise((resolve) => { this._waiters.push(resolve); }); + return new Promise((resolve, reject) => { + let waiter; // the FIFO entry — captured so onAbort can find + splice exactly this one + const onAbort = () => { + const idx = this._waiters.indexOf(waiter); + if (idx === -1) return; // already granted a slot (shifted out by release()/setLimit) — too late to cancel + this._waiters.splice(idx, 1); // remove, not just flag — keeps `queued` accounting exact + reject(new SemaphoreAbortError("acquire aborted while queued")); + }; + waiter = () => { + signal?.removeEventListener("abort", onAbort); + resolve(); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + this._waiters.push(waiter); + }); } - // Release a slot. If a waiter is queued, hand the slot directly to it (inflight stays - // constant across the handoff); otherwise decrement. + // Release a slot. Always frees the caller's own slot first, then re-grants it to the next + // waiter ONLY if the (post-decrement) inflight count is still under the current limit (F1 + // fix). This is what makes a runtime-lowered limit actually bite: if the limit was lowered + // while over-subscribed, releases stop re-granting and inflight drains toward the new limit + // instead of a freed slot being handed straight back out at the old, higher occupancy. release() { - const next = this._waiters.shift(); - if (next) { - next(); // the woken waiter already "owns" the slot — inflight unchanged - } else if (this._inflight > 0) { - this._inflight--; + if (this._inflight > 0) this._inflight--; + if (this._inflight < this.limit) { + const next = this._waiters.shift(); + if (next) { + this._inflight++; + next(); + } } } // Run fn() under one slot. Releases in a finally so a throw (PR-A's honesty gates, // wallclock truncation, paste-not-landed, tmux spawn failure) NEVER leaks a slot. - async run(fn) { - await this.acquire(); + // `signal` (optional, F2) is forwarded to acquire() so a queued run() can be cancelled. + async run(fn, signal) { + await this.acquire(signal); try { return await fn(); } finally { diff --git a/server.mjs b/server.mjs index c8c9875..42317c9 100644 --- a/server.mjs +++ b/server.mjs @@ -43,7 +43,7 @@ import { DEFAULT_PORT } from "./lib/constants.mjs"; import { isLoopbackBind } from "./lib/net.mjs"; import { runTuiTurn, reapStaleTuiSessions, resolveTuiHome } from "./lib/tui/session.mjs"; import { detectTuiUpstreamError } from "./lib/tui/transcript.mjs"; -import { TuiSemaphore, recordTuiEntrypoint, buildTuiHealthBlock } from "./lib/tui/semaphore.mjs"; +import { TuiSemaphore, SemaphoreAbortError, recordTuiEntrypoint, buildTuiHealthBlock } from "./lib/tui/semaphore.mjs"; import { createSerialMutex, createTtlCache, isTokenExpiring, orderLabelsLastGoodFirst } from "./lib/spawn-auth.mjs"; const __dirname = dirname(fileURLToPath(import.meta.url)); @@ -507,16 +507,61 @@ class ConcurrencyOverflowError extends Error { constructor(message) { super(message); this.name = "ConcurrencyOverflowError"; this.httpStatus = 429; this.retryAfter = CLAUDE_QUEUE_RETRY_AFTER; } } +// Tagged error for audit finding F2: the client disconnected while queued (or was already gone +// before we even tried to queue it). Distinct from ConcurrencyOverflowError so callers never send +// a response on this path — there is no socket left to write to. +class RequestDisconnectedError extends Error { + constructor(message) { super(message); this.name = "RequestDisconnectedError"; } +} + +// Build an AbortSignal that fires when `res` (an http.ServerResponse) closes — i.e. the client +// disconnected. Used to cancel a QUEUED concurrency-slot wait (F2) so a client that gives up +// before a slot is granted is spliced out of the wait queue instead of eventually spawning a +// claude process for a dead socket. If `res` has already closed by the time we get here (its +// underlying stream already torn down), the signal is returned pre-aborted so acquire() rejects +// immediately without ever touching the queue — the "close already fired before we attach" case. +// `detach()` MUST be called once the wait settles (granted or rejected) to avoid a listener leak. +function closeSignalFor(res) { + const controller = new AbortController(); + if (!res || typeof res.on !== "function") return { signal: controller.signal, detach() {} }; + if (res.destroyed) { + controller.abort(); + return { signal: controller.signal, detach() {} }; + } + const onClose = () => controller.abort(); + res.on("close", onClose); + return { signal: controller.signal, detach() { res.removeListener("close", onClose); } }; +} + // Acquire a -p concurrency slot, queuing if all are busy (up to CLAUDE_MAX_QUEUE). Resolves to a // release() fn that MUST be called exactly once on every exit path (wired into ctx.cleanup()). -// Rejects with ConcurrencyOverflowError when the wait-queue is full. Increments stats.queued while -// waiting (decremented on acquire) and stats.queueRejections on overflow. -async function acquireClaudeSlot() { - stats.queued = claudeSemaphore.queued + 1; // reflect this waiter before we (maybe) block +// Rejects with ConcurrencyOverflowError when the wait-queue is full, or with +// RequestDisconnectedError when `res` closes before a slot is granted (F2) — the caller must not +// spawn claude in that case. `res` is optional (back-compat for any caller without a live response +// object); omitting it just means a queued wait can't be cancelled early. +// +// F8 fix: stats.queued is set from claudeSemaphore.queued AFTER calling acquire() (not before) — +// acquire() synchronously updates _inflight/_waiters before its Promise ever resolves, so reading +// .queued right after the call already reflects reality. The old code set `queued + 1` BEFORE +// calling acquire() to account for "this waiter", which over-reported by 1 whenever the slot was +// granted immediately (the common case, not a queue at all). +async function acquireClaudeSlot(res) { + const { signal, detach } = closeSignalFor(res); + const slot = claudeSemaphore.acquire(signal); + stats.queued = claudeSemaphore.queued; // accurate: acquire() already updated the queue synchronously try { - await claudeSemaphore.acquire(); + await slot; } catch (e) { + detach(); stats.queued = claudeSemaphore.queued; + if (e instanceof SemaphoreAbortError) { + // Client-driven cancellation, not backpressure — do NOT count it as a queueRejection or + // log it as concurrency_queue_full (that log/counter means "the queue itself is full"). + logEvent("info", "concurrency_wait_cancelled", { + reason: "client_disconnected", inflight: claudeSemaphore.inflight, queued: claudeSemaphore.queued, + }); + throw new RequestDisconnectedError("client disconnected while waiting for a concurrency slot"); + } stats.queueRejections++; logEvent("warn", "concurrency_queue_full", { limit: claudeSemaphore.limit, maxQueue: claudeSemaphore.maxQueue, @@ -526,6 +571,7 @@ async function acquireClaudeSlot() { `backpressure: concurrency limit (${claudeSemaphore.limit}) reached and wait queue ` + `(${claudeSemaphore.maxQueue}) is full — retry shortly`); } + detach(); stats.queued = claudeSemaphore.queued; let released = false; return function releaseClaudeSlot() { @@ -1135,14 +1181,29 @@ function spawnClaudeProcess(model, messages, conversationId, keyName, releaseSlo // We accumulate full text across all content_block_delta events plus the // assistant-aggregate fallback, then resolve with the assembled string. // Reference: OLP ADR 0009 Amendment 1 + commit 97e7d16. -async function callClaude(model, messages, conversationId, keyName) { +// `res` (optional, F2) is the client's http.ServerResponse — passed through so a queued wait +// can be cancelled the moment the client disconnects, instead of spawning claude for a dead +// socket once a slot finally frees up. +async function callClaude(model, messages, conversationId, keyName, res) { // FIX ⑥: acquire a concurrency slot first (queues up to CLAUDE_MAX_QUEUE; rejects with a - // ConcurrencyOverflowError → 429 when the queue is full). The release fn is passed into the - // spawn so the idempotent cleanup() frees it on every exit path. If the spawn itself throws - // synchronously (before cleanup is wired), release here so the slot never leaks. - const releaseSlot = await acquireClaudeSlot(); - // F3: resolve the per-spawn HOME/token decision (may serialize on the real-HOME fallback mutex). - const spawnDecision = await resolveSpawnDecision(); + // ConcurrencyOverflowError → 429 when the queue is full, or a RequestDisconnectedError (F2) + // if the client goes away first). The release fn is passed into the spawn so the idempotent + // cleanup() frees it on every exit path. If the spawn itself throws synchronously (before + // cleanup is wired), release here so the slot never leaks. + // F2×F3 composition: the slot acquire comes FIRST and is the cancellable step — a client + // that disconnects while queued rejects here, BEFORE resolveSpawnDecision() runs, so a + // cancelled request can never acquire (or briefly hold) the real-HOME fallback mutex. + const releaseSlot = await acquireClaudeSlot(res); + // F3: resolve the per-spawn HOME/token decision (may serialize on the real-HOME fallback + // mutex). If it throws, release the just-acquired slot before propagating — cleanup() is + // not wired yet at this point. + let spawnDecision; + try { + spawnDecision = await resolveSpawnDecision(); + } catch (err) { + releaseSlot(); + throw err; + } return new Promise((resolve, reject) => { let ctx; try { @@ -1221,28 +1282,44 @@ async function callClaude(model, messages, conversationId, keyName) { // flag that could perturb cc_entrypoint classification. // Authority: claude CLI v2.1.158 interactive mode (cc_entrypoint=cli). // SECURITY: A-path single-user ONLY — home is NOT isolation (see ADR 0007). -function callClaudeTui(model, messages, _conversationId, _keyName) { +// `res` (optional, F2) is the client's http.ServerResponse — see closeSignalFor. +async function callClaudeTui(model, messages, _conversationId, _keyName, res) { const cliModel = MODEL_MAP[model] || model; const prompt = messagesToPrompt(messages); // includes system as [System] inline recordModelRequest(cliModel, prompt.length); - // C-4: gate the heavy interactive boot behind the TUI semaphore. run() acquires a slot - // (queuing if all are busy, up to maxQueue), then releases in a finally so any throw from - // runTuiTurn (tmux spawn failure, paste-not-landed) OR from the honesty gates below - // (truncation / error banner) can NEVER leak a slot. tuiSemaphore.inflight feeds /health. - return tuiSemaphore.run(() => runTuiTurn({ - prompt, - model: cliModel, - claudeBin: CLAUDE, - home: TUI_HOME, - realHome: process.env.HOME, - cwd: TUI_CWD, - port: PORT, // F7 fix: port-scopes the tmux session name so a sibling OCP instance on a - // different port never collides with this instance's reap/kill-server logic. - wallclockMs: TUI_WALLCLOCK_MS, - entrypointMode: TUI_ENTRYPOINT, - }).then(({ text, entrypoint, truncated }) => { + // C-4: gate the heavy interactive boot behind the TUI semaphore (queuing if all slots are + // busy, up to maxQueue). F2: `signal` (tied to `res` "close") cancels a QUEUED wait the + // instant the client disconnects, so a dead socket never triggers a cold-boot tmux+claude + // spawn; detach() drops the "close" listener as soon as the wait settles rather than + // holding it for the whole (up to 120s) turn. + const { signal, detach } = closeSignalFor(res); + try { + await tuiSemaphore.acquire(signal); + } catch (err) { + detach(); + throw err instanceof SemaphoreAbortError + ? new RequestDisconnectedError("client disconnected while waiting for a TUI concurrency slot") + : err; + } + detach(); + // release() runs in a finally so any throw from runTuiTurn (tmux spawn failure, + // paste-not-landed) OR from the honesty gates below (truncation / error banner) can NEVER + // leak a slot. tuiSemaphore.inflight feeds /health. + try { + const { text, entrypoint, truncated } = await runTuiTurn({ + prompt, + model: cliModel, + claudeBin: CLAUDE, + home: TUI_HOME, + realHome: process.env.HOME, + cwd: TUI_CWD, + port: PORT, // F7 fix: port-scopes the tmux session name so a sibling OCP instance on a + // different port never collides with this instance's reap/kill-server logic. + wallclockMs: TUI_WALLCLOCK_MS, + entrypointMode: TUI_ENTRYPOINT, + }); // ── Honesty gates (issue #133) ─ run BEFORE recordModelSuccess / cache write-back. - // A throw here propagates to the .catch below (recordModelError + reject), so the + // A throw here propagates to the catch below (recordModelError + reject), so the // result never reaches the downstream setCachedResponse / singleflight / SUCCESS path. // C-2: the wall-clock cap hit with partial text and NO terminal marker — the turn @@ -1277,10 +1354,12 @@ function callClaudeTui(model, messages, _conversationId, _keyName) { logEvent("warn", "tui_entrypoint_mismatch", { expected: "cli", got: entrypoint, model: cliModel }); } return text; - }).catch((err) => { + } catch (err) { recordModelError(cliModel, false); throw err; - })); + } finally { + tuiSemaphore.release(); + } } // ── SSE heartbeat (opt-in idle watchdog) ──────────────────────────────── @@ -1326,18 +1405,30 @@ async function callClaudeStreaming(model, messages, conversationId, res, authInf // FIX ⑥: acquire a concurrency slot first (queues up to CLAUDE_MAX_QUEUE). On overflow, surface // HTTP 429 + Retry-After (NOT 500). Release is wired into cleanup() for every exit path; if the // spawn throws synchronously before cleanup is wired, release here. + // F2: pass `res` so a queued wait is cancelled the instant this client disconnects — the client + // is already gone in that case, so there is no response to send back. let releaseSlot; try { - releaseSlot = await acquireClaudeSlot(); + releaseSlot = await acquireClaudeSlot(res); } catch (err) { + if (err instanceof RequestDisconnectedError) return; // client gone — nothing to write to if (err instanceof ConcurrencyOverflowError) { return jsonResponse(res, 429, { error: { message: sanitizeError(err.message), type: "rate_limit_error" } }, { "Retry-After": String(err.retryAfter) }); } return jsonResponse(res, 500, { error: { message: sanitizeError(err.message), type: "proxy_error" } }); } - // F3: resolve the per-spawn HOME/token decision (may serialize on the real-HOME fallback mutex). - const spawnDecision = await resolveSpawnDecision(); + // F3: resolve the per-spawn HOME/token decision (may serialize on the real-HOME fallback + // mutex). F2×F3 composition: this runs strictly AFTER the (cancellable) slot acquire, so a + // request cancelled while queued never touches the fallback mutex. If it throws, release + // the just-acquired slot before responding — cleanup() is not wired yet at this point. + let spawnDecision; + try { + spawnDecision = await resolveSpawnDecision(); + } catch (err) { + releaseSlot(); + return jsonResponse(res, 500, { error: { message: sanitizeError(err.message), type: "proxy_error" } }); + } let ctx; try { ctx = spawnClaudeProcess(model, messages, conversationId, authInfo.keyName, releaseSlot, spawnDecision); @@ -2009,9 +2100,12 @@ function applySettingUpdate(key, value) { switch (key) { case "timeout": TIMEOUT = value; break; - // FIX ⑥: keep the -p wait-queue semaphore's limit in sync with the runtime MAX_CONCURRENT - // so a /settings change to maxConcurrent actually changes how many claude procs run at once. - case "maxConcurrent": MAX_CONCURRENT = value; claudeSemaphore.limit = Math.max(1, value); break; + // FIX ⑥ + F1: keep the -p wait-queue semaphore's limit in sync with the runtime MAX_CONCURRENT + // so a /settings change to maxConcurrent actually changes how many claude procs run at once — + // in BOTH directions. setLimit() (not a bare `.limit =` assignment) is required: lowering + // needs release() to stop over-granting until inflight drains under the new cap, and raising + // needs queued waiters woken immediately to use the new headroom. See lib/tui/semaphore.mjs. + case "maxConcurrent": MAX_CONCURRENT = value; claudeSemaphore.setLimit(value); break; case "sessionTTL": SESSION_TTL = value; break; case "maxPromptChars": MAX_PROMPT_CHARS = value; break; case "cacheTTL": CACHE_TTL = value; break; @@ -2168,7 +2262,7 @@ async function handleChatCompletions(req, res) { const t0TuiStream = Date.now(); const promptCharsTuiStream = messages.reduce((a, m) => a + contentToText(m.content).length, 0); try { - const content = await callClaudeTui(model, messages, conversationId, req._authKeyName); + const content = await callClaudeTui(model, messages, conversationId, req._authKeyName, res); if (CACHE_TTL > 0 && req._cacheHash) { try { setCachedResponse(req._cacheHash, model, content); } catch (e) { logEvent("error", "cache_write_failed", { error: e.message }); } } @@ -2205,7 +2299,7 @@ async function handleChatCompletions(req, res) { // will re-read the freshly-populated cache entry here rather than spawning. const recheck = getCachedResponse(req._cacheHash, CACHE_TTL); if (recheck) return recheck.response; - const c = await upstreamCall(model, messages, conversationId, req._authKeyName); + const c = await upstreamCall(model, messages, conversationId, req._authKeyName, res); try { setCachedResponse(req._cacheHash, model, c); } catch (e) { logEvent("error", "cache_write_failed", { error: e.message }); } return c; }); @@ -2226,7 +2320,7 @@ async function handleChatCompletions(req, res) { // Fallback: cache disabled (CACHE_TTL=0) or no _cacheHash — original path untouched. try { - const content = await upstreamCall(model, messages, conversationId, req._authKeyName); + const content = await upstreamCall(model, messages, conversationId, req._authKeyName, res); const id = `chatcmpl-${randomUUID()}`; completionResponse(res, id, model, content); try { recordUsage({ keyId: req._authKeyId, keyName: req._authKeyName, model, promptChars, responseChars: content.length, elapsedMs: Date.now() - t0Usage, success: true }); } catch (e) { logEvent("error", "usage_record_failed", { error: e.message }); } diff --git a/test-features.mjs b/test-features.mjs index 229cd66..d4555b5 100644 --- a/test-features.mjs +++ b/test-features.mjs @@ -2005,7 +2005,7 @@ test("resolveTuiHome: explicit OCP_TUI_HOME wins regardless of env token (back-c }); // ── TUI concurrency limiter + drift observability (PR-B: audit C-4 / C-5) ── -import { TuiSemaphore, recordTuiEntrypoint, buildTuiHealthBlock } from "./lib/tui/semaphore.mjs"; +import { TuiSemaphore, SemaphoreAbortError, recordTuiEntrypoint, buildTuiHealthBlock } from "./lib/tui/semaphore.mjs"; console.log("\nTUI concurrency limiter (C-4):"); @@ -2110,6 +2110,190 @@ await asyncTest("FIX ⑥: slot released on normal completion is immediately reus assert.equal(sem.inflight, 0); }); +// ── Audit F1 — runtime-lowered/raised limit must actually bite ────────────── +// server.mjs reuses this same TuiSemaphore as `claudeSemaphore`; a PATCH /settings +// maxConcurrent update now calls `claudeSemaphore.setLimit(value)` (see applySettingUpdate's +// "maxConcurrent" case). These tests pin the semaphore-level contract that fix depends on. +console.log("\nF1 — runtime concurrency-limit changes (setLimit / release honoring the current limit):"); + +await asyncTest("F1: lowering the limit mid-load — release() stops re-granting until inflight drains under the new limit", async () => { + const sem = new TuiSemaphore(3, { maxQueue: 16 }); + const g = [deferred(), deferred(), deferred()]; + const held = g.map((d) => sem.run(async () => { await d.p; })); + await new Promise((r) => setImmediate(r)); + assert.equal(sem.inflight, 3, "3 tasks hold the 3 slots"); + // A 4th arrives while at capacity — it queues. + const g4 = deferred(); + const queued4 = sem.run(async () => { await g4.p; }); + await new Promise((r) => setImmediate(r)); + assert.equal(sem.queued, 1, "4th request queued"); + + // Operator lowers maxConcurrent from 3 to 1 while all 3 original slots are still inflight + // (mirrors a PATCH /settings maxConcurrent=1 hitting server.mjs mid-burst). + sem.setLimit(1); + assert.equal(sem.limit, 1); + + // Releasing one of the 3 original holders must NOT hand the freed slot to the queued 4th + // request — before the F1 fix, release() handed slots off unconditionally, so inflight + // would have stayed pinned at the OLD higher occupancy forever. + g[0].resolve(); + await held[0]; + await new Promise((r) => setImmediate(r)); + assert.equal(sem.inflight, 2, "inflight drains toward the new limit, not re-granted"); + assert.equal(sem.queued, 1, "4th request is STILL queued — not over-admitted"); + + g[1].resolve(); + await held[1]; + await new Promise((r) => setImmediate(r)); + assert.equal(sem.inflight, 1, "inflight now exactly at the new limit (1)"); + assert.equal(sem.queued, 1, "still queued — inflight(1) is not < limit(1), so no grant yet"); + + // Releasing the LAST original holder finally drops inflight under the new limit — only + // now does the queued 4th request get granted. + g[2].resolve(); + await held[2]; + await new Promise((r) => setImmediate(r)); + assert.equal(sem.inflight, 1, "queued 4th request now holds the single slot"); + assert.equal(sem.queued, 0, "queue drained"); + g4.resolve(); + await queued4; + assert.equal(sem.inflight, 0); +}); + +await asyncTest("F1: raising the limit wakes queued waiters immediately, up to the new headroom", async () => { + const sem = new TuiSemaphore(1, { maxQueue: 16 }); + const g1 = deferred(); + const t1 = sem.run(async () => { await g1.p; }); // holds the only slot + await new Promise((r) => setImmediate(r)); + const started = []; + const g2 = deferred(), g3 = deferred(); + const t2 = sem.run(async () => { started.push(2); await g2.p; }); + const t3 = sem.run(async () => { started.push(3); await g3.p; }); + await new Promise((r) => setImmediate(r)); + assert.equal(sem.queued, 2, "both queue behind the single holder"); + assert.deepEqual(started, [], "neither queued task has started"); + + // Operator raises maxConcurrent from 1 to 3 (2 units of new headroom) — BOTH queued + // waiters must be woken immediately, without waiting for t1 to release. + sem.setLimit(3); + await new Promise((r) => setImmediate(r)); + assert.equal(sem.inflight, 3, "t1 + both newly-woken waiters now hold slots"); + assert.equal(sem.queued, 0, "queue drained by the limit raise"); + assert.deepEqual(started.sort(), [2, 3], "both queued tasks started without waiting for t1's release"); + + g1.resolve(); g2.resolve(); g3.resolve(); + await Promise.all([t1, t2, t3]); + assert.equal(sem.inflight, 0); +}); + +await asyncTest("F1: raising the limit wakes only as many waiters as the new headroom allows (FIFO)", async () => { + const sem = new TuiSemaphore(1, { maxQueue: 16 }); + const g1 = deferred(); + const t1 = sem.run(async () => { await g1.p; }); + await new Promise((r) => setImmediate(r)); + const started = []; + const g2 = deferred(), g3 = deferred(); + const t2 = sem.run(async () => { started.push(2); await g2.p; }); + const t3 = sem.run(async () => { started.push(3); await g3.p; }); + await new Promise((r) => setImmediate(r)); + assert.equal(sem.queued, 2); + + sem.setLimit(2); // only 1 unit of new headroom (1 -> 2) — exactly one queued waiter wakes + await new Promise((r) => setImmediate(r)); + assert.equal(sem.inflight, 2); + assert.equal(sem.queued, 1, "one waiter still queued — only one slot of headroom existed"); + assert.deepEqual(started, [2], "FIFO: the earlier-queued waiter (t2) wakes, not t3"); + + // Freeing t1's slot afterward still honors the (now current) limit of 2 via release()'s + // normal path — the still-queued t3 gets in once a slot actually frees. + g1.resolve(); + await t1; + await new Promise((r) => setImmediate(r)); + assert.deepEqual(started, [2, 3], "t3 granted once a slot frees, honoring the raised limit"); + assert.equal(sem.queued, 0); + + g2.resolve(); g3.resolve(); + await t2; await t3; + assert.equal(sem.inflight, 0); +}); + +// ── Audit F2 — queued waiters must be cancellable on client disconnect ────── +// server.mjs wires an AbortSignal derived from the client's res "close" event into +// claudeSemaphore.acquire()/tuiSemaphore.acquire() (see closeSignalFor + acquireClaudeSlot / +// callClaudeTui). These tests pin the semaphore-level cancellation contract that depends on. +console.log("\nF2 — queued-wait cancellation via AbortSignal (client disconnect while queued):"); + +await asyncTest("F2: aborting a QUEUED waiter rejects with SemaphoreAbortError and SPLICES it out (queued drops immediately, not just flagged)", async () => { + const sem = new TuiSemaphore(1, { maxQueue: 16 }); + const g1 = deferred(); + const t1 = sem.run(async () => { await g1.p; }); // holds the only slot + await new Promise((r) => setImmediate(r)); + const controller = new AbortController(); + const acquire2 = sem.acquire(controller.signal); // queues behind t1 + await new Promise((r) => setImmediate(r)); + assert.equal(sem.queued, 1, "second acquire queued"); + + controller.abort(); // simulates the client disconnecting while still queued + await assert.rejects(acquire2, SemaphoreAbortError, "cancelled waiter rejects with SemaphoreAbortError"); + assert.equal(sem.queued, 0, "cancelled waiter is REMOVED — queue length drops immediately"); + assert.equal(sem.inflight, 1, "t1's slot is untouched by the cancellation"); + + // Prove the cancelled waiter never later acquires a slot: free t1's slot and confirm + // nobody is waiting to receive it (the queue is genuinely empty, not just decremented). + g1.resolve(); + await t1; + assert.equal(sem.inflight, 0, "slot freed with nobody queued — the cancelled waiter never got it"); +}); + +await asyncTest("F2: an already-aborted signal rejects acquire() immediately, never touching the wait queue", async () => { + const sem = new TuiSemaphore(1, { maxQueue: 16 }); + const g1 = deferred(); + const t1 = sem.run(async () => { await g1.p; }); // holds the only slot + await new Promise((r) => setImmediate(r)); + + const controller = new AbortController(); + controller.abort(); // client already gone before this request ever tries to acquire + await assert.rejects(sem.acquire(controller.signal), SemaphoreAbortError); + assert.equal(sem.queued, 0, "never entered the wait queue at all"); + + g1.resolve(); await t1; +}); + +await asyncTest("F2: cancelling one queued waiter preserves FIFO order for the others", async () => { + const sem = new TuiSemaphore(1, { maxQueue: 16 }); + const g1 = deferred(); + const t1 = sem.run(async () => { await g1.p; }); + await new Promise((r) => setImmediate(r)); + + const started = []; + const cA = new AbortController(); + const cB = new AbortController(); + const accA = sem.acquire(cA.signal).then(() => started.push("A")); + const accB = sem.acquire(cB.signal).then(() => started.push("B")); + const g3 = deferred(); + const t3 = sem.run(async () => { started.push("C"); await g3.p; }); + await new Promise((r) => setImmediate(r)); + assert.equal(sem.queued, 3, "A, B, C all queued behind t1"); + + cB.abort(); // B (the middle waiter) disconnects + await assert.rejects(accB, SemaphoreAbortError); + assert.equal(sem.queued, 2, "B removed; A and C remain, in original relative order"); + + g1.resolve(); + await t1; + await new Promise((r) => setImmediate(r)); + assert.deepEqual(started, ["A"], "A (queued first, still present) is granted next — FIFO preserved after B's removal"); + assert.equal(sem.inflight, 1); + assert.equal(sem.queued, 1, "C still waiting"); + + sem.release(); // A was acquired directly (not via run()) — free its slot manually + await new Promise((r) => setImmediate(r)); + assert.deepEqual(started, ["A", "C"], "C granted next"); + g3.resolve(); + await t3; + assert.equal(sem.inflight, 0); +}); + console.log("\nTUI drift observability (C-5):"); test("recordTuiEntrypoint: observed 'cli' is NOT a mismatch and sets lastEntrypoint", () => {