feat(resilience): add hierarchical concurrency admission (#11493)

Cherry-pick of the 3 value commits from #11323 (by @RaviTharuma), dropping 17 already-merged rebase-baggage commits. Validated in a combined-batch worktree off release/v3.8.51 tip: - typecheck:core, file-size, changelog-integrity, complexity, cognitive-complexity gates — all OK - Focused tests, isolated re-run flake-free: accountSemaphore.test.ts 11/11, chatcore-hierarchical-admission.test.ts 2/2, resilience-settings-provider-quota-overrides.test.ts 12/12, i18n-vi-completeness.test.ts 5/5 - lint clean once isolated from the unrelated ESLint-10 suppressions regression carried by a separate PR in the batch Closes #7778. Full credit to @RaviTharuma for the design and implementation.
2026-08-25 16:42:16 +03:00 · 2026-08-25 07:19:09 -03:00
parent 23fa2709a9
commit a179ffed5b
17 changed files with 489 additions and 248 deletions
--- a/docs/changelog/fragments/7778.md
+++ b/docs/changelog/fragments/7778.md
@@ -0,0 +1 @@
+feat(resilience): atomically enforce cumulative global, provider, and account concurrency limits
--- a/docs/plans/7778-hierarchical-admission.md
+++ b/docs/plans/7778-hierarchical-admission.md
@@ -0,0 +1,23 @@
+# #7778 hierarchical admission cleanup plan
+
+1. Lock the existing single-key semaphore contract and the new atomic multi-key
+   contract with focused tests: no partial reservations, FIFO queueing, abort,
+   timeout, queue-full, idempotent release, stats, and cleanup.
+2. Generalize the existing account semaphore in place. Keep `acquire()` as a
+   compatibility wrapper around `acquireMany()`; do not add a second scheduler
+   or a dependency.
+3. Replace the account-only acquisition in `chatCore` with one cumulative
+   global/provider/account acquisition immediately before `withRateLimit`.
+   Reacquire the whole set whenever account rotation changes the connection,
+   and retain the release through streaming completion.
+4. Extend the existing resilience settings pipeline (types, defaults,
+   normalization, schema, API response, UI, and translations) with the global
+   and provider caps. Relabel the old Bottleneck concurrency control as
+   connection/quota-scope concurrency so its real scope is explicit.
+5. Run focused tests, lint, typecheck, static checks, and the full test suite;
+   document the behavioral change in the changelog.
+
+Behavior intentionally preserved: zero/null concurrency bypasses a gate,
+account-only callers keep using `acquire()`, blocked-account controls retain
+their key format and API, and provider rate-limit queue behavior remains
+unchanged.
--- a/open-sse/handlers/chatCore.ts
+++ b/open-sse/handlers/chatCore.ts
@@ -396,7 +396,7 @@ import {
 } from "../services/rateLimitManager.ts";
 import * as localLimiterErrors from "../services/rateLimitManager/errors.ts";
 import {
-  acquire as acquireAccountSemaphore,
+  acquireMany as acquireConcurrencyGates,
  markBlocked as markAccountSemaphoreBlocked,
 } from "../services/accountSemaphore.ts";
 import { lockModel, lockModelIfPerModelQuota } from "../services/accountFallback.ts";
@@ -447,6 +447,7 @@ import { extractFacts } from "@/lib/memory/extraction";
 import { handleToolCallExecution } from "@/lib/skills/interception";
 import { MEMORY_BUILTIN_TOOL_NAMES } from "@/lib/skills/memoryBuiltins";
 import { OMNIROUTE_RESPONSE_HEADERS } from "@/shared/constants/headers";
+import { resolveProviderId } from "@/shared/constants/providers";
 import { getClaudeCodeCompatibleRequestDefaults } from "@/lib/providers/requestDefaults";
 import {
  buildClaudeCodeCompatibleRequest,
@@ -525,6 +526,7 @@ export async function handleChatCore({
  managedLease = null,
 }) {
  let { provider, model, extendedContext } = modelInfo;
+  const resilienceSettings = resolveResilienceSettings(cachedSettings);
  if (!skipResourcePressureGuard) {
    try {
      const pressureGuard = checkResourcePressureGuard();
@@ -3051,6 +3053,10 @@ export async function handleChatCore({
              connectionId: attemptConnectionId,
              credentials: execCreds,
            });
+            const canonicalProviderKey = resolveProviderId(String(provider).trim().toLowerCase());
+            const providerConcurrency =
+              resilienceSettings.providerQuotaOverrides[canonicalProviderKey]
+                ?.providerConcurrency ?? 0;

            trace("pre_semaphore", {
              semaphoreKey: accountSemaphoreKey,
@@ -3061,13 +3067,27 @@ export async function handleChatCore({
                stage: "waiting_account_slot",
              });
            }
-            const releaseAccountSemaphore =
-              accountSemaphoreKey && accountSemaphoreMaxConcurrency != null
-                ? await acquireAccountSemaphore(accountSemaphoreKey, {
-                    maxConcurrency: accountSemaphoreMaxConcurrency,
-                    signal: streamController.signal,
-                  })
-                : () => {};
+            const releaseAccountSemaphore = await acquireConcurrencyGates(
+              [
+                {
+                  key: "global",
+                  maxConcurrency: resilienceSettings.requestQueue.globalConcurrentRequests,
+                },
+                {
+                  key: `provider:${canonicalProviderKey}`,
+                  maxConcurrency: providerConcurrency,
+                },
+                {
+                  key: accountSemaphoreKey || "",
+                  maxConcurrency: accountSemaphoreKey ? accountSemaphoreMaxConcurrency : null,
+                },
+              ],
+              {
+                timeoutMs: resilienceSettings.requestQueue.maxWaitMs,
+                maxQueueSize: resilienceSettings.requestQueue.maxQueueDepth,
+                signal: streamController.signal,
+              }
+            );
            trace("post_semaphore");
            updatePendingScope(pendingScope, {
              stage: "waiting_rate_limit",
@@ -3328,6 +3348,7 @@ export async function handleChatCore({
                      "ANTIGRAVITY_BYOP_ROTATION",
                      `BYOP 422 on connection ${String(byopFailedId).slice(0, 8)} → rotating to ${String(byopNextCreds.connectionId).slice(0, 8)}`
                    );
+                    releaseAccountSemaphore();
                    Object.assign(credentials, byopNextCreds);
                    antigravityByopRotationPending = true;
                    continue;
--- a/open-sse/services/accountSemaphore.ts
+++ b/open-sse/services/accountSemaphore.ts
@@ -1,20 +1,40 @@
 /**
- * Account Semaphore
+ * Hierarchical in-memory concurrency admission.
 *
- * In-memory provider/account concurrency limiter keyed by provider and account.
- * Requests beyond the configured concurrency cap wait in a FIFO queue until a slot opens,
- * the gate is unblocked, or the queue timeout expires.
+ * `acquire()` preserves the account-semaphore API. `acquireMany()` admits one
+ * request only when every applicable global/provider/account gate has room.
 */
-
 export interface AccountSemaphoreKeyParts {
  provider: string;
  accountKey: string;
 }

-interface QueuedAcquire {
+export interface AcquireAccountSemaphoreOptions {
+  maxConcurrency?: number | null;
+  timeoutMs?: number;
+  signal?: AbortSignal | null;
+  maxQueueSize?: number;
+}
+
+export interface SemaphoreRequirement {
+  key: string;
+  maxConcurrency?: number | null;
+}
+
+export type AcquireManyOptions = Omit<AcquireAccountSemaphoreOptions, "maxConcurrency">;
+
+interface AcquireRequest {
+  keys: string[];
  resolve: (release: () => void) => void;
  reject: (error: Error) => void;
-  timer: ReturnType<typeof setTimeout>;
+  timer: ReturnType<typeof setTimeout> | null;
+  signal: AbortSignal | null;
+  abortListener: (() => void) | null;
+  settled: boolean;
+}
+
+interface QueuedAcquire {
+  request: AcquireRequest;
 }

 interface AccountGate {
@@ -25,13 +45,6 @@ interface AccountGate {
  cleanupTimer: ReturnType<typeof setTimeout> | null;
 }

-export interface AcquireAccountSemaphoreOptions {
-  maxConcurrency?: number | null;
-  timeoutMs?: number;
-  signal?: AbortSignal | null;
-  maxQueueSize?: number;
-}
-
 export interface AccountSemaphoreStatsEntry {
  running: number;
  queued: number;
@@ -41,12 +54,9 @@ export interface AccountSemaphoreStatsEntry {

 const DEFAULT_TIMEOUT_MS = 30_000;
 const DEFAULT_MAX_QUEUE_SIZE = 20;
-
 const gates = new Map<string, AccountGate>();
+const queuedRequests = new Set<AcquireRequest>();

-/**
- * Build the canonical account semaphore key.
- */
 export function buildAccountSemaphoreKey({
  provider,
  accountKey,
@@ -55,25 +65,23 @@ export function buildAccountSemaphoreKey({
 }

 function isBypassed(maxConcurrency?: number | null): boolean {
-  return maxConcurrency == null || maxConcurrency <= 0;
+  return maxConcurrency == null || !Number.isFinite(maxConcurrency) || maxConcurrency <= 0;
 }

 function createNoopReleaseFn(): () => void {
  let released = false;
-
  return () => {
    if (released) return;
    released = true;
  };
 }

-function ensureGate(semaphoreKey: string, maxConcurrency: number): AccountGate {
-  const existing = gates.get(semaphoreKey);
+function ensureGate(key: string, maxConcurrency: number): AccountGate {
+  const existing = gates.get(key);
  if (existing) {
    existing.maxConcurrency = maxConcurrency;
    return existing;
  }
-
  const created: AccountGate = {
    running: 0,
    maxConcurrency,
@@ -81,7 +89,7 @@ function ensureGate(semaphoreKey: string, maxConcurrency: number): AccountGate {
    blockedUntil: null,
    cleanupTimer: null,
  };
-  gates.set(semaphoreKey, created);
+  gates.set(key, created);
  return created;
 }

@@ -100,91 +108,101 @@ function clearCleanupTimer(gate: AccountGate): void {
  gate.cleanupTimer = null;
 }

-function cleanupGateIfIdle(semaphoreKey: string): void {
-  const gate = gates.get(semaphoreKey);
-  if (!gate) return;
-  if (gate.running > 0 || gate.queue.length > 0 || isBlocked(gate)) return;
+function cleanupGateIfIdle(key: string): void {
+  const gate = gates.get(key);
+  if (!gate || gate.running > 0 || gate.queue.length > 0 || isBlocked(gate)) return;
  clearCleanupTimer(gate);
-  gates.delete(semaphoreKey);
+  gates.delete(key);
 }

-function scheduleCleanup(semaphoreKey: string): void {
-  const gate = gates.get(semaphoreKey);
+function scheduleCleanup(key: string): void {
+  const gate = gates.get(key);
  if (!gate) return;
  clearCleanupTimer(gate);
-
  gate.cleanupTimer = setTimeout(() => {
    gate.cleanupTimer = null;
-    cleanupGateIfIdle(semaphoreKey);
+    cleanupGateIfIdle(key);
  }, 0);
-
  gate.cleanupTimer.unref?.();
 }

-function drainQueue(semaphoreKey: string): void {
-  const gate = gates.get(semaphoreKey);
-  if (!gate) return;
-
-  while (gate.queue.length > 0 && gate.running < gate.maxConcurrency && !isBlocked(gate)) {
-    const next = gate.queue.shift();
-    if (!next) break;
-    clearTimeout(next.timer);
-    gate.running++;
-    next.resolve(createReleaseFn(semaphoreKey));
-  }
-
-  if (gate.running === 0 && gate.queue.length === 0) {
-    scheduleCleanup(semaphoreKey);
-  }
-}
-
-function createReleaseFn(semaphoreKey: string): () => void {
-  let released = false;
-
-  return () => {
-    if (released) return;
-    released = true;
-
-    const gate = gates.get(semaphoreKey);
-    if (!gate) return;
-    if (gate.running > 0) {
-      gate.running--;
-    }
-
-    if (gate.queue.length > 0) {
-      drainQueue(semaphoreKey);
-      return;
-    }
-
-    scheduleCleanup(semaphoreKey);
-  };
-}
-
-function createSemaphoreTimeoutError(
-  semaphoreKey: string,
-  timeoutMs: number
-): Error & { code: string } {
-  const error = new Error(`Semaphore timeout after ${timeoutMs}ms for ${semaphoreKey}`) as Error & {
-    code: string;
-  };
-  error.code = "SEMAPHORE_TIMEOUT";
+function makeAbortError(signal: AbortSignal): Error {
+  if (signal.reason instanceof Error) return signal.reason;
+  const error = new Error(
+    typeof signal.reason === "string" ? signal.reason : "The operation was aborted"
+  );
+  error.name = "AbortError";
  return error;
 }

-function makeAbortError(signal: AbortSignal): Error {
-  const reason = signal.reason;
-  if (reason instanceof Error) return reason;
-  const err = new Error(typeof reason === "string" ? reason : "The operation was aborted");
-  err.name = "AbortError";
-  return err;
+function createSemaphoreError(code: string, message: string): Error & { code: string } {
+  const error = new Error(message) as Error & { code: string };
+  error.code = code;
+  return error;
+}
+
+function removeRequest(request: AcquireRequest): void {
+  queuedRequests.delete(request);
+  if (request.timer) clearTimeout(request.timer);
+  if (request.abortListener && request.signal) {
+    request.signal.removeEventListener("abort", request.abortListener);
+  }
+  for (const key of request.keys) {
+    const gate = gates.get(key);
+    if (!gate) continue;
+    const index = gate.queue.findIndex((queued) => queued.request === request);
+    if (index >= 0) gate.queue.splice(index, 1);
+    if (gate.running === 0 && gate.queue.length === 0) scheduleCleanup(key);
+  }
+}
+
+function canAcquire(request: AcquireRequest): boolean {
+  return request.keys.every((key) => {
+    const gate = gates.get(key);
+    return (
+      gate != null &&
+      !isBlocked(gate) &&
+      gate.running < gate.maxConcurrency &&
+      gate.queue[0]?.request === request
+    );
+  });
+}
+
+function createCompositeReleaseFn(keys: string[]): () => void {
+  let released = false;
+  return () => {
+    if (released) return;
+    released = true;
+    for (const key of keys) {
+      const gate = gates.get(key);
+      if (gate && gate.running > 0) gate.running--;
+    }
+    drainQueues();
+    for (const key of keys) {
+      const gate = gates.get(key);
+      if (gate && gate.running === 0 && gate.queue.length === 0) scheduleCleanup(key);
+    }
+  };
+}
+
+function drainQueues(): void {
+  let progressed = true;
+  while (progressed) {
+    progressed = false;
+    for (const request of queuedRequests) {
+      if (request.settled || !canAcquire(request)) continue;
+      request.settled = true;
+      removeRequest(request);
+      for (const key of request.keys) gates.get(key)!.running++;
+      request.resolve(createCompositeReleaseFn(request.keys));
+      progressed = true;
+      break;
+    }
+  }
 }

-/**
- * Acquire a slot for a provider/model/account tuple.
- * Returns an idempotent release function that is safe to call in finally blocks.
- */
 export function acquire(
-  semaphoreKey: string,
+  key: string,
  {
    maxConcurrency = null,
    timeoutMs = DEFAULT_TIMEOUT_MS,
@@ -192,146 +210,118 @@ export function acquire(
    maxQueueSize = DEFAULT_MAX_QUEUE_SIZE,
  }: AcquireAccountSemaphoreOptions = {}
 ): Promise<() => void> {
-  if (isBypassed(maxConcurrency)) {
-    return Promise.resolve(createNoopReleaseFn());
+  return acquireMany([{ key, maxConcurrency }], { timeoutMs, signal, maxQueueSize });
+}
+
+/**
+ * Acquire all enabled requirements as one FIFO reservation.
+ *
+ * Waiting never increments any gate, preventing a saturated child gate from
+ * holding capacity in a parent gate.
+ */
+export function acquireMany(
+  requirements: SemaphoreRequirement[],
+  {
+    timeoutMs = DEFAULT_TIMEOUT_MS,
+    signal = null,
+    maxQueueSize = DEFAULT_MAX_QUEUE_SIZE,
+  }: AcquireManyOptions = {}
+): Promise<() => void> {
+  const enabled = new Map<string, number>();
+  for (const requirement of requirements) {
+    if (isBypassed(requirement.maxConcurrency)) continue;
+    const limit = Math.trunc(requirement.maxConcurrency as number);
+    enabled.set(requirement.key, Math.min(enabled.get(requirement.key) ?? limit, limit));
+  }
+  if (enabled.size === 0) return Promise.resolve(createNoopReleaseFn());
+  if (signal?.aborted) return Promise.reject(makeAbortError(signal));
+
+  const keys = [...enabled.keys()].sort();
+  for (const key of keys) {
+    const gate = ensureGate(key, enabled.get(key)!);
+    clearCleanupTimer(gate);
+    if (maxQueueSize > 0 && gate.queue.length >= maxQueueSize) {
+      return Promise.reject(
+        createSemaphoreError(
+          "SEMAPHORE_QUEUE_FULL",
+          `Semaphore queue full (${maxQueueSize}) for ${key}`
+        )
+      );
+    }
  }

-  if (signal?.aborted) {
-    return Promise.reject(makeAbortError(signal));
-  }
-
-  // isBypassed() above already excluded null/<=0 — ensureGate requires a plain
-  // number, but a boolean-returning helper isn't a type predicate TS can narrow on.
-  const gate = ensureGate(semaphoreKey, maxConcurrency as number);
-  clearCleanupTimer(gate);
-
-  if (gate.running < gate.maxConcurrency && !isBlocked(gate)) {
-    gate.running++;
-    return Promise.resolve(createReleaseFn(semaphoreKey));
-  }
-
-  if (gate.queue.length >= maxQueueSize) {
-    const err = new Error(`Semaphore queue full (${maxQueueSize}) for ${semaphoreKey}`) as Error & {
-      code: string;
-    };
-    err.code = "SEMAPHORE_QUEUE_FULL";
-    return Promise.reject(err);
+  if (
+    keys.every((key) => {
+      const gate = gates.get(key)!;
+      return gate.queue.length === 0 && gate.running < gate.maxConcurrency && !isBlocked(gate);
+    })
+  ) {
+    for (const key of keys) gates.get(key)!.running++;
+    return Promise.resolve(createCompositeReleaseFn(keys));
  }

  return new Promise((resolve, reject) => {
-    let abortListener: (() => void) | null = null;
-
-    const cleanup = () => {
-      if (abortListener && signal) {
-        signal.removeEventListener("abort", abortListener);
-      }
+    const request: AcquireRequest = {
+      keys,
+      resolve,
+      reject,
+      timer: null,
+      signal,
+      abortListener: null,
+      settled: false,
    };
-
-    const timer = setTimeout(() => {
-      cleanup();
-      const nextGate = gates.get(semaphoreKey);
-      if (!nextGate) {
-        reject(createSemaphoreTimeoutError(semaphoreKey, timeoutMs));
-        return;
-      }
-
-      const queueIndex = nextGate.queue.findIndex((item) => item.timer === timer);
-      if (queueIndex !== -1) {
-        nextGate.queue.splice(queueIndex, 1);
-      }
-
-      if (nextGate.running === 0 && nextGate.queue.length === 0) {
-        scheduleCleanup(semaphoreKey);
-      }
-
-      reject(createSemaphoreTimeoutError(semaphoreKey, timeoutMs));
+    request.timer = setTimeout(() => {
+      if (request.settled) return;
+      request.settled = true;
+      removeRequest(request);
+      reject(
+        createSemaphoreError(
+          "SEMAPHORE_TIMEOUT",
+          `Semaphore timeout after ${timeoutMs}ms for ${keys.join(",")}`
+        )
+      );
+      drainQueues();
    }, timeoutMs);
-
-    timer.unref?.();
-
-    const queueItem: QueuedAcquire = {
-      resolve: (release) => {
-        cleanup();
-        resolve(release);
-      },
-      reject: (error) => {
-        cleanup();
-        reject(error);
-      },
-      timer,
-    };
-
-    gate.queue.push(queueItem);
-
+    request.timer.unref?.();
    if (signal) {
-      abortListener = () => {
-        cleanup();
-        clearTimeout(timer);
-
-        const nextGate = gates.get(semaphoreKey);
-        if (!nextGate) {
-          reject(makeAbortError(signal));
-          return;
-        }
-
-        const queueIndex = nextGate.queue.findIndex((item) => item.timer === timer);
-        if (queueIndex !== -1) {
-          nextGate.queue.splice(queueIndex, 1);
-        }
-
-        if (nextGate.running === 0 && nextGate.queue.length === 0) {
-          scheduleCleanup(semaphoreKey);
-        }
-
+      request.abortListener = () => {
+        if (request.settled) return;
+        request.settled = true;
+        removeRequest(request);
        reject(makeAbortError(signal));
+        drainQueues();
      };
-      if (signal.aborted) {
-        abortListener();
-      } else {
-        signal.addEventListener("abort", abortListener);
-      }
+      signal.addEventListener("abort", request.abortListener, { once: true });
    }
+    queuedRequests.add(request);
+    for (const key of keys) gates.get(key)!.queue.push({ request });
+    drainQueues();
  });
 }

-/**
- * Temporarily block new acquisitions for a key while allowing in-flight requests to finish.
- */
-export function markBlocked(semaphoreKey: string, cooldownMs: number): void {
-  const safeCooldownMs = Number.isFinite(cooldownMs) && cooldownMs > 0 ? cooldownMs : 0;
-  if (safeCooldownMs <= 0) {
-    const gate = gates.get(semaphoreKey);
-    if (!gate) return;
-    gate.blockedUntil = null;
-    drainQueue(semaphoreKey);
-    return;
-  }
-
-  const gate = gates.get(semaphoreKey) ?? ensureGate(semaphoreKey, 1);
+export function markBlocked(key: string, until: Date | string | number): void {
+  const untilMs =
+    until instanceof Date
+      ? until.getTime()
+      : typeof until === "number"
+        ? Date.now() + Math.max(0, until)
+        : new Date(until).getTime();
+  if (!Number.isFinite(untilMs) || untilMs <= Date.now()) return;
+  const gate = ensureGate(key, gates.get(key)?.maxConcurrency ?? 1);
  clearCleanupTimer(gate);
-  gate.blockedUntil = Date.now() + safeCooldownMs;
+  gate.blockedUntil = untilMs;
+}

-  const timer = setTimeout(() => {
-    const nextGate = gates.get(semaphoreKey);
-    if (!nextGate) return;
-    if (nextGate.blockedUntil && Date.now() >= nextGate.blockedUntil) {
-      nextGate.blockedUntil = null;
-      drainQueue(semaphoreKey);
-      if (nextGate.running === 0 && nextGate.queue.length === 0) {
-        scheduleCleanup(semaphoreKey);
-      }
-    }
-  }, safeCooldownMs + 50);
-
-  timer.unref?.();
+export function unblock(key: string): void {
+  const gate = gates.get(key);
+  if (!gate) return;
+  gate.blockedUntil = null;
+  drainQueues();
+  cleanupGateIfIdle(key);
 }

-/**
- * Return the current in-memory semaphore snapshot.
- */
 export function getStats(): Record<string, AccountSemaphoreStatsEntry> {
  const stats: Record<string, AccountSemaphoreStatsEntry> = {};
-
  for (const [key, gate] of gates) {
    stats[key] = {
      running: gate.running,
@@ -340,48 +330,50 @@ export function getStats(): Record<string, AccountSemaphoreStatsEntry> {
      blockedUntil: gate.blockedUntil ? new Date(gate.blockedUntil).toISOString() : null,
    };
  }
-
  return stats;
 }

-/**
- * Check if an account semaphore key is currently at or over its max concurrency limit.
- * Returns true if running >= maxConcurrency or blocked.
- */
 export function isAccountSemaphoreFull(
  provider: string,
  accountKey: string,
  maxConcurrency?: number | null
 ): boolean {
  if (isBypassed(maxConcurrency)) return false;
-  const key = buildAccountSemaphoreKey({ provider, accountKey });
-  const gate = gates.get(key);
+  const gate = gates.get(buildAccountSemaphoreKey({ provider, accountKey }));
  if (!gate) return false;
  const effectiveCap = maxConcurrency ?? gate.maxConcurrency;
-  if (isBypassed(effectiveCap)) return false;
-  return gate.running >= effectiveCap || isBlocked(gate);
+  return !isBypassed(effectiveCap) && (gate.running >= effectiveCap || isBlocked(gate));
 }

-/**
- * Reset a single key and reject queued waiters.
- */
-export function reset(semaphoreKey: string): void {
-  const gate = gates.get(semaphoreKey);
+export function reset(key: string): void {
+  const gate = gates.get(key);
  if (!gate) return;
-
  clearCleanupTimer(gate);
-  for (const entry of gate.queue) {
-    clearTimeout(entry.timer);
-    entry.reject(new Error("Semaphore reset"));
+  const error = createSemaphoreError("SEMAPHORE_RESET", `Semaphore reset for ${key}`);
+  const rejections: AcquireRequest[] = [];
+  for (const queued of [...gate.queue]) {
+    const request = queued.request;
+    if (request.settled) continue;
+    request.settled = true;
+    removeRequest(request);
+    rejections.push(request);
  }
-  gates.delete(semaphoreKey);
+  gates.delete(key);
+  for (const request of rejections) request.reject(error);
+  drainQueues();
 }

-/**
- * Reset all keys and reject queued waiters.
- */
 export function resetAll(): void {
-  for (const key of gates.keys()) {
-    reset(key);
+  const error = createSemaphoreError("SEMAPHORE_RESET", "Semaphore reset");
+  const rejections: AcquireRequest[] = [];
+  for (const request of [...queuedRequests]) {
+    if (request.settled) continue;
+    request.settled = true;
+    removeRequest(request);
+    rejections.push(request);
  }
+  for (const gate of gates.values()) clearCleanupTimer(gate);
+  gates.clear();
+  queuedRequests.clear();
+  for (const request of rejections) request.reject(error);
 }
--- a/src/app/(dashboard)/dashboard/settings/components/ResilienceTab.tsx
+++ b/src/app/(dashboard)/dashboard/settings/components/ResilienceTab.tsx
@@ -13,6 +13,7 @@ type RequestQueueSettings = {
  requestsPerMinute: number;
  minTimeBetweenRequestsMs: number;
  concurrentRequests: number;
+  globalConcurrentRequests: number;
  maxWaitMs: number;
 };

@@ -240,13 +241,21 @@ function RequestQueueCard({
              }
            />
            <NumberField
-              label={t("resilienceConcurrentRequests")}
+              label={t("resilienceConnectionScopeConcurrentRequests")}
              value={draft.concurrentRequests}
              min={1}
              onChange={(concurrentRequests) =>
                setDraft((prev) => ({ ...prev, concurrentRequests }))
              }
            />
+            <NumberField
+              label={t("resilienceGlobalConcurrentRequests")}
+              value={draft.globalConcurrentRequests}
+              min={0}
+              onChange={(globalConcurrentRequests) =>
+                setDraft((prev) => ({ ...prev, globalConcurrentRequests }))
+              }
+            />
            <NumberField
              label={t("resilienceMaxQueueWait")}
              value={draft.maxWaitMs}
@@ -278,11 +287,21 @@ function RequestQueueCard({
              </div>
            </div>
            <div className="rounded-xl border border-border bg-bg-subtle p-4">
-              <div className="text-xs text-text-muted">{t("resilienceConcurrentRequests")}</div>
+              <div className="text-xs text-text-muted">
+                {t("resilienceConnectionScopeConcurrentRequests")}
+              </div>
              <div className="mt-1 text-sm font-semibold text-text-main">
                {value.concurrentRequests}
              </div>
            </div>
+            <div className="rounded-xl border border-border bg-bg-subtle p-4">
+              <div className="text-xs text-text-muted">
+                {t("resilienceGlobalConcurrentRequests")}
+              </div>
+              <div className="mt-1 text-sm font-semibold text-text-main">
+                {value.globalConcurrentRequests || t("statusDisabled")}
+              </div>
+            </div>
            <div className="rounded-xl border border-border bg-bg-subtle p-4">
              <div className="text-xs text-text-muted">{t("resilienceMaxQueueWait")}</div>
              <div className="mt-1 text-sm font-semibold text-text-main">
--- a/src/i18n/messages/de.json
+++ b/src/i18n/messages/de.json
@@ -7538,6 +7538,8 @@
    "resilienceRequestsPerMinute": "Requests per minute",
    "resilienceMinTimeBetweenRequests": "Minimum time between requests",
    "resilienceConcurrentRequests": "Concurrent requests",
+    "resilienceConnectionScopeConcurrentRequests": "Gleichzeitige Anfragen pro Verbindung/Quota-Bereich",
+    "resilienceGlobalConcurrentRequests": "Globale gleichzeitige Upstream-Anfragen (0 = deaktiviert)",
    "resilienceMaxQueueWaitTime": "Maximale Wartezeit in der Warteschlange",
    "resilienceBaseCooldown": "Base cooldown",
    "resilienceUseUpstreamRetryHints": "Use upstream retry hints",
--- a/src/i18n/messages/en.json
+++ b/src/i18n/messages/en.json
@@ -7550,6 +7550,8 @@
    "resilienceRequestsPerMinute": "Requests per minute",
    "resilienceMinTimeBetweenRequests": "Minimum time between requests",
    "resilienceConcurrentRequests": "Concurrent requests",
+    "resilienceConnectionScopeConcurrentRequests": "Concurrent requests per connection/quota scope",
+    "resilienceGlobalConcurrentRequests": "Global concurrent upstream requests (0 = disabled)",
    "resilienceMaxQueueWaitTime": "Maximum queue wait time",
    "resilienceBaseCooldown": "Base cooldown",
    "resilienceUseUpstreamRetryHints": "Use upstream retry hints",
--- a/src/i18n/messages/pt-BR.json
+++ b/src/i18n/messages/pt-BR.json
@@ -7550,6 +7550,8 @@
    "resilienceRequestsPerMinute": "Requests per minute",
    "resilienceMinTimeBetweenRequests": "Minimum time between requests",
    "resilienceConcurrentRequests": "Concurrent requests",
+    "resilienceConnectionScopeConcurrentRequests": "Solicitações simultâneas por conexão/escopo de cota",
+    "resilienceGlobalConcurrentRequests": "Solicitações upstream simultâneas globais (0 = desativado)",
    "resilienceMaxQueueWaitTime": "Tempo máximo de espera na fila",
    "resilienceBaseCooldown": "Base cooldown",
    "resilienceUseUpstreamRetryHints": "Use upstream retry hints",
--- a/src/i18n/messages/vi.json
+++ b/src/i18n/messages/vi.json
@@ -7550,6 +7550,8 @@
    "resilienceRequestsPerMinute": "Số yêu cầu mỗi phút",
    "resilienceMinTimeBetweenRequests": "Thời gian tối thiểu giữa các yêu cầu",
    "resilienceConcurrentRequests": "Số yêu cầu đồng thời",
+    "resilienceConnectionScopeConcurrentRequests": "Số yêu cầu đồng thời trên mỗi kết nối/phạm vi hạn ngạch",
+    "resilienceGlobalConcurrentRequests": "Số yêu cầu thượng nguồn đồng thời toàn cục (0 = tắt)",
    "resilienceMaxQueueWaitTime": "Thời gian chờ hàng đợi tối đa",
    "resilienceBaseCooldown": "Thời gian giãn cách cơ bản",
    "resilienceUseUpstreamRetryHints": "Sử dụng gợi ý thử lại từ máy chủ thượng nguồn",
--- a/src/lib/resilience/settings.ts
+++ b/src/lib/resilience/settings.ts
@@ -58,6 +58,7 @@ export const DEFAULT_RESILIENCE_SETTINGS: ResilienceSettings = {
    requestsPerMinute: DEFAULT_API_LIMITS.requestsPerMinute,
    minTimeBetweenRequestsMs: DEFAULT_API_LIMITS.minTimeBetweenRequests,
    concurrentRequests: DEFAULT_API_LIMITS.concurrentRequests,
+    globalConcurrentRequests: 0,
    maxWaitMs: DEFAULT_REQUEST_QUEUE_MAX_WAIT_MS,
    maxQueueDepth: DEFAULT_REQUEST_QUEUE_MAX_DEPTH,
  },
@@ -208,6 +209,8 @@ function buildLegacyFallback(settings: JsonRecord): ResilienceSettings {
        DEFAULT_RESILIENCE_SETTINGS.requestQueue.concurrentRequests,
        { min: 1, max: 10_000 }
      ),
+      globalConcurrentRequests:
+        DEFAULT_RESILIENCE_SETTINGS.requestQueue.globalConcurrentRequests,
      maxWaitMs: DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxWaitMs,
      maxQueueDepth: DEFAULT_RESILIENCE_SETTINGS.requestQueue.maxQueueDepth,
    },
--- a/src/lib/resilience/settings/normalize.ts
+++ b/src/lib/resilience/settings/normalize.ts
@@ -124,6 +124,11 @@ export function normalizeRequestQueueSettings(
    min: 1,
    max: 10_000,
  });
+  const globalConcurrentRequests = toInteger(
+    record.globalConcurrentRequests,
+    fallback.globalConcurrentRequests,
+    { min: 0, max: 100_000 }
+  );
  const maxWaitMs = toInteger(record.maxWaitMs, fallback.maxWaitMs, {
    min: 1,
    max: 24 * 60 * 60 * 1000,
@@ -141,6 +146,7 @@ export function normalizeRequestQueueSettings(
    requestsPerMinute,
    minTimeBetweenRequestsMs,
    concurrentRequests,
+    globalConcurrentRequests,
    maxWaitMs,
    maxQueueDepth,
  };
@@ -431,8 +437,16 @@ function normalizeProviderQuotaOverrideEntry(raw: unknown): ProviderQuotaOverrid
  const out: ProviderQuotaOverrideSettings = {};
  const rpm = typeof record.rpm === "number" ? record.rpm : Number(record.rpm);
  if (Number.isFinite(rpm) && rpm > 0) out.rpm = Math.trunc(rpm);
-  const concurrency = typeof record.concurrency === "number" ? record.concurrency : Number(record.concurrency);
+  const concurrency =
+    typeof record.concurrency === "number" ? record.concurrency : Number(record.concurrency);
  if (Number.isFinite(concurrency) && concurrency > 0) out.concurrency = Math.trunc(concurrency);
+  const providerConcurrency =
+    typeof record.providerConcurrency === "number"
+      ? record.providerConcurrency
+      : Number(record.providerConcurrency);
+  if (Number.isFinite(providerConcurrency) && providerConcurrency >= 0) {
+    out.providerConcurrency = Math.trunc(providerConcurrency);
+  }
  return Object.keys(out).length > 0 ? out : null;
 }

--- a/src/lib/resilience/settings/types.ts
+++ b/src/lib/resilience/settings/types.ts
@@ -16,6 +16,8 @@ export interface RequestQueueSettings {
  requestsPerMinute: number;
  minTimeBetweenRequestsMs: number;
  concurrentRequests: number;
+  /** Whole-process upstream concurrency cap. Zero disables the global gate. */
+  globalConcurrentRequests: number;
  /**
   * Legacy persisted key used as Bottleneck's post-dispatch execution
   * expiration. It does not bound time spent in Bottleneck's QUEUED state.
@@ -169,6 +171,8 @@ export interface ProviderQuotaOverrideSettings {
  rpm?: number;
  /** Overrides the static per-connection concurrency cap. */
  concurrency?: number;
+  /** Shared concurrency cap across every connection for this provider. */
+  providerConcurrency?: number;
 }

 export interface StreamRecoverySettings {
--- a/src/shared/validation/schemas/settings.ts
+++ b/src/shared/validation/schemas/settings.ts
@@ -32,6 +32,7 @@ export const legacyResilienceDefaultsSchema = z
    requestsPerMinute: z.number().int().min(1).optional(),
    minTimeBetweenRequests: z.number().int().min(0).optional(),
    concurrentRequests: z.number().int().min(1).optional(),
+    globalConcurrentRequests: z.number().int().min(0).max(100_000).optional(),
  })
  .strict();

@@ -170,6 +171,7 @@ export const updateResilienceSchema = z
          .object({
            rpm: z.number().int().min(1).optional(),
            concurrency: z.number().int().min(1).optional(),
+            providerConcurrency: z.number().int().min(0).max(100_000).optional(),
          })
          .strict()
      )
--- a/tests/e2e/resilience-plan-alignment.spec.ts
+++ b/tests/e2e/resilience-plan-alignment.spec.ts
@@ -7,6 +7,7 @@ const resilienceSettings = {
    requestsPerMinute: 100,
    minTimeBetweenRequestsMs: 200,
    concurrentRequests: 10,
+    globalConcurrentRequests: 0,
    maxWaitMs: 120000,
  },
  connectionCooldown: {
--- a/tests/unit/accountSemaphore.test.ts
+++ b/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({
--- a/tests/unit/chatcore-hierarchical-admission.test.ts
+++ b/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");
+});
--- a/tests/unit/resilience-settings-provider-quota-overrides.test.ts
+++ b/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".
				`@@ -0,0 +1 @@`
				`feat(resilience): atomically enforce cumulative global, provider, and account concurrency limits`