feat(resilience): add hierarchical concurrency admission

tests/e2e/resilience-plan-alignment.spec.ts

@@ -7,6 +7,7 @@ const resilienceSettings = {
     requestsPerMinute: 100,
     minTimeBetweenRequestsMs: 200,
     concurrentRequests: 10,
+    globalConcurrentRequests: 0,
     maxWaitMs: 120000,
   },
   connectionCooldown: {

tests/unit/accountSemaphore.test.ts

@@ -3,6 +3,7 @@ import { afterEach, describe, it } from "node:test";
 
 import {
   acquire,
+  acquireMany,
   buildAccountSemaphoreKey,
   getStats,
   markBlocked,
@@ -14,6 +15,108 @@ afterEach(() => {
   resetAll();
 });
 
+describe("accountSemaphore acquireMany", () => {
+  it("atomically acquires every enabled gate and releases them once", async () => {
+    const release = await acquireMany([
+      { key: "global", maxConcurrency: 2 },
+      { key: "provider:codex", maxConcurrency: 1 },
+      { key: "account:codex:one", maxConcurrency: 1 },
+      { key: "disabled", maxConcurrency: 0 },
+    ]);
+
+    assert.deepEqual(getStats(), {
+      global: { running: 1, queued: 0, maxConcurrency: 2, blockedUntil: null },
+      "provider:codex": { running: 1, queued: 0, maxConcurrency: 1, blockedUntil: null },
+      "account:codex:one": { running: 1, queued: 0, maxConcurrency: 1, blockedUntil: null },
+    });
+
+    release();
+    release();
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    assert.deepEqual(getStats(), {});
+  });
+
+  it("queues one atomic request without partially reserving free gates", async () => {
+    const releaseProvider = await acquire("provider:codex", { maxConcurrency: 1 });
+    const waiting = acquireMany(
+      [
+        { key: "global", maxConcurrency: 1 },
+        { key: "provider:codex", maxConcurrency: 1 },
+        { key: "account:codex:two", maxConcurrency: 1 },
+      ],
+      { timeoutMs: 200 }
+    );
+    await new Promise((resolve) => setTimeout(resolve, 10));
+
+    assert.equal(getStats().global?.running ?? 0, 0);
+    assert.equal(getStats()["account:codex:two"]?.running ?? 0, 0);
+    assert.equal(getStats()["provider:codex"]?.queued, 1);
+
+    releaseProvider();
+    const release = await waiting;
+    assert.equal(getStats().global?.running, 1);
+    assert.equal(getStats()["provider:codex"]?.running, 1);
+    assert.equal(getStats()["account:codex:two"]?.running, 1);
+    release();
+  });
+
+  it("removes an atomic waiter from every gate on abort", async () => {
+    const releaseGlobal = await acquire("global", { maxConcurrency: 1 });
+    const controller = new AbortController();
+    const waiting = acquireMany(
+      [
+        { key: "global", maxConcurrency: 1 },
+        { key: "provider:codex", maxConcurrency: 1 },
+      ],
+      { signal: controller.signal, timeoutMs: 200 }
+    );
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    controller.abort();
+    await assert.rejects(waiting, { name: "AbortError" });
+    assert.equal(getStats().global?.queued, 0);
+    assert.equal(getStats()["provider:codex"]?.queued ?? 0, 0);
+    releaseGlobal();
+  });
+
+  it("times out an atomic waiter without leaking reservations", async () => {
+    const releaseGlobal = await acquire("global", { maxConcurrency: 1 });
+    await assert.rejects(
+      acquireMany(
+        [
+          { key: "global", maxConcurrency: 1 },
+          { key: "provider:codex", maxConcurrency: 1 },
+        ],
+        { timeoutMs: 10 }
+      ),
+      (error: Error & { code?: string }) => error.code === "SEMAPHORE_TIMEOUT"
+    );
+    assert.equal(getStats().global?.queued, 0);
+    assert.equal(getStats()["provider:codex"]?.running ?? 0, 0);
+    releaseGlobal();
+  });
+
+  it("rejects an atomic waiter when any required gate queue is full", async () => {
+    const releaseGlobal = await acquire("global", { maxConcurrency: 1 });
+    const queued = acquire("global", { maxConcurrency: 1, maxQueueSize: 1, timeoutMs: 200 });
+    await new Promise((resolve) => setTimeout(resolve, 10));
+
+    await assert.rejects(
+      acquireMany(
+        [
+          { key: "global", maxConcurrency: 1 },
+          { key: "provider:codex", maxConcurrency: 1 },
+        ],
+        { maxQueueSize: 1, timeoutMs: 200 }
+      ),
+      (error: Error & { code?: string }) => error.code === "SEMAPHORE_QUEUE_FULL"
+    );
+    assert.equal(getStats()["provider:codex"]?.queued ?? 0, 0);
+
+    releaseGlobal();
+    (await queued)();
+  });
+});
+
 describe("accountSemaphore", async () => {
   it("queues requests beyond the account cap and drains on release", async () => {
     const key = buildAccountSemaphoreKey({

tests/unit/chatcore-hierarchical-admission.test.ts

@@ -0,0 +1,37 @@
+import assert from "node:assert/strict";
+import { readFileSync } from "node:fs";
+import { test } from "node:test";
+
+const source = readFileSync(
+  new URL("../../open-sse/handlers/chatCore.ts", import.meta.url),
+  "utf8"
+);
+
+test("chatCore acquires cumulative gates immediately before withRateLimit", () => {
+  const acquire = source.indexOf("await acquireConcurrencyGates(");
+  const rateLimit = source.indexOf("await withRateLimit(", acquire);
+  assert.ok(acquire >= 0, "hierarchical admission must be present");
+  assert.ok(rateLimit > acquire, "hierarchical admission must precede withRateLimit");
+
+  const admission = source.slice(acquire, rateLimit);
+  assert.match(admission, /key: "global"/);
+  assert.match(admission, /key: `provider:\$\{canonicalProviderKey\}`/);
+  assert.match(admission, /key: accountSemaphoreKey/);
+  assert.match(admission, /globalConcurrentRequests/);
+  assert.match(admission, /providerConcurrency/);
+  assert.match(admission, /maxWaitMs/);
+  assert.match(admission, /maxQueueDepth/);
+});
+
+test("each rotated account attempt acquires and releases a fresh composite slot", () => {
+  const attemptLoop = source.indexOf(
+    "while (attempts < maxAttempts || antigravityByopRotationPending)"
+  );
+  const acquire = source.indexOf("await acquireConcurrencyGates(", attemptLoop);
+  const finallyRelease = source.indexOf("releaseAccountSemaphore();", acquire);
+  const retryContinue = source.indexOf("continue;", acquire);
+
+  assert.ok(attemptLoop >= 0 && acquire > attemptLoop);
+  assert.ok(finallyRelease > acquire, "each attempt must release the composite slot");
+  assert.ok(retryContinue > acquire, "rotation remains inside the per-attempt acquisition loop");
+});

tests/unit/resilience-settings-provider-quota-overrides.test.ts

@@ -19,13 +19,26 @@ function cloneDefaults(): ResilienceSettings {
 test("updateResilienceSchema accepts providerQuotaOverrides entries", () => {
   const parsed = updateResilienceSchema.safeParse({
     providerQuotaOverrides: {
-      minimax: { rpm: 30, concurrency: 4 },
+      minimax: { rpm: 30, concurrency: 4, providerConcurrency: 8 },
       nvidia: { rpm: 60 },
     },
   });
   assert.equal(parsed.success, true, "valid override map should parse");
 });
 
+test("provider concurrency accepts zero as disabled and survives normalization", () => {
+  const resolved = resolveResilienceSettings({
+    resilienceSettings: {
+      providerQuotaOverrides: {
+        codex: { providerConcurrency: 3 },
+        claude: { providerConcurrency: 0 },
+      },
+    },
+  });
+  assert.equal(resolved.providerQuotaOverrides.codex.providerConcurrency, 3);
+  assert.equal(resolved.providerQuotaOverrides.claude.providerConcurrency, 0);
+});
+
 test("updateResilienceSchema allows a body containing only providerQuotaOverrides", () => {
   // The superRefine requires at least one field; a lone override map is a
   // legitimate update and must not trigger "Must provide resilience settings".