// TUI-path concurrency limiter (audit finding C-4). // // WHY THIS EXISTS, SEPARATE FROM server.mjs's MAX_CONCURRENT: // The global MAX_CONCURRENT gate lives in spawnClaudeProcess() (the -p / stream-json // path). callClaudeTui() NEVER calls spawnClaudeProcess — it calls runTuiTurn(), which // boots a full interactive `claude` inside a fresh tmux session. So nothing bounded the // TUI path: N concurrent TUI requests spawned N simultaneous cold-boot tmux+claude // processes. On a small host (a Pi 4 serving a family) a burst of ~5 is an OOM risk, and // it also multiplies subscription rate-limit pressure. This is an INDEPENDENT limiter for // the TUI path that mirrors MAX_CONCURRENT's intent without coupling to it (the two pools // are different shapes: a stream-json spawn is cheap and fast; a TUI turn is a heavy // cold-boot + up to 120s wallclock). // // QUEUE vs REJECT: we QUEUE (await a slot), mirroring the spirit of MAX_CONCURRENT's // intent not to drop requests, rather than rejecting immediately. To avoid unbounded // memory growth from a runaway client, the wait queue itself is bounded by maxQueue // (default: a generous multiple of the concurrency limit). When the queue is full, run() // rejects with a tui_queue_full error (the caller surfaces it as a 503) — a deterministic // backpressure signal rather than silent OOM. // // 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 } = {}) { this.limit = Math.max(1, parseInt(limit, 10) || 1); // Default queue cap: 32× the limit. Large enough that real family-burst traffic never // hits it, small enough that a pathological flood can't grow the queue without bound. this.maxQueue = Number.isFinite(maxQueue) ? maxQueue : this.limit * 32; this._inflight = 0; this._waiters = []; // FIFO queue of resolve callbacks waiting for a slot } 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. // `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(); } if (this._waiters.length >= this.maxQueue) { return Promise.reject(new Error( `tui_queue_full: TUI concurrency limit (${this.limit}) reached and wait queue ` + `(${this.maxQueue}) is full`)); } 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. 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() { 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. // `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 { this.release(); } } } // ── TUI drift observability (audit C-5) — pure helpers, importable for testing ── // Record an observed cc_entrypoint into the (mutable) tuiStats counter. Sets lastEntrypoint // unconditionally and increments entrypointMismatches when the spawn was supposed to be // subscription-pool ("cli") but the transcript reported something else (a silent drift to // the metered Agent SDK pool — the audit's top risk after the 6/15 billing flip). // Returns true iff this observation was a mismatch (so the caller can also emit a log). export function recordTuiEntrypoint(tuiStats, observed, expectedMode = "cli") { tuiStats.lastEntrypoint = observed ?? null; const mismatch = expectedMode === "cli" && observed !== "cli"; if (mismatch) tuiStats.entrypointMismatches++; return mismatch; } // Build the additive /health `tui` block (ADR 0007 PR-B amendment). Pure: given the // config + live counters, returns the exact object embedded in /health. New fields only — // behaviour-preserving for existing /health consumers (grandfathered B.2 under ADR 0006). // // `pool` (optional, warm pane pool — lib/tui/pool.mjs): a TuiPanePool, or null/undefined // when the pool is off (the default). Reported as `pool: null` when off so the block's // shape stays stable, and as the pool's stats (size / warm / hits / misses / …) when on — // the operator's window onto both the hit rate and the standing idle-process cost. export function buildTuiHealthBlock({ enabled, entrypointMode, maxConcurrent }, tuiStats, semaphore, pool = null) { return { enabled, entrypointMode, // cli | auto | off lastEntrypoint: tuiStats.lastEntrypoint, // last observed cc_entrypoint, or null entrypointMismatches: tuiStats.entrypointMismatches, inflight: semaphore.inflight, // current concurrent TUI turns queued: semaphore.queued, // turns waiting for a slot maxConcurrent, pool: pool ? pool.stats() : null, // warm pane pool, or null when disabled }; }