ocp/lib/tui/semaphore.mjs

// 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).

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; }

  // 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() {
    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) => { this._waiters.push(resolve); });
  }

  // Release a slot. If a waiter is queued, hand the slot directly to it (inflight stays
  // constant across the handoff); otherwise decrement.
  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--;
    }
  }

  // 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();
    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).
export function buildTuiHealthBlock({ enabled, entrypointMode, maxConcurrent }, tuiStats, semaphore) {
  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,
  };
}