From ece486dc38fee169c7c7a26de63089eeeeb909a2 Mon Sep 17 00:00:00 2001
From: Diego Rodrigues de Sa e Souza <diegosouza.pw@gmail.com>
Date: Thu, 6 Aug 2026 10:41:04 -0300
Subject: [PATCH] fix(resilience): enforce RPM with rolling leases (#9604)

Validated in local merge-train (diegosouzapw batch)
---
 changelog.d/fixes/9259-rolling-rpm-leases.md  |   1 +
 config/quality/eslint-suppressions.json       |   5 -
 config/quality/file-size-baseline.json        |   5 +-
 open-sse/services/providerDefaultRateLimit.ts |   6 +-
 open-sse/services/rateLimitManager.ts         | 408 ++++++++--------
 open-sse/services/rollingRpmGate.ts           | 236 +++++++++
 open-sse/services/slidingWindowLimiter.ts     |  94 +++-
 tests/unit/chat-rate-limit-body-lock.test.ts  |   7 +-
 tests/unit/rate-limit-manager.test.ts         | 446 +++++++++++++++++-
 ...e-limit-queue-timeout-message-4165.test.ts |  97 ++--
 tests/unit/sliding-window-limiter.test.ts     |  73 ++-
 11 files changed, 1089 insertions(+), 289 deletions(-)
 create mode 100644 changelog.d/fixes/9259-rolling-rpm-leases.md
 create mode 100644 open-sse/services/rollingRpmGate.ts

diff --git a/changelog.d/fixes/9259-rolling-rpm-leases.md b/changelog.d/fixes/9259-rolling-rpm-leases.md
new file mode 100644
index 0000000000..a20f6c4367
--- /dev/null
+++ b/changelog.d/fixes/9259-rolling-rpm-leases.md
@@ -0,0 +1 @@
+- **fix(resilience):** Enforce RPM limits with rolling leases and atomic global/provider/account admission ([#9259](https://github.com/diegosouzapw/OmniRoute/pull/9259)). The configured global RPM budget is shared across all enabled provider connections within one process; provider/account overrides add narrower scopes.
diff --git a/config/quality/eslint-suppressions.json b/config/quality/eslint-suppressions.json
index 7373467d5b..c45f616662 100644
--- a/config/quality/eslint-suppressions.json
+++ b/config/quality/eslint-suppressions.json
@@ -249,11 +249,6 @@
       "count": 1
     }
   },
-  "open-sse/services/rateLimitManager.ts": {
-    "no-restricted-syntax": {
-      "count": 1
-    }
-  },
   "open-sse/services/taskAwareRouter.ts": {
     "@typescript-eslint/no-explicit-any": {
       "count": 4
diff --git a/config/quality/file-size-baseline.json b/config/quality/file-size-baseline.json
index b0bcf0cc8c..c498ceefaa 100644
--- a/config/quality/file-size-baseline.json
+++ b/config/quality/file-size-baseline.json
@@ -1,5 +1,6 @@
 {
-  "_rebaseline_2026_07_24_8470_hyperagent_sticky_thread": "PR #8470 (artickc, fix/hyperagent-tool-loop-thread-sticky) own growth: open-sse/executors/hyperagent.ts 936->1025 (wc -l; check-file-size.mjs counts via split(\"\\n\").length so the gate sees 937->1026, +89, crosses the 1000 cap). Fixes a real bug where a reverse-conversion proxy (text-Intent/JSON to Claude Code native tool_calls) rewrites assistant messages between agentic tool-loop turns, breaking HyperAgent's conversation-prefix fingerprint and cold-starting the thread mid tool-loop. Adds Anthropic tool_use/tool_result flattening to extractMessageText() plus a new rootUserFingerprint()/root-key lookup tier in resolveHyperAgentThreadBinding()/storeHyperAgentThreadAfterTurn() so the thread stays sticky across the tool loop. Cohesive additions inside the existing single-file executor; not extractable without splitting the executor mid-request-flow. Covered by tests/unit/executor-hyperagent.test.ts (19/19, +5 new cases for tool_result/tool_use flattening + root-key stickiness). Pre-merge review flagged a cross-conversation root-key collision risk (tracked in the PR's own mandatory pre-merge checklist, not yet addressed) — unrelated to this file-size ratchet, tracked separately by /fix-prs.",
+  "_rebaseline_2026_08_02_9259_rolling_rpm": "PR #9259 (issue #8733) own growth: open-sse/services/rateLimitManager.ts baseline 1060->1167 (+107; final source 1153). The existing withRateLimit chokepoint now composes process-local rolling RPM leases with Bottleneck admission, releases pre-dispatch leases on queue timeout/abort/connection disable, preserves caller abort reasons, and wires 429/header state into the extracted rollingRpmGate.ts. The remaining growth is irreducible lifecycle wiring at the dispatch boundary plus the real watchdog test hooks needed to verify queued-wedge recovery; moving it further would obscure lease ownership and Bottleneck cleanup. Covered by the focused rate-limit manager/sliding-window suite (33/33); distributed multi-instance coordination remains explicitly out of scope.",
+  "_rebaseline_2026_07_24_8470_hyperagent_sticky_thread": "PR #8470 (artickc, fix/hyperagent-tool-loop-thread-sticky) own growth: open-sse/executors/hyperagent.ts 936->1025 (wc -l; check-file-size.mjs counts via split(\"\\n\").length so the gate sees 937->1026, +89, crosses the 1000 cap). Fixes a real bug where a reverse-conversion proxy (text-Intent/JSON to Claude Code native tool_calls) rewrites assistant messages between agentic tool-loop turns, breaking HyperAgent’s conversation-prefix fingerprint and cold-starting the thread mid tool-loop. Adds Anthropic tool_use/tool_result flattening to extractMessageText() plus a new rootUserFingerprint()/root-key lookup tier in resolveHyperAgentThreadBinding()/storeHyperAgentThreadAfterTurn() so the thread stays sticky across the tool loop. Cohesive additions inside the existing single-file executor; not extractable without splitting the executor mid-request-flow. Covered by tests/unit/executor-hyperagent.test.ts (19/19, +5 new cases for tool_result/tool_use flattening + root-key stickiness). Pre-merge review flagged a cross-conversation root-key collision risk (tracked in the PR’s own mandatory pre-merge checklist, not yet addressed) — unrelated to this file-size ratchet, tracked separately by /fix-prs.",
   "_rebaseline_2026_07_25_8494_capability_filter_fail_closed": "PR #8494 (fix/capability-filters-fail-closed, #8488) own growth: open-sse/services/combo.ts 3640->3693 (+53) adds a fail-closed guard after filterTargetsByRequestCompatibility() — when every eligible target is excluded by request-capability filtering (vision/tools/etc) instead of quota/health, the combo now returns an explicit `capability_mismatch` 400 (describeCapabilityFilterExhaustion, imported from combo/comboStructure.ts) rather than silently falling through to a generic no-targets error, plus a `compatFilterFailOpen` escape hatch (combo config OR settings) mirrored at both the main/auto and round-robin call sites for symmetry. combo/comboStructure.ts (previously under cap, un-frozen) grows 794->918 (+124) — new home for describeCapabilityFilterExhaustion + providerSupportsEmulatedToolCalling (#5240 emulated tool-calling exemption so fail-closed does not regress prompt-emulation-only combos like all-chatgpt-web). Irreducible orchestration wiring at the existing filter chokepoint (same precedent as #7301's universal-cooldown-retry generalization). Companion test tests/unit/combo-routing-engine.test.ts 3409->3449 (+40, fail-closed/fail-open coverage across both call sites) also rebaselined. Covered by tests/unit/8488-capability-filter-fail-closed.test.ts (new) + 95/95 passing across both files. Structural shrink of combo.ts tracked in #3501.",
   "_rebaseline_2026_07_25_8499_ts7_result_union_predicates": "PR #8499 (backryun, chore/ts7-types-executor-scattered) own growth: muse-spark-web.ts 1396->1405 (+9, irreducible). Under this workspace's `strictNullChecks: false`, the boolean-literal discriminant on `GraphqlResult` (`{ ok: true } | { ok: false; error: string }`) narrows the positive `.ok===true` branch but leaves `!result.ok` at the full union under TS7, making `.error` unreachable to the checker at the two call sites (warmup, mode-switch). Fixed by adding a single `isGraphqlFailure()` type-predicate helper (doc comment + 3-line body) reused at both call sites instead of duplicating the predicate inline — not extractable to a shared module without splitting a single-file executor's local narrowing helper out of its own file. Covered by the existing muse-spark-web executor test suite (no behavior change, pure narrowing fix).",
   "_rebaseline_2026_07_22_8131_windowshide_cloudflared_spawn": "PR #8167 (Dingding-leo, fix/windows-hide-child-process, #8131) own growth: src/lib/cloudflaredTunnel.ts 934->935 (+1, irreducible call-site wiring — the single `windowsHide: true` option added to the existing cloudflared spawn() options object so no transient conhost.exe/cmd console window flashes open on Windows). Covered by the pre-merge-fix regression test tests/unit/windows-hide-child-process-spawns-8131.test.ts (added for the two additional spawn() sites the PR missed: ServiceSupervisor.ts, versionManager/processManager.ts) plus the windowsHide assertion added to tests/unit/services/installers/runNpm-shell-5379.test.ts (installers/utils.ts buildNpmExecOptions).",
@@ -544,7 +545,7 @@
     "open-sse/services/claudeCodeCompatible.ts": 1202,
     "open-sse/services/combo.ts": 3648,
     "open-sse/services/compression/strategySelector.ts": 1060,
-    "open-sse/services/rateLimitManager.ts": 1105,
+    "open-sse/services/rateLimitManager.ts": 1167,
     "open-sse/translator/response/openai-responses.ts": 1204,
     "open-sse/utils/cursorAgentProtobuf.ts": 1505,
     "open-sse/utils/stream.ts": 2889,
diff --git a/open-sse/services/providerDefaultRateLimit.ts b/open-sse/services/providerDefaultRateLimit.ts
index 5560c71e76..8df2ab9624 100644
--- a/open-sse/services/providerDefaultRateLimit.ts
+++ b/open-sse/services/providerDefaultRateLimit.ts
@@ -11,8 +11,10 @@
  * API (#6846 Phase 1). Every other provider still gets zero behavior change; the
  * whole path is a no-op unless an entry (or a resolved override, see below) exists.
  *
- * Wired as a pre-schedule gate in `withRateLimit` (rateLimitManager.ts). Bottleneck
- * still applies on top — this only adds a floor for header-less providers.
+ * Composed into the rolling lease gate in `withRateLimit` (rateLimitManager.ts).
+ * The exported acquire helpers remain available for provider-specific callers
+ * and tests, while the main request path acquires global and provider scopes
+ * atomically.
  */
 import { SlidingWindowLimiter, type RateLimitWindow } from "./slidingWindowLimiter.ts";
 
diff --git a/open-sse/services/rateLimitManager.ts b/open-sse/services/rateLimitManager.ts
index 87f48a46fa..cb4350b2f8 100644
--- a/open-sse/services/rateLimitManager.ts
+++ b/open-sse/services/rateLimitManager.ts
@@ -13,7 +13,12 @@ import { parseRetryAfterFromBody } from "./accountFallback.ts";
 import { getAntigravityQuotaFamily } from "./antigravityQuotaFamily.ts";
 import { getProviderCategory } from "../config/providerRegistry.ts";
 import { getCodexRateLimitKey } from "../executors/codex.ts";
-import { awaitProviderDefaultSlot, setProviderQuotaOverrides } from "./providerDefaultRateLimit.ts";
+import {
+  getProviderDefaultRateLimit,
+  setProviderQuotaOverrides,
+} from "./providerDefaultRateLimit.ts";
+import { keyContainsConnection, RollingRpmGate } from "./rollingRpmGate.ts";
+import { toNumber } from "@/shared/utils/numeric";
 import {
   DEFAULT_RESILIENCE_SETTINGS,
   resolveResilienceSettings,
@@ -39,25 +44,36 @@ interface LearnedLimitEntry {
 interface LimiterUpdateSettings {
   maxConcurrent?: number | null;
   minTime: number;
-  reservoir?: number | null;
-  reservoirRefreshAmount?: number | null;
-  reservoirRefreshInterval?: number | null;
 }
 
 type JsonRecord = Record<string, unknown>;
+type QueueTimeoutReason = "local-queue" | "upstream-cooldown";
 
 function toRecord(value: unknown): JsonRecord {
   return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {};
 }
 
-function toNumber(value: unknown, fallback = 0): number {
-  const parsed =
-    typeof value === "number"
-      ? value
-      : typeof value === "string" && value.trim().length > 0
-        ? Number(value)
-        : Number.NaN;
-  return Number.isFinite(parsed) ? parsed : fallback;
+function createQueueTimeoutError(
+  provider: string,
+  model: string | null,
+  maxWaitMs: number,
+  reason: QueueTimeoutReason = "local-queue",
+  cause?: unknown
+) {
+  const target = model ? `${provider}/${model}` : provider;
+  const message =
+    reason === "upstream-cooldown"
+      ? `Request dropped after waiting ${maxWaitMs}ms for an upstream rate-limit cooldown for ${target}. ` +
+        `The provider cooldown outlasted OmniRoute's local wait budget; this is not local queue saturation.`
+      : `Request dropped after exceeding the local rate-limit queue budget maxWaitMs (${maxWaitMs}ms) for ` +
+        `${target} — this is OmniRoute's request queue ` +
+        `(resilienceSettings.requestQueue.maxWaitMs), not an upstream timeout. Raise it in ` +
+        `Settings → Resilience if this is queue saturation rather than a slow provider.`;
+  const queueErr = new Error(message, cause === undefined ? undefined : { cause }) as Error & {
+    code?: string;
+  };
+  queueErr.code = "RATE_LIMIT_QUEUE_TIMEOUT";
+  return queueErr;
 }
 
 function isNodeTestRunnerChild(): boolean {
@@ -101,16 +117,15 @@ let initialized = false;
 let currentRequestQueueSettings: RequestQueueSettings = DEFAULT_RESILIENCE_SETTINGS.requestQueue;
 
 // Watchdog: detect Bottleneck limiters that are wedged (queue has work, but no
-// jobs are dispatched). When the reservoir/refresh state desyncs from reality,
-// this catches it and force-resets so traffic isn't stuck forever.
+// jobs are dispatched). RPM admission happens before Bottleneck, so a queued
+// Bottleneck job with no active work is a concurrency scheduler failure.
 const lastDispatchAt = new Map<string, number>();
+let nextJobTraceId = 1;
 let watchdogInterval: ReturnType<typeof setInterval> | null = null;
 const WATCHDOG_INTERVAL_MS = 30_000;
-// Threshold has to exceed any *legitimate* gap between dispatches:
-//  - default reservoirRefreshInterval is 60s
-//  - adaptive minTime can climb to ~60s for 1-RPM providers (see updateFromHeaders)
-// 120s gives a 2× margin against both, while still catching the actual wedge
-// case we observed (queue stalled for 3+ minutes with no progress).
+// Threshold has to exceed any legitimate gap caused by adaptive minTime while
+// still catching the actual wedge case we observed (queue stalled for 3+
+// minutes with no progress).
 const WEDGE_THRESHOLD_MS = 120_000;
 
 /**
@@ -128,19 +143,10 @@ function isAutoEnableActive(settings: RequestQueueSettings): boolean {
   return settings.autoEnableApiKeyProviders;
 }
 
-// Sentinels for "no rate limit" / effectively infinite capacity. The reservoir
-// value uses Number.MAX_SAFE_INTEGER so the bucket can never realistically be
-// exhausted; maxConcurrent uses a smaller-but-still-vast ceiling since
-// Bottleneck tracks concurrent jobs in memory and an unbounded number would
-// risk internal counter overflow under sustained pressure.
-const EFFECTIVELY_INFINITE = Number.MAX_SAFE_INTEGER;
+// Bottleneck handles concurrency and pacing. RPM is enforced by the rolling
+// lease limiter above rather than by a fixed-window reservoir.
 const EFFECTIVELY_INFINITE_CONCURRENCY = 1000;
 
-// Resolve an RPM override. 0 or missing means "infinite" (no rate cap).
-function resolveRpm(override: number | undefined | null): number {
-  return typeof override === "number" && override > 0 ? override : EFFECTIVELY_INFINITE;
-}
-
 // Resolve a minTime override. 0 or missing means "no minimum gap".
 function resolveMinTime(override: number | undefined | null): number {
   return typeof override === "number" && override > 0 ? override : 0;
@@ -152,15 +158,9 @@ function resolveMaxConcurrent(override: number | undefined | null): number {
 }
 
 function buildLimiterDefaults() {
-  // 0 or missing values mean "infinite" / no rate limit applies. This treats
-  // the global request-queue settings the same way per-connection overrides
-  // are interpreted (see resolveRpm / resolveMinTime / resolveMaxConcurrent).
   return {
     maxConcurrent: resolveMaxConcurrent(currentRequestQueueSettings.concurrentRequests),
     minTime: resolveMinTime(currentRequestQueueSettings.minTimeBetweenRequestsMs),
-    reservoir: resolveRpm(currentRequestQueueSettings.requestsPerMinute),
-    reservoirRefreshAmount: resolveRpm(currentRequestQueueSettings.requestsPerMinute),
-    reservoirRefreshInterval: 60 * 1000,
   };
 }
 
@@ -271,12 +271,18 @@ function reconcileEnabledConnections(
 
 function watchdogTick() {
   const now = Date.now();
+  rpmGate.cleanupExpired(now);
   // Clean up idle limiters that haven't been used recently
   for (const [key, limiter] of Array.from(limiters)) {
     const lastUsed = limiterLastUsed.get(key) ?? 0;
     if (now - lastUsed > INACTIVE_LIMITER_MS) {
       const counts = limiter.counts();
-      if (counts.QUEUED === 0 && counts.RUNNING === 0 && counts.EXECUTING === 0) {
+      if (
+        counts.RECEIVED === 0 &&
+        counts.QUEUED === 0 &&
+        counts.RUNNING === 0 &&
+        counts.EXECUTING === 0
+      ) {
         limiters.delete(key);
         lastDispatchAt.delete(key);
         limiterLastUsed.delete(key);
@@ -289,7 +295,10 @@ function watchdogTick() {
   }
   for (const [key, limiter] of Array.from(limiters)) {
     const counts = limiter.counts();
-    if (counts.QUEUED === 0) continue;
+    // RECEIVED-only work is still active and must not be evicted. Once a job
+    // is stably queued, Bottleneck reports it in QUEUED with RECEIVED=0; that
+    // is the state the wedge detector is designed to recover.
+    if (counts.RECEIVED > 0 || counts.QUEUED === 0) continue;
     if (counts.RUNNING > 0 || counts.EXECUTING > 0) continue;
     const lastDispatch = lastDispatchAt.get(key);
     // No heartbeat yet → seed it and skip this tick. Prevents false wedge
@@ -302,7 +311,7 @@ function watchdogTick() {
     if (stalledMs < WEDGE_THRESHOLD_MS) continue;
 
     warnRateLimit(
-      `🚨 [RATE-LIMIT] WEDGED: ${key} queued=${counts.QUEUED} running=0 executing=0 stalled=${stalledMs}ms — force-resetting`
+      `🚨 [RATE-LIMIT] WEDGED: ${key} received=${counts.RECEIVED} queued=${counts.QUEUED} running=0 executing=0 stalled=${stalledMs}ms — force-resetting`
     );
     // Live incident (log id 1784465227489-a2cbc0): disconnect() releases the
     // heartbeat timer but does NOT reject the QUEUED jobs already sitting on
@@ -353,15 +362,38 @@ export function stopRateLimitWatchdog(): void {
   watchdogInterval = null;
 }
 
+export function __installLimiterForTests(
+  provider: string,
+  connectionId: string,
+  limiter: Bottleneck,
+  model = null
+): void {
+  const key = getLimiterKey(provider, connectionId, model);
+  limiters.set(key, limiter);
+  lastDispatchAt.set(key, Date.now());
+  limiterLastUsed.set(key, Date.now());
+}
+
+export function __runRateLimitWatchdogForTests(): void {
+  watchdogTick();
+}
+
+export function __getLimiterForTests(provider: string, connectionId: string, model = null) {
+  return getLimiter(provider, connectionId, model);
+}
+
+export function __setLastDispatchAtForTests(
+  provider: string,
+  connectionId: string,
+  model: string | null,
+  timestamp: number
+): void {
+  lastDispatchAt.set(getLimiterKey(provider, connectionId, model), timestamp);
+}
+
 function evictWedgeLimiter(key: string, limiter: Bottleneck): void {
   if (limiters.get(key) !== limiter) return;
-  limiters.delete(key);
-  lastDispatchAt.delete(key);
-  limiterLastUsed.delete(key);
-  trackAsyncOperation(limiter.disconnect());
-  trackAsyncOperation(
-    limiter.stop({ dropWaitingJobs: true, dropErrorMessage: "rate-limit-watchdog-wedge-reset" })
-  );
+  evictLimiterAndDropQueued(key, limiter, "rate-limit-watchdog-wedge-reset");
 }
 
 /**
@@ -422,13 +454,10 @@ export async function initializeRateLimits() {
     // budget + concurrency cap (nvidia today). No-op for every provider without
     // an entry in either providerQuotaOverrides or PROVIDER_DEFAULT_RATE_LIMITS.
     setProviderQuotaOverrides(resilience.providerQuotaOverrides);
-    const { explicitCount, autoCount } = reconcileEnabledConnections(
-      connections as unknown[],
-      currentRequestQueueSettings
-    );
-    await updateAllLimiterSettings();
 
-    // Load per-connection rate limit overrides
+    // Load per-connection rate limit overrides before reconciliation can create
+    // any limiter. The RPM gate reads these overrides at admission time, and
+    // Bottleneck still needs the non-RPM connection settings immediately.
     connectionRateLimitOverrides.clear();
     for (const conn of connections as Array<Record<string, unknown>>) {
       const overrides = conn.rateLimitOverrides;
@@ -437,6 +466,12 @@ export async function initializeRateLimits() {
       }
     }
 
+    const { explicitCount, autoCount } = reconcileEnabledConnections(
+      connections as unknown[],
+      currentRequestQueueSettings
+    );
+    updateAllLimiterSettings();
+
     if (explicitCount > 0 || autoCount > 0) {
       logRateLimit(
         `🛡️ [RATE-LIMIT] Loaded ${explicitCount} explicit + ${autoCount} auto-enabled protection(s)`
@@ -474,22 +509,14 @@ export function enableRateLimitProtection(connectionId) {
  */
 export function disableRateLimitProtection(connectionId) {
   enabledConnections.delete(connectionId);
-  // Evict limiters for this connection from the cache. Do NOT call limiter.stop() —
-  // it permanently rejects future .schedule() calls with "This limiter has been stopped",
-  // and in-flight requests holding a reference to the old instance would fail with 502.
-  // Call disconnect() (not stop()) to release Bottleneck's internal heartbeat timer
-  // without permanently poisoning the instance for any remaining in-flight jobs.
-  // Eviction-only would leak the heartbeat timer until GC; disconnect() releases it
-  // synchronously so the runtime memory footprint stays flat under heavy connection churn.
-  // .stop() is reserved exclusively for SIGTERM/SIGINT shutdown (see shutdownLimiters).
+  // Drop queued jobs before evicting the limiter. Otherwise disconnect() leaves
+  // callers waiting on an instance that is no longer reachable from the cache.
   for (const [key, limiter] of Array.from(limiters)) {
-    if (key.includes(connectionId)) {
-      limiters.delete(key);
-      lastDispatchAt.delete(key);
-      limiterLastUsed.delete(key);
-      trackAsyncOperation(limiter.disconnect());
+    if (keyContainsConnection(key, connectionId)) {
+      evictLimiterAndDropQueued(key, limiter, "rate-limit-connection-disabled");
     }
   }
+  rpmGate.clearConnection(connectionId);
 }
 
 /**
@@ -517,13 +544,11 @@ export function refreshConnectionRateLimits(connectionId, overrides) {
   }
   // Evict limiters referencing this connection so they get recreated on next use
   for (const [key, limiter] of Array.from(limiters)) {
-    if (key.includes(connectionId)) {
-      limiters.delete(key);
-      lastDispatchAt.delete(key);
-      limiterLastUsed.delete(key);
-      trackAsyncOperation(limiter.disconnect());
+    if (keyContainsConnection(key, connectionId)) {
+      evictLimiterAndDropQueued(key, limiter, "rate-limit-settings-refresh");
     }
   }
+  rpmGate.clearConnection(connectionId);
 }
 
 /**
@@ -546,6 +571,15 @@ function getLimiterKey(provider, connectionId, model = null) {
   return `${provider}:${connectionId}`;
 }
 
+const rpmGate = new RollingRpmGate({
+  getGlobalRpm: () => currentRequestQueueSettings.requestsPerMinute,
+  getProviderWindow: getProviderDefaultRateLimit,
+  getConnectionRpm: (connectionId) => connectionRateLimitOverrides.get(connectionId)?.rpm,
+  getLimiterKey,
+  createQueueTimeoutError: (provider, model, maxWaitMs, reason) =>
+    createQueueTimeoutError(provider, model, maxWaitMs, reason),
+});
+
 function getLimiter(provider, connectionId, model = null) {
   const key = getLimiterKey(provider, connectionId, model);
 
@@ -554,23 +588,14 @@ function getLimiter(provider, connectionId, model = null) {
     const overrides = connectionRateLimitOverrides.get(connectionId);
     if (overrides) {
       // 0 (or missing) means "no override — fall through to buildLimiterDefaults()".
-      // Without this guard, an rpm of 0 sets reservoir=0, which Bottleneck treats
-      // as "depleted" and blocks ALL requests indefinitely. Treating 0 as "use
-      // default" lets users effectively disable per-connection limits without
-      // globally raising the system default.
       if (typeof overrides.maxConcurrent === "number" && overrides.maxConcurrent > 0) {
         defaults.maxConcurrent = overrides.maxConcurrent;
       }
       if (typeof overrides.minTime === "number" && overrides.minTime > 0) {
         defaults.minTime = overrides.minTime;
       }
-      if (typeof overrides.rpm === "number" && overrides.rpm > 0) {
-        defaults.reservoir = overrides.rpm;
-        defaults.reservoirRefreshAmount = overrides.rpm;
-        defaults.reservoirRefreshInterval = 60 * 1000;
-      }
       // TODO: TPM/TPD integration — requires a token-bucket vs request-bucket
-      // separation (Bottleneck's reservoir is request-count, not token-count).
+      // separation. RPM is handled by the rolling lease gate below.
       // When added, treat 0/missing the same way: fall through to system default.
     }
     const limiter = new Bottleneck({
@@ -593,6 +618,15 @@ function getLimiter(provider, connectionId, model = null) {
   return limiters.get(key);
 }
 
+function evictLimiterAndDropQueued(key: string, limiter: Bottleneck, reason: string): void {
+  if (limiters.get(key) === limiter) {
+    limiters.delete(key);
+    lastDispatchAt.delete(key);
+    limiterLastUsed.delete(key);
+  }
+  trackAsyncOperation(limiter.stop({ dropWaitingJobs: true, dropErrorMessage: reason }));
+}
+
 /**
  * Acquire a rate limit slot before making a request.
  * If rate limiting is disabled for this connection, returns immediately.
@@ -604,32 +638,7 @@ function getLimiter(provider, connectionId, model = null) {
  * @param {AbortSignal} signal - Optional abort signal to cancel waiting
  * @returns {Promise<unknown>} Result of fn()
  */
-async function getQueueHealthSnapshot(key: string, limiter: Bottleneck) {
-  const counts = limiter.counts();
-  let reservoirRemaining: number | null = null;
-  try {
-    reservoirRemaining = await limiter.currentReservoir();
-  } catch {
-    // Snapshot logging must never affect request handling.
-  }
-  const lastDispatch = lastDispatchAt.get(key);
-  return {
-    queued: counts.QUEUED,
-    running: counts.RUNNING,
-    executing: counts.EXECUTING,
-    reservoirRemaining,
-    lastDispatchAgeMs: lastDispatch ? Date.now() - lastDispatch : null,
-  };
-}
-
-export async function withRateLimit(
-  provider,
-  connectionId,
-  model,
-  fn,
-  signal = null,
-  retryAfterWedge = true
-) {
+export async function withRateLimit(provider, connectionId, model, fn, signal = null) {
   if (!enabledConnections.has(connectionId)) {
     return fn();
   }
@@ -642,18 +651,20 @@ export async function withRateLimit(
     throw err;
   }
 
-  // Proactive sliding-window fallback for header-less providers with a declared cap
-  // (Fase 8.2). No-op unless PROVIDER_DEFAULT_RATE_LIMITS has an entry for `provider`.
-  await awaitProviderDefaultSlot(
+  const maxWaitMs = currentRequestQueueSettings.maxWaitMs;
+  const queueStartedAt = Date.now();
+  const rpmLease = await rpmGate.acquire(
     provider,
     connectionId,
+    model,
     signal,
-    currentRequestQueueSettings.maxWaitMs
+    maxWaitMs,
+    queueStartedAt
   );
-
   const limiter = getLimiter(provider, connectionId, model);
-  const maxWaitMs = currentRequestQueueSettings.maxWaitMs;
-  const scheduleOpts = maxWaitMs && maxWaitMs > 0 ? { expiration: maxWaitMs } : {};
+  const key = getLimiterKey(provider, connectionId, model);
+  const jobId = `${key}:job-${nextJobTraceId++}`;
+  const scheduleOpts = { id: jobId };
 
   // Issue #6593: opt-in admission cap — fast-reject before Bottleneck's
   // schedule() (and before any downstream compression/prompt work runs) when
@@ -664,30 +675,77 @@ export async function withRateLimit(
     model ? `${provider}/${model}` : provider
   );
   if (admissionErr) {
+    rpmLease?.release();
     logRateLimit(
       `🚧 [RATE-LIMIT] ${getLimiterKey(provider, connectionId, model)} — queue full, rejecting fast (maxQueueDepth=${currentRequestQueueSettings.maxQueueDepth})`
     );
     throw admissionErr;
   }
 
+  let dispatched = false;
+  let queueExpired = false;
+  let dispatchCancelled = false;
+  let queueTimer: ReturnType<typeof setTimeout> | undefined;
+  const remainingWaitMs =
+    maxWaitMs > 0 ? Math.max(1, maxWaitMs - (Date.now() - queueStartedAt)) : 0;
+  const queueTimeoutPromise =
+    remainingWaitMs > 0
+      ? new Promise<never>((_, reject) => {
+          queueTimer = setTimeout(() => {
+            if (dispatched) return;
+            queueExpired = true;
+            logRateLimit(
+              `⏰ [RATE-LIMIT] ${key} — job exceeded ${Math.ceil(maxWaitMs / 1000)}s queue wait budget, dropping`
+            );
+            reject(new Error("rate-limit-queue-timeout"));
+          }, remainingWaitMs);
+        })
+      : null;
+  const scheduled = limiter.schedule(scheduleOpts, async () => {
+    if (queueExpired) {
+      throw createQueueTimeoutError(provider, model, maxWaitMs);
+    }
+    if (dispatchCancelled) {
+      const error = new Error("The operation was aborted before limiter dispatch");
+      error.name = "AbortError";
+      throw error;
+    }
+    if (signal?.aborted) {
+      const error = new Error("The operation was aborted before limiter dispatch");
+      error.name = "AbortError";
+      throw error;
+    }
+    dispatched = true;
+    if (queueTimer) clearTimeout(queueTimer);
+    return fn();
+  });
+
   try {
     if (signal) {
       let abortListener: (() => void) | undefined;
       const abortPromise = new Promise<never>((_, reject) => {
         const onAbort = () => {
           const reason = signal.reason;
-          // Preserve native Error reasons (including AbortController's
-          // read-only DOMException) instead of mutating or wrapping them.
+          // Reject before evicting the queued job so the caller observes its
+          // abort reason instead of Bottleneck's internal drop error.
           if (reason instanceof Error) {
             reject(reason);
-            return;
+          } else {
+            const err = new Error(
+              typeof reason === "string" ? reason : "The operation was aborted"
+            );
+            err.name = "AbortError";
+            if (reason !== undefined) {
+              (err as Error & { cause?: unknown }).cause = reason;
+            }
+            reject(err);
           }
-          const err = new Error(typeof reason === "string" ? reason : "The operation was aborted");
-          err.name = "AbortError";
-          if (reason !== undefined) {
-            (err as Error & { cause?: unknown }).cause = reason;
+          if (!dispatched) {
+            dispatchCancelled = true;
+            if (queueTimer) clearTimeout(queueTimer);
+            // Leave the cancelled job in Bottleneck so queued peers are not dropped.
+            // Its scheduled callback will consume one queue turn and exit before fn().
           }
-          reject(err);
         };
         if (signal.aborted) {
           onAbort();
@@ -698,51 +756,30 @@ export async function withRateLimit(
       });
 
       try {
-        return await Promise.race([limiter.schedule(scheduleOpts, fn), abortPromise]);
+        const races: Promise<unknown>[] = [scheduled, abortPromise];
+        if (queueTimeoutPromise) races.push(queueTimeoutPromise);
+        return await Promise.race(races);
       } finally {
         if (abortListener) {
           signal.removeEventListener("abort", abortListener);
         }
       }
     } else {
-      return await limiter.schedule(scheduleOpts, fn);
+      return await (queueTimeoutPromise
+        ? Promise.race([scheduled, queueTimeoutPromise])
+        : scheduled);
     }
   } catch (err) {
-    // Bottleneck's raw `This job timed out after <maxWaitMs> ms.` is
-    // indistinguishable from an upstream gateway timeout, so it leaks into 502
-    // bodies / call-log `last_error` and gets misdiagnosed as a provider outage
-    // (#4165). Rewrite it into a clear, OmniRoute-owned error (knob named,
-    // upstream disclaimed, original kept as `cause`, `code` for classification).
-    // If the limiter is idle with capacity after the expiry, the scheduler is wedged.
-    // Reset it and retry this never-dispatched function once on a fresh limiter.
-    if (err?.message?.includes("This job timed out")) {
-      const key = getLimiterKey(provider, connectionId, model);
-      const queueState = await getQueueHealthSnapshot(key, limiter);
-      logRateLimit(
-        `⏰ [RATE-LIMIT] ${key} — job expired after ${Math.ceil((maxWaitMs || 0) / 1000)}s in queue, dropping`
-      );
-      const limiterIsWedged =
-        retryAfterWedge &&
-        queueState.running === 0 &&
-        queueState.executing === 0 &&
-        typeof queueState.reservoirRemaining === "number" &&
-        queueState.reservoirRemaining > 0 &&
-        typeof queueState.lastDispatchAgeMs === "number" &&
-        queueState.lastDispatchAgeMs >= Math.max(1, maxWaitMs || 0);
-      if (limiterIsWedged) {
-        logRateLimit(`🔄 [RATE-LIMIT] ${key} — recovering idle limiter after queue expiry`);
-        evictWedgeLimiter(key, limiter);
-        return withRateLimit(provider, connectionId, model, fn, signal, false);
-      }
-      const queueErr = new Error(
-        `Request dropped after exceeding the local rate-limit queue budget maxWaitMs (${maxWaitMs}ms) for ` +
-          `${model ? `${provider}/${model}` : provider} — this is OmniRoute's request queue ` +
-          `(resilienceSettings.requestQueue.maxWaitMs), not an upstream timeout. Raise it in ` +
-          `Settings → Resilience if this is queue saturation rather than a slow provider.`,
+    if (queueTimer) clearTimeout(queueTimer);
+    if (!dispatched) rpmLease?.release();
+    if (err?.message === "rate-limit-upstream-429") {
+      const rateLimitErr = new Error(
+        `Request dropped while the ${provider} connection was under an upstream rate-limit cooldown`,
         { cause: err }
-      ) as Error & { code?: string };
-      queueErr.code = "RATE_LIMIT_QUEUE_TIMEOUT";
-      throw queueErr;
+      ) as Error & { code?: string; status?: number };
+      rateLimitErr.code = "RATE_LIMIT_UPSTREAM_429";
+      rateLimitErr.status = 429;
+      throw rateLimitErr;
     }
     // The watchdog's stop({ dropWaitingJobs: true }) wedge-recovery (above) rejects
     // queued jobs with this exact message. Rewrite it the same way as the timeout
@@ -759,6 +796,9 @@ export async function withRateLimit(
       wedgeErr.code = "RATE_LIMIT_QUEUE_WEDGED";
       throw wedgeErr;
     }
+    if (err?.message === "rate-limit-queue-timeout") {
+      throw createQueueTimeoutError(provider, model, maxWaitMs);
+    }
     throw err;
   }
 }
@@ -802,20 +842,12 @@ export function updateFromHeaders(provider, connectionId, headers, status, model
       `🚫 [RATE-LIMIT] ${provider}:${connectionId.slice(0, 8)} — 429 received, pausing for ${Math.ceil(retryAfterMs / 1000)}s, dropping ${counts.QUEUED} queued request(s)`
     );
 
-    // Evict from the cache so follow-up learning from the same error body
-    // can materialize a fresh limiter immediately. Do NOT call limiter.stop() —
-    // it permanently rejects future .schedule() calls with "This limiter has been stopped".
-    // In-flight requests holding a reference to the evicted instance will fail (they
-    // were already going to fail — the 429 means the API rejected them), but future
-    // requests will get a fresh Bottleneck instance via getLimiter().
-    // Call disconnect() (not stop()) to release Bottleneck's internal heartbeat timer
-    // without permanently poisoning the instance for any remaining in-flight jobs.
-    // Without disconnect() here, every 429 leaks a heartbeat timer until GC reclaims
-    // the abandoned Bottleneck; under sustained quota pressure that is a real leak.
-    limiters.delete(limiterKey);
-    lastDispatchAt.delete(limiterKey);
-    limiterLastUsed.delete(limiterKey);
-    trackAsyncOperation(limiter.disconnect());
+    rpmGate.block(provider, connectionId, model, retryAfterMs);
+
+    // Evict from the cache before stopping so follow-up requests get a fresh
+    // instance. Stopping the unreachable instance rejects its queued jobs and
+    // releases its heartbeat without poisoning the replacement limiter.
+    evictLimiterAndDropQueued(limiterKey, limiter, "rate-limit-upstream-429");
     return;
   }
 
@@ -830,28 +862,31 @@ export function updateFromHeaders(provider, connectionId, headers, status, model
 
   // Normal response — update limiter from headers
   if (!isNaN(limit) && limit > 0) {
-    const resetMs = parseResetTime(resetStr) || 60000;
-
     // Calculate optimal minTime from RPM limit
     const minTime = Math.max(0, Math.floor(60000 / limit) - 10); // Small buffer
 
     const updates: LimiterUpdateSettings = { minTime };
+    const resetMs = parseResetTime(resetStr) || 60000;
 
-    // If remaining is low (< 10% of limit), set reservoir to throttle immediately
+    // Keep adaptive pacing from response headers, but do not mutate an RPM
+    // reservoir. RPM admission is enforced by the rolling lease gate.
     if (!isNaN(remaining)) {
       if (remaining < limit * 0.1) {
-        updates.reservoir = remaining;
-        updates.reservoirRefreshAmount = limit;
-        updates.reservoirRefreshInterval = resetMs;
+        rpmGate.learnHeaderWindow(
+          provider,
+          connectionId,
+          model,
+          remaining,
+          resetMs,
+          Date.now() + resetMs
+        );
         logRateLimit(
           `⚠️ [RATE-LIMIT] ${provider}:${connectionId.slice(0, 8)} — ${remaining}/${limit} remaining, throttling`
         );
       } else if (remaining > limit * 0.5) {
         // Plenty of headroom — relax the limiter
         updates.minTime = 0;
-        updates.reservoir = null;
-        updates.reservoirRefreshAmount = null;
-        updates.reservoirRefreshInterval = null;
+        rpmGate.clearLearnedHeaderWindow(provider, connectionId, model);
       }
     }
 
@@ -984,6 +1019,8 @@ export async function __resetRateLimitManagerForTests() {
   }
   limiters.clear();
   enabledConnections.clear();
+  connectionRateLimitOverrides.clear();
+  rpmGate.reset();
   initialized = false;
   lastDispatchAt.clear();
   limiterLastUsed.clear();
@@ -1088,17 +1125,10 @@ export function updateFromResponseBody(provider, connectionId, responseBody, sta
   const { retryAfterMs, reason } = parseRetryAfterFromBody(responseBody);
 
   if (retryAfterMs && retryAfterMs > 0) {
-    const limiter = getLimiter(provider, connectionId, model);
+    getLimiter(provider, connectionId, model);
     logRateLimit(
       `🚫 [RATE-LIMIT] ${provider}:${connectionId.slice(0, 8)} — body-parsed retry: ${Math.ceil(retryAfterMs / 1000)}s (${reason})`
     );
-
-    trackAsyncOperation(
-      applyLimiterSettings(limiter, {
-        reservoir: 0,
-        reservoirRefreshAmount: 60,
-        reservoirRefreshInterval: retryAfterMs,
-      })
-    );
+    rpmGate.block(provider, connectionId, model, retryAfterMs);
   }
 }
diff --git a/open-sse/services/rollingRpmGate.ts b/open-sse/services/rollingRpmGate.ts
new file mode 100644
index 0000000000..e62912ba5a
--- /dev/null
+++ b/open-sse/services/rollingRpmGate.ts
@@ -0,0 +1,236 @@
+import {
+  SlidingWindowLimiter,
+  type RateLimitScope,
+  type RateLimitWindow,
+  type SlidingWindowLease,
+} from "./slidingWindowLimiter.ts";
+
+type GetLimiterKey = (provider: string, connectionId: string, model?: string | null) => string;
+type QueueTimeoutReason = "local-queue" | "upstream-cooldown";
+type QueueTimeoutErrorFactory = (
+  provider: string,
+  model: string | null,
+  maxWaitMs: number,
+  reason?: QueueTimeoutReason
+) => Error;
+
+export interface RollingRpmGateOptions {
+  getGlobalRpm: () => number | null | undefined;
+  getProviderWindow: (provider: string) => RateLimitWindow | undefined;
+  getConnectionRpm: (connectionId: string) => number | null | undefined;
+  getLimiterKey: GetLimiterKey;
+  createQueueTimeoutError: QueueTimeoutErrorFactory;
+}
+
+interface LearnedHeaderWindow {
+  window: RateLimitWindow;
+  expiresAt: number;
+}
+
+function createAbortError(signal: AbortSignal): Error {
+  const reason = signal.reason;
+  if (reason instanceof Error) return reason;
+  const error = new Error(typeof reason === "string" ? reason : "The operation was aborted");
+  error.name = "AbortError";
+  if (reason !== undefined) (error as Error & { cause?: unknown }).cause = reason;
+  return error;
+}
+
+function sleepOrAbort(ms: number, signal: AbortSignal | null): Promise<void> {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      signal?.removeEventListener("abort", onAbort);
+      resolve();
+    }, ms);
+    const onAbort = () => {
+      clearTimeout(timer);
+      signal?.removeEventListener("abort", onAbort);
+      reject(createAbortError(signal as AbortSignal));
+    };
+    if (signal) {
+      if (signal.aborted) onAbort();
+      else signal.addEventListener("abort", onAbort, { once: true });
+    }
+  });
+}
+
+/**
+ * Process-local trailing-window RPM admission. Distributed deployments need a
+ * shared coordination store before this scope can be treated as cluster-wide.
+ */
+export function keyContainsConnection(key: string, connectionId: string): boolean {
+  const marker = `:${connectionId}`;
+  return key.endsWith(marker) || key.includes(`${marker}:`);
+}
+
+export class RollingRpmGate {
+  private readonly limiter = new SlidingWindowLimiter();
+  private readonly blockedUntil = new Map<string, number>();
+  private readonly learnedHeaderWindows = new Map<string, LearnedHeaderWindow>();
+  private readonly windowMs = 60_000;
+
+  constructor(private readonly options: RollingRpmGateOptions) {}
+
+  async acquire(
+    provider: string,
+    connectionId: string,
+    model: string | null,
+    signal: AbortSignal | null,
+    maxWaitMs: number,
+    startedAt: number
+  ): Promise<SlidingWindowLease | null> {
+    const blockKey = this.options.getLimiterKey(provider, connectionId, model);
+    let scopes = this.getScopes(provider, connectionId, model);
+    if (scopes.length === 0 && (this.blockedUntil.get(blockKey) ?? 0) <= Date.now()) {
+      return null;
+    }
+    for (;;) {
+      if (signal?.aborted) throw createAbortError(signal);
+
+      scopes = this.getScopes(provider, connectionId, model);
+      if (scopes.length === 0 && (this.blockedUntil.get(blockKey) ?? 0) <= Date.now()) {
+        return null;
+      }
+
+      const now = Date.now();
+      const blockedUntil = this.blockedUntil.get(blockKey) ?? 0;
+      if (blockedUntil <= now) this.blockedUntil.delete(blockKey);
+      let scopeBlockedUntil = 0;
+      for (const scope of scopes) {
+        const scopeBlocked = this.blockedUntil.get(scope.key) ?? 0;
+        if (scopeBlocked > now) scopeBlockedUntil = Math.max(scopeBlockedUntil, scopeBlocked);
+        else if (scopeBlocked > 0) this.blockedUntil.delete(scope.key);
+      }
+      const forcedWaitMs = Math.max(0, blockedUntil - now, scopeBlockedUntil - now);
+
+      if (forcedWaitMs === 0) {
+        const result = this.limiter.tryAcquireMany(scopes);
+        if (result.allowed) return result.lease ?? null;
+        const retryAfterMs = Math.max(1, result.retryAfterMs);
+        const remainingMs = maxWaitMs > 0 ? maxWaitMs - (now - startedAt) : retryAfterMs;
+        if (maxWaitMs > 0 && remainingMs <= 0) {
+          throw this.options.createQueueTimeoutError(provider, model, maxWaitMs);
+        }
+        await sleepOrAbort(
+          Math.min(retryAfterMs, maxWaitMs > 0 ? remainingMs : retryAfterMs),
+          signal
+        );
+        continue;
+      }
+
+      const remainingMs = maxWaitMs > 0 ? maxWaitMs - (now - startedAt) : forcedWaitMs;
+      if (maxWaitMs > 0 && remainingMs <= 0) {
+        throw this.options.createQueueTimeoutError(provider, model, maxWaitMs, "upstream-cooldown");
+      }
+      await sleepOrAbort(
+        Math.min(forcedWaitMs, maxWaitMs > 0 ? remainingMs : forcedWaitMs),
+        signal
+      );
+    }
+  }
+
+  block(provider: string, connectionId: string, model: string | null, retryAfterMs: number): void {
+    if (retryAfterMs > 0) {
+      const key = this.options.getLimiterKey(provider, connectionId, model);
+      this.blockedUntil.set(key, Date.now() + retryAfterMs);
+    }
+  }
+
+  learnHeaderWindow(
+    provider: string,
+    connectionId: string,
+    model: string | null,
+    requests: number,
+    windowMs: number,
+    expiresAt: number
+  ): void {
+    const key = `header:${this.options.getLimiterKey(provider, connectionId, model)}`;
+    if (requests <= 0) {
+      this.learnedHeaderWindows.delete(key);
+      this.blockedUntil.set(key, expiresAt);
+      return;
+    }
+    this.blockedUntil.delete(key);
+    this.learnedHeaderWindows.set(key, {
+      window: { requests, windowMs },
+      expiresAt,
+    });
+  }
+
+  clearLearnedHeaderWindow(provider: string, connectionId: string, model: string | null): void {
+    const key = `header:${this.options.getLimiterKey(provider, connectionId, model)}`;
+    this.learnedHeaderWindows.delete(key);
+    this.blockedUntil.delete(key);
+  }
+
+  clearConnection(connectionId: string): void {
+    for (const key of this.blockedUntil.keys()) {
+      if (keyContainsConnection(key, connectionId)) this.blockedUntil.delete(key);
+    }
+    for (const key of this.learnedHeaderWindows.keys()) {
+      if (keyContainsConnection(key, connectionId)) this.learnedHeaderWindows.delete(key);
+    }
+  }
+
+  reset(): void {
+    this.limiter.reset();
+    this.blockedUntil.clear();
+    this.learnedHeaderWindows.clear();
+  }
+
+  cleanupExpired(now = Date.now()): void {
+    for (const [key, expiresAt] of this.blockedUntil) {
+      if (expiresAt <= now) this.blockedUntil.delete(key);
+    }
+    for (const [key, window] of this.learnedHeaderWindows) {
+      if (window.expiresAt <= now) this.learnedHeaderWindows.delete(key);
+    }
+  }
+
+  private getScopes(
+    provider: string,
+    connectionId: string,
+    model: string | null
+  ): RateLimitScope[] {
+    const scopes: RateLimitScope[] = [];
+    const globalRpm = this.options.getGlobalRpm();
+    if (typeof globalRpm === "number" && globalRpm > 0) {
+      scopes.push({
+        key: "global",
+        window: { requests: globalRpm, windowMs: this.windowMs },
+      });
+    }
+
+    const providerWindow = this.options.getProviderWindow(provider);
+    if (providerWindow) scopes.push({ key: `provider:${provider}`, window: providerWindow });
+
+    const connectionRpm = this.options.getConnectionRpm(connectionId);
+    if (typeof connectionRpm === "number" && connectionRpm > 0) {
+      scopes.push({
+        key: `provider-account:${provider}:${connectionId}`,
+        window: { requests: connectionRpm, windowMs: this.windowMs },
+      });
+    }
+
+    const headerKey = `header:${this.options.getLimiterKey(provider, connectionId, model)}`;
+    const headerBlockedUntil = this.blockedUntil.get(headerKey) ?? 0;
+    const headerRemainingMs = headerBlockedUntil - Date.now();
+    if (headerRemainingMs > 0) {
+      scopes.push({
+        key: headerKey,
+        window: { requests: 1, windowMs: headerRemainingMs },
+      });
+      return scopes;
+    }
+    if (headerBlockedUntil > 0) this.blockedUntil.delete(headerKey);
+    const headerWindow = this.learnedHeaderWindows.get(headerKey);
+    if (headerWindow) {
+      if (headerWindow.expiresAt > Date.now()) {
+        scopes.push({ key: headerKey, window: headerWindow.window });
+      } else {
+        this.learnedHeaderWindows.delete(headerKey);
+      }
+    }
+    return scopes;
+  }
+}
diff --git a/open-sse/services/slidingWindowLimiter.ts b/open-sse/services/slidingWindowLimiter.ts
index 71be85870f..651e82c2cc 100644
--- a/open-sse/services/slidingWindowLimiter.ts
+++ b/open-sse/services/slidingWindowLimiter.ts
@@ -27,13 +27,33 @@ export interface AcquireResult {
   retryAfterMs: number;
 }
 
+export interface RateLimitScope {
+  key: string;
+  window: RateLimitWindow;
+}
+
+export interface SlidingWindowLease {
+  /** Release a lease that was acquired but never dispatched upstream. */
+  release(): void;
+}
+
+export interface MultiAcquireResult extends AcquireResult {
+  lease?: SlidingWindowLease;
+}
+
+interface Hit {
+  id: number;
+  timestamp: number;
+}
+
 // Hard ceiling on distinct keys tracked, so a pathological key space (e.g. a
 // per-request id leaking into the key) can never grow the map without bound.
 const MAX_KEYS = 5000;
 
 export class SlidingWindowLimiter {
-  private readonly hits = new Map<string, number[]>();
+  private readonly hits = new Map<string, Hit[]>();
   private readonly now: () => number;
+  private nextHitId = 1;
 
   constructor(opts: { now?: () => number } = {}) {
     this.now = opts.now ?? Date.now;
@@ -45,26 +65,60 @@ export class SlidingWindowLimiter {
    * (without recording) when the trailing window is saturated.
    */
   tryAcquire(key: string, window: RateLimitWindow): AcquireResult {
-    const { requests, windowMs } = window;
-    // A non-positive cap or window means "no limit configured" → always allow.
-    if (!(requests > 0) || !(windowMs > 0)) return { allowed: true, retryAfterMs: 0 };
+    const result = this.tryAcquireMany([{ key, window }]);
+    return { allowed: result.allowed, retryAfterMs: result.retryAfterMs };
+  }
+
+  /**
+   * Acquire all supplied scopes atomically. No scope is recorded unless every
+   * configured scope has capacity, preventing a global lease from being held
+   * while a narrower provider/account lease is unavailable.
+   */
+  tryAcquireMany(scopes: readonly RateLimitScope[]): MultiAcquireResult {
+    const activeScopes = scopes.filter(({ window }) => window.requests > 0 && window.windowMs > 0);
+    if (activeScopes.length === 0) return { allowed: true, retryAfterMs: 0 };
 
     const now = this.now();
-    const cutoff = now - windowMs;
-    const previous = this.hits.get(key);
-    // Drop timestamps that have aged out of the trailing window.
-    const live = previous ? previous.filter((ts) => ts > cutoff) : [];
+    const prepared = activeScopes.map((scope) => {
+      const cutoff = now - scope.window.windowMs;
+      const previous = this.hits.get(scope.key);
+      const live = previous ? previous.filter((hit) => hit.timestamp > cutoff) : [];
+      const retryAfterMs =
+        live.length >= scope.window.requests
+          ? Math.max(0, live[0].timestamp + scope.window.windowMs - now)
+          : 0;
+      return { scope, live, retryAfterMs };
+    });
 
-    if (live.length >= requests) {
-      // The oldest in-window hit is the first to expire and free a slot.
-      const retryAfterMs = Math.max(0, live[0] + windowMs - now);
-      this.hits.set(key, live); // persist the pruned list; do NOT record a blocked attempt
-      return { allowed: false, retryAfterMs };
-    }
+    const retryAfterMs = prepared.reduce((max, entry) => Math.max(max, entry.retryAfterMs), 0);
+    for (const entry of prepared) this.set(entry.scope.key, entry.live);
+    if (retryAfterMs > 0) return { allowed: false, retryAfterMs };
 
-    live.push(now);
-    this.set(key, live);
-    return { allowed: true, retryAfterMs: 0 };
+    const entries = prepared.map((entry) => {
+      const hit = { id: this.nextHitId++, timestamp: now };
+      entry.live.push(hit);
+      this.set(entry.scope.key, entry.live);
+      return { key: entry.scope.key, id: hit.id };
+    });
+
+    let released = false;
+    return {
+      allowed: true,
+      retryAfterMs: 0,
+      lease: {
+        release: () => {
+          if (released) return;
+          released = true;
+          for (const entry of entries) {
+            const live = this.hits.get(entry.key);
+            if (!live) continue;
+            const remaining = live.filter((hit) => hit.id !== entry.id);
+            if (remaining.length > 0) this.hits.set(entry.key, remaining);
+            else this.hits.delete(entry.key);
+          }
+        },
+      },
+    };
   }
 
   /** Clear history for one key, or all keys when called with no argument. */
@@ -73,7 +127,11 @@ export class SlidingWindowLimiter {
     else this.hits.delete(key);
   }
 
-  private set(key: string, live: number[]): void {
+  private set(key: string, live: Hit[]): void {
+    if (live.length === 0) {
+      this.hits.delete(key);
+      return;
+    }
     if (!this.hits.has(key) && this.hits.size >= MAX_KEYS) {
       // Evict the least-recently-inserted key (Map preserves insertion order).
       const oldest = this.hits.keys().next().value;
diff --git a/tests/unit/chat-rate-limit-body-lock.test.ts b/tests/unit/chat-rate-limit-body-lock.test.ts
index 3c69fa0de7..7be2cdac2b 100644
--- a/tests/unit/chat-rate-limit-body-lock.test.ts
+++ b/tests/unit/chat-rate-limit-body-lock.test.ts
@@ -32,8 +32,8 @@ test.after(async () => {
 test("handleChat applies body-derived retry-after to the runtime limiter", async () => {
   const connection = await seedConnection("openai", { apiKey: "sk-openai-body-retry" });
 
-  globalThis.fetch = async () =>
-    new Response(
+  globalThis.fetch = async () => {
+    return new Response(
       JSON.stringify({
         error: {
           message: "Rate limit exceeded. Please retry after 20s.",
@@ -44,6 +44,7 @@ test("handleChat applies body-derived retry-after to the runtime limiter", async
         headers: { "Content-Type": "application/json" },
       }
     );
+  };
 
   const response = await handleChat(
     buildRequest({
@@ -65,7 +66,7 @@ test("handleChat applies body-derived retry-after to the runtime limiter", async
     "gpt-4.1"
   );
   assert.ok(limiterState, "expected limiter state to exist for the active connection");
-  assert.equal(limiterState.reservoir, 0, "body-derived retry-after should drain the limiter");
+  assert.equal(limiterState.reservoir, null, "RPM is enforced by the rolling lease gate");
 });
 
 test("handleChat tolerates non-JSON rate-limit bodies without breaking fallback flow", async () => {
diff --git a/tests/unit/rate-limit-manager.test.ts b/tests/unit/rate-limit-manager.test.ts
index 4208259b8f..1d2ec645e1 100644
--- a/tests/unit/rate-limit-manager.test.ts
+++ b/tests/unit/rate-limit-manager.test.ts
@@ -1,5 +1,6 @@
 import test from "node:test";
 import assert from "node:assert/strict";
+import Bottleneck from "bottleneck";
 import fs from "node:fs";
 import os from "node:os";
 import path from "node:path";
@@ -12,7 +13,6 @@ const providersDb = await import("../../src/lib/db/providers.ts");
 const resilienceSettings = await import("../../src/lib/resilience/settings.ts");
 const rateLimitManager = await import("../../open-sse/services/rateLimitManager.ts");
 const accountFallback = await import("../../open-sse/services/accountFallback.ts");
-const Bottleneck = (await import("bottleneck")).default;
 
 function wait(ms) {
   return new Promise((resolve) => setTimeout(resolve, ms));
@@ -60,43 +60,296 @@ test("rate limit manager bypasses disabled connections and exposes inactive stat
   assert.deepEqual(rateLimitManager.getAllRateLimitStatus(), {});
 });
 
-test("idle-capacity queue expiry resets the limiter and retries once", async () => {
+test("queue expiry does not invoke the provider after a late dispatch", async () => {
   await rateLimitManager.applyRequestQueueSettings({
     ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
     autoEnableApiKeyProviders: false,
-    maxWaitMs: 20,
+    maxWaitMs: 100,
     requestsPerMinute: 0,
     concurrentRequests: 1,
     minTimeBetweenRequestsMs: 0,
     maxQueueDepth: 0,
   });
 
-  const originalSchedule = Bottleneck.prototype.schedule;
-  let attempts = 0;
-  Bottleneck.prototype.schedule = function (...args) {
-    attempts++;
-    if (attempts === 1) {
-      return new Promise((_, reject) => {
-        setTimeout(() => reject(new Error("This job timed out after 20 ms.")), 30);
-      });
+  rateLimitManager.enableRateLimitProtection("queue-expiry-conn");
+  let resolveFirstStarted: () => void = () => undefined;
+  const firstStarted = new Promise<void>((resolve) => {
+    resolveFirstStarted = resolve;
+  });
+  const first = rateLimitManager.withRateLimit(
+    "openai",
+    "queue-expiry-conn",
+    "gpt-4o",
+    async () => {
+      resolveFirstStarted();
+      await wait(300);
+      return "first";
     }
-    return originalSchedule.apply(this, args);
-  };
+  );
+  await firstStarted;
 
-  try {
-    rateLimitManager.enableRateLimitProtection("idle-capacity-conn");
-    const result = await rateLimitManager.withRateLimit(
-      "openai",
-      "idle-capacity-conn",
-      "gpt-4o",
-      async () => "recovered"
-    );
+  let secondCalls = 0;
+  await assert.rejects(
+    rateLimitManager.withRateLimit("openai", "queue-expiry-conn", "gpt-4o", async () => {
+      secondCalls++;
+      return "late";
+    }),
+    (error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
+  );
 
-    assert.equal(result, "recovered");
-    assert.equal(attempts, 2, "the expired job should be retried once on a fresh limiter");
-  } finally {
-    Bottleneck.prototype.schedule = originalSchedule;
+  await first;
+  await wait(50);
+  assert.equal(secondCalls, 0, "a queue-expired job must not invoke the provider later");
+});
+
+test("queue expiry does not drop other queued jobs", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+    autoEnableApiKeyProviders: false,
+    maxWaitMs: 500,
+    requestsPerMinute: 0,
+    concurrentRequests: 1,
+    minTimeBetweenRequestsMs: 0,
+    maxQueueDepth: 0,
+  });
+
+  rateLimitManager.enableRateLimitProtection("queue-peer-conn");
+  let resolveFirstExecuting: () => void = () => undefined;
+  const firstExecuting = new Promise<void>((resolve) => {
+    resolveFirstExecuting = resolve;
+  });
+  let releaseFirst: () => void = () => undefined;
+  const first = rateLimitManager.withRateLimit("openai", "queue-peer-conn", null, async () => {
+    resolveFirstExecuting();
+    await new Promise<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+    return "first";
+  });
+  await firstExecuting;
+
+  const second = rateLimitManager.withRateLimit(
+    "openai",
+    "queue-peer-conn",
+    null,
+    async () => "expired"
+  );
+  await wait(400);
+
+  let thirdCalls = 0;
+  const third = rateLimitManager.withRateLimit("openai", "queue-peer-conn", null, async () => {
+    thirdCalls++;
+    return "third";
+  });
+  await assert.rejects(
+    second,
+    (error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
+  );
+  releaseFirst();
+  await Promise.all([first, third]);
+  assert.equal(thirdCalls, 1, "a peer queued job must survive another job's expiry");
+});
+
+test("global RPM lease is shared across enabled provider connections", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+    autoEnableApiKeyProviders: false,
+    maxWaitMs: 1000,
+    requestsPerMinute: 2,
+    concurrentRequests: 10,
+    minTimeBetweenRequestsMs: 0,
+    maxQueueDepth: 0,
+  });
+
+  rateLimitManager.enableRateLimitProtection("global-rpm-a");
+  rateLimitManager.enableRateLimitProtection("global-rpm-b");
+  let calls = 0;
+  await rateLimitManager.withRateLimit("openai", "global-rpm-a", null, async () => {
+    calls++;
+  });
+  await rateLimitManager.withRateLimit("anthropic", "global-rpm-b", null, async () => {
+    calls++;
+  });
+
+  await assert.rejects(
+    rateLimitManager.withRateLimit("openai", "global-rpm-a", null, async () => {
+      calls++;
+    }),
+    (error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
+  );
+  assert.equal(calls, 2, "the global lease blocks the third dispatch across providers");
+});
+
+test("provider/account RPM lease failure does not consume the global lease", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+    autoEnableApiKeyProviders: false,
+    maxWaitMs: 1000,
+    requestsPerMinute: 2,
+    concurrentRequests: 10,
+    minTimeBetweenRequestsMs: 0,
+    maxQueueDepth: 0,
+  });
+
+  rateLimitManager.enableRateLimitProtection("provider-rpm-a");
+  rateLimitManager.enableRateLimitProtection("provider-rpm-b");
+  rateLimitManager.refreshConnectionRateLimits("provider-rpm-a", { rpm: 1 });
+
+  let calls = 0;
+  await rateLimitManager.withRateLimit("openai", "provider-rpm-a", null, async () => {
+    calls++;
+  });
+
+  await assert.rejects(
+    rateLimitManager.withRateLimit("openai", "provider-rpm-a", null, async () => {
+      calls++;
+    }),
+    (error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
+  );
+
+  await rateLimitManager.withRateLimit("anthropic", "provider-rpm-b", null, async () => {
+    calls++;
+  });
+  assert.equal(calls, 2, "the failed provider lease did not consume the second global lease");
+});
+
+test("aborted queued work releases its pre-dispatch RPM lease", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+    autoEnableApiKeyProviders: false,
+    maxWaitMs: 1000,
+    requestsPerMinute: 2,
+    concurrentRequests: 1,
+    minTimeBetweenRequestsMs: 0,
+    maxQueueDepth: 0,
+  });
+
+  rateLimitManager.enableRateLimitProtection("abort-lease-conn");
+  rateLimitManager.enableRateLimitProtection("abort-lease-other");
+  let resolveFirstExecuting: () => void = () => undefined;
+  const firstExecuting = new Promise<void>((resolve) => {
+    resolveFirstExecuting = resolve;
+  });
+  let releaseFirst: () => void = () => undefined;
+  const firstStarted = new Promise<void>((resolve) => {
+    releaseFirst = resolve;
+  });
+  const first = rateLimitManager.withRateLimit("openai", "abort-lease-conn", null, async () => {
+    resolveFirstExecuting();
+    await firstStarted;
+    return "first";
+  });
+  await firstExecuting;
+
+  const controller = new AbortController();
+  let abortedCalls = 0;
+  const queued = rateLimitManager.withRateLimit(
+    "openai",
+    "abort-lease-conn",
+    null,
+    async () => {
+      abortedCalls++;
+      return "should-not-dispatch";
+    },
+    controller.signal
+  );
+  await wait(20);
+  controller.abort();
+  await assert.rejects(queued, (error: { name?: string }) => error.name === "AbortError");
+
+  let thirdCalls = 0;
+  await rateLimitManager.withRateLimit("anthropic", "abort-lease-other", null, async () => {
+    thirdCalls++;
+  });
+  releaseFirst();
+  await first;
+  assert.equal(abortedCalls, 0, "aborted queued work must not invoke the provider");
+  assert.equal(thirdCalls, 1, "aborted work must return its unused global lease");
+});
+
+test("aborting one queued request does not drop queued peers", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+    autoEnableApiKeyProviders: false,
+    maxWaitMs: 1000,
+    requestsPerMinute: 0,
+    concurrentRequests: 1,
+    minTimeBetweenRequestsMs: 0,
+    maxQueueDepth: 0,
+  });
+
+  const connectionId = "abort-peer-conn";
+  rateLimitManager.enableRateLimitProtection(connectionId);
+  let resolveFirstExecuting: () => void = () => undefined;
+  const firstExecuting = new Promise<void>((resolve) => {
+    resolveFirstExecuting = resolve;
+  });
+  let releaseFirst: () => void = () => undefined;
+  const first = rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => {
+    resolveFirstExecuting();
+    await new Promise<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+  });
+
+  await firstExecuting;
+  const limiter = rateLimitManager.__getLimiterForTests("test-provider", connectionId);
+  const controller = new AbortController();
+  let abortedCalls = 0;
+  const aborted = rateLimitManager.withRateLimit(
+    "test-provider",
+    connectionId,
+    null,
+    async () => {
+      abortedCalls++;
+    },
+    controller.signal
+  );
+  let peerCalls = 0;
+  const peer = rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => {
+    peerCalls++;
+  });
+  for (let attempt = 0; attempt < 200 && limiter.counts().QUEUED < 2; attempt++) {
+    await wait(5);
   }
+  assert.ok(limiter.counts().QUEUED >= 2, "both queued requests must be present before abort");
+  controller.abort();
+  await assert.rejects(aborted, (error: { name?: string }) => error.name === "AbortError");
+
+  releaseFirst();
+  await Promise.all([first, peer]);
+  assert.equal(abortedCalls, 0, "aborted queued work must not invoke the provider");
+  assert.equal(peerCalls, 1);
+});
+
+test("dispatched provider failures retain their RPM lease", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+    autoEnableApiKeyProviders: false,
+    maxWaitMs: 500,
+    requestsPerMinute: 1,
+    concurrentRequests: 10,
+    minTimeBetweenRequestsMs: 0,
+    maxQueueDepth: 0,
+  });
+
+  rateLimitManager.enableRateLimitProtection("failed-dispatch-a");
+  rateLimitManager.enableRateLimitProtection("failed-dispatch-b");
+  await assert.rejects(
+    rateLimitManager.withRateLimit("openai", "failed-dispatch-a", null, async () => {
+      throw new Error("upstream failure");
+    }),
+    /upstream failure/
+  );
+
+  let secondCalls = 0;
+  await assert.rejects(
+    rateLimitManager.withRateLimit("anthropic", "failed-dispatch-b", null, async () => {
+      secondCalls++;
+    }),
+    (error: { code?: string }) => error.code === "RATE_LIMIT_QUEUE_TIMEOUT"
+  );
+  assert.equal(secondCalls, 0, "a dispatched failure still counts against the RPM window");
 });
 
 test("withRateLimit forwards AbortController DOMException without mutating it", async () => {
@@ -209,6 +462,130 @@ test("rate limit manager handles 429 limiter teardown and disable cleanup", asyn
   assert.equal(rateLimitManager.getRateLimitStatus("gemini", "conn-disable").active, false);
 });
 
+test("rate limit manager blocks admission after an upstream 429 retry hint", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    concurrentRequests: 1,
+    requestsPerMinute: 0,
+    maxWaitMs: 100,
+  });
+  rateLimitManager.enableRateLimitProtection("conn-429-block");
+  rateLimitManager.updateFromHeaders(
+    "openai",
+    "conn-429-block",
+    { "retry-after": "1s" },
+    429,
+    "gpt-4o"
+  );
+
+  let providerCalls = 0;
+  await assert.rejects(
+    rateLimitManager.withRateLimit("openai", "conn-429-block", "gpt-4o", async () => {
+      providerCalls++;
+    }),
+    (error: unknown) => {
+      const code = error && typeof error === "object" && "code" in error ? error.code : undefined;
+      assert.equal(code, "RATE_LIMIT_QUEUE_TIMEOUT");
+      assert.match(String((error as Error).message), /upstream rate-limit cooldown/);
+      return true;
+    }
+  );
+  assert.equal(providerCalls, 0);
+});
+
+test("rate limit manager blocks a zero-remaining header window until reset", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    concurrentRequests: 1,
+    requestsPerMinute: 0,
+    maxWaitMs: 100,
+  });
+  rateLimitManager.enableRateLimitProtection("conn-zero-remaining");
+  rateLimitManager.updateFromHeaders(
+    "openai",
+    "conn-zero-remaining",
+    {
+      "x-ratelimit-limit-requests": "10",
+      "x-ratelimit-remaining-requests": "0",
+      "x-ratelimit-reset-requests": "1s",
+    },
+    200
+  );
+
+  let providerCalls = 0;
+  await assert.rejects(
+    rateLimitManager.withRateLimit("openai", "conn-zero-remaining", null, async () => {
+      providerCalls++;
+    }),
+    (error: unknown) => {
+      const code = error && typeof error === "object" && "code" in error ? error.code : undefined;
+      assert.equal(code, "RATE_LIMIT_QUEUE_TIMEOUT");
+      assert.match(String((error as Error).message), /upstream rate-limit cooldown/);
+      return true;
+    }
+  );
+  assert.equal(providerCalls, 0);
+});
+
+test("rate limit manager keeps learned header windows model-scoped where limiters are model-scoped", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    concurrentRequests: 1,
+    requestsPerMinute: 0,
+    maxWaitMs: 100,
+  });
+  rateLimitManager.enableRateLimitProtection("conn-model-header");
+  rateLimitManager.updateFromHeaders(
+    "github",
+    "conn-model-header",
+    {
+      "x-ratelimit-limit-requests": "10",
+      "x-ratelimit-remaining-requests": "0",
+      "x-ratelimit-reset-requests": "1s",
+    },
+    200,
+    "model-a"
+  );
+
+  let providerCalls = 0;
+  await rateLimitManager.withRateLimit("github", "conn-model-header", "model-b", async () => {
+    providerCalls++;
+  });
+  assert.equal(providerCalls, 1);
+});
+
+test("rate limit watchdog resets a queued limiter with received work", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    concurrentRequests: 1,
+    requestsPerMinute: 0,
+    maxWaitMs: 5_000,
+  });
+  rateLimitManager.enableRateLimitProtection("conn-wedge");
+  const limiter = new Bottleneck({ reservoir: 0, id: "test-provider:conn-wedge" });
+  rateLimitManager.__installLimiterForTests("test-provider", "conn-wedge", limiter);
+
+  let providerCalls = 0;
+  const pending = rateLimitManager.withRateLimit("test-provider", "conn-wedge", null, async () => {
+    providerCalls++;
+  });
+  await wait(100);
+  const counts = limiter.counts();
+  assert.equal(counts.RECEIVED, 0);
+  assert.ok(counts.QUEUED > 0);
+
+  rateLimitManager.__setLastDispatchAtForTests(
+    "test-provider",
+    "conn-wedge",
+    null,
+    Date.now() - 120_001
+  );
+  rateLimitManager.__runRateLimitWatchdogForTests();
+
+  await assert.rejects(pending, (error: unknown) => {
+    const code = error && typeof error === "object" && "code" in error ? error.code : undefined;
+    assert.equal(code, "RATE_LIMIT_QUEUE_WEDGED");
+    return true;
+  });
+  assert.equal(providerCalls, 0);
+});
+
 test("rate limit manager uses model-scoped limiter keys for GitHub Copilot (#1624)", async () => {
   rateLimitManager.enableRateLimitProtection("conn-github");
   rateLimitManager.updateFromHeaders(
@@ -240,6 +617,11 @@ test("rate limit manager uses model-scoped limiter keys for GitHub Copilot (#162
 });
 
 test("rate limit manager parses retry hints from response bodies and locks models", async () => {
+  await rateLimitManager.applyRequestQueueSettings({
+    ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+    maxWaitMs: 100,
+    requestsPerMinute: 0,
+  });
   rateLimitManager.enableRateLimitProtection("conn-body");
   rateLimitManager.updateFromResponseBody(
     "openai",
@@ -263,6 +645,20 @@ test("rate limit manager parses retry hints from response bodies and locks model
   assert.equal(limiterState?.key, "openai:conn-body");
   assert.equal(rateLimitManager.getRateLimitStatus("openai", "conn-body").active, true);
 
+  let providerCalls = 0;
+  await assert.rejects(
+    rateLimitManager.withRateLimit("openai", "conn-body", "gpt-4o", async () => {
+      providerCalls++;
+    }),
+    (error: unknown) => {
+      const code = error && typeof error === "object" && "code" in error ? error.code : undefined;
+      assert.equal(code, "RATE_LIMIT_QUEUE_TIMEOUT");
+      assert.match(String((error as Error).message), /upstream rate-limit cooldown/);
+      return true;
+    }
+  );
+  assert.equal(providerCalls, 0);
+
   rateLimitManager.updateFromResponseBody(
     "openai",
     "conn-body",
diff --git a/tests/unit/rate-limit-queue-timeout-message-4165.test.ts b/tests/unit/rate-limit-queue-timeout-message-4165.test.ts
index 37b8d13ad5..e514bb9428 100644
--- a/tests/unit/rate-limit-queue-timeout-message-4165.test.ts
+++ b/tests/unit/rate-limit-queue-timeout-message-4165.test.ts
@@ -1,19 +1,13 @@
 /**
  * #4165 — surface a clear error when the request-queue (Bottleneck) drops a job.
  *
- * OmniRoute schedules every rate-limited request through Bottleneck with
- * `{ expiration: requestQueue.maxWaitMs }` (open-sse/services/rateLimitManager.ts).
- * When a job exceeds that budget Bottleneck throws the raw message
- * `"This job timed out after <N> ms."` — which is indistinguishable from an
- * upstream gateway timeout. In #4165 an operator spent ~3h misdiagnosing local
- * queue saturation as a provider outage because the 502 body / call-log
- * `last_error` carried that upstream-looking string across many providers.
+ * Queue waiting is bounded by a separate timer. Bottleneck's job expiration is
+ * intentionally not used because it measures the entire scheduled lifetime and
+ * would kill an already-dispatched provider call that is making progress.
  *
- * The fix rewrites that specific Bottleneck error into a clear, OmniRoute-owned
- * message that names the knob (`resilienceSettings.requestQueue.maxWaitMs`) and
- * explicitly says it is NOT an upstream timeout, while preserving the original
- * error as `.cause` and tagging `.code = "RATE_LIMIT_QUEUE_TIMEOUT"` so callers
- * can classify it. Behavior is unchanged: the job is still dropped.
+ * The queue-only timer still rewrites pre-dispatch expiry into a clear,
+ * OmniRoute-owned error that names the knob (`resilienceSettings.requestQueue.maxWaitMs`)
+ * and explicitly says it is NOT an upstream timeout.
  */
 import test from "node:test";
 import assert from "node:assert/strict";
@@ -41,8 +35,7 @@ test.after(() => {
   fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
 });
 
-// Drive a real Bottleneck `expiration` failure: a tiny maxWaitMs and a job that
-// runs longer than it.
+// A dispatched provider call may run longer than maxWaitMs without being killed.
 async function triggerQueueTimeout() {
   await rateLimitManager.applyRequestQueueSettings({
     ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
@@ -60,36 +53,60 @@ async function triggerQueueTimeout() {
   });
 }
 
-test("#4165 queue-timeout surfaces a clear OmniRoute error, not the raw upstream-looking string", async () => {
-  let caught: (Error & { code?: string; cause?: { message?: string } }) | undefined;
+async function triggerQueuedTimeout() {
+  await rateLimitManager.applyRequestQueueSettings({
+    ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+    autoEnableApiKeyProviders: false,
+    concurrentRequests: 1,
+    requestsPerMinute: 0,
+    minTimeBetweenRequestsMs: 0,
+    maxWaitMs: 40,
+  });
+  const connectionId = "conn-queued-timeout";
+  rateLimitManager.enableRateLimitProtection(connectionId);
+
+  let resolveFirstExecuting: () => void = () => undefined;
+  const firstExecuting = new Promise<void>((resolve) => {
+    resolveFirstExecuting = resolve;
+  });
+  let releaseFirst: () => void = () => undefined;
+  const first = rateLimitManager.withRateLimit("test-provider", connectionId, null, async () => {
+    resolveFirstExecuting();
+    await new Promise<void>((resolve) => {
+      releaseFirst = resolve;
+    });
+  });
+  await firstExecuting;
+
+  let caught: unknown;
   try {
-    await triggerQueueTimeout();
-    assert.fail("expected the queued job to be dropped");
-  } catch (err) {
-    caught = err as Error & { code?: string; cause?: { message?: string } };
+    await rateLimitManager.withRateLimit(
+      "test-provider",
+      connectionId,
+      null,
+      async () => "should-not-dispatch"
+    );
+    assert.fail("expected the queued job to expire");
+  } catch (error) {
+    caught = error;
+  } finally {
+    releaseFirst();
+    await first;
   }
-  assert.ok(caught, "an error should have been thrown");
+  return caught;
+}
 
-  // Tagged so combo / callers can classify it as a local queue drop.
-  assert.equal(caught.code, "RATE_LIMIT_QUEUE_TIMEOUT", "error must carry the queue-timeout code");
+test("#4165 a dispatched provider call is not killed by the queue budget", async () => {
+  const result = await triggerQueueTimeout();
+  assert.equal(result, "should-not-reach");
+});
 
-  // The surfaced message must read as a local queue limit, naming the knob,
-  // and must NOT masquerade as an upstream "This job timed out" gateway error.
-  assert.match(caught.message, /maxWaitMs/, "message should name the maxWaitMs knob");
-  assert.match(
-    caught.message,
-    /not an upstream/i,
-    "message should explicitly disclaim an upstream timeout"
-  );
-  assert.doesNotMatch(
-    caught.message,
-    /This job timed out/,
-    "raw Bottleneck/upstream-looking string must not leak into the surfaced message"
-  );
-
-  // The original Bottleneck error is preserved for debugging.
-  assert.ok(caught.cause, "original error should be preserved as cause");
-  assert.match(String(caught.cause?.message ?? ""), /This job timed out/);
+test("#4165 queue expiry surfaces a clear local error", async () => {
+  const caught = (await triggerQueuedTimeout()) as Error & { code?: string };
+  assert.equal(caught.code, "RATE_LIMIT_QUEUE_TIMEOUT");
+  assert.match(caught.message, /maxWaitMs/);
+  assert.match(caught.message, /not an upstream/i);
+  assert.doesNotMatch(caught.message, /This job timed out/);
 });
 
 test("#4165 a job that completes within maxWaitMs is unaffected", async () => {
diff --git a/tests/unit/sliding-window-limiter.test.ts b/tests/unit/sliding-window-limiter.test.ts
index 4b2a40fe07..24b6791a40 100644
--- a/tests/unit/sliding-window-limiter.test.ts
+++ b/tests/unit/sliding-window-limiter.test.ts
@@ -25,7 +25,10 @@ test("allows up to N requests in the window and blocks the (N+1)-th", () => {
 
   const blocked = limiter.tryAcquire("k", win);
   assert.equal(blocked.allowed, false, "the 4th request in a 3/1000ms window is blocked");
-  assert.ok(blocked.retryAfterMs > 0 && blocked.retryAfterMs <= 1000, "retryAfterMs points at the oldest hit expiry");
+  assert.ok(
+    blocked.retryAfterMs > 0 && blocked.retryAfterMs <= 1000,
+    "retryAfterMs points at the oldest hit expiry"
+  );
 });
 
 test("the window slides: a slot frees once the oldest hit ages out", () => {
@@ -39,7 +42,11 @@ test("the window slides: a slot frees once the oldest hit ages out", () => {
   assert.equal(limiter.tryAcquire("k", win).allowed, false, "2/1000ms is saturated at t=400");
 
   clock.advance(601); // t=1001 — the t=0 hit (>1000ms old) ages out
-  assert.equal(limiter.tryAcquire("k", win).allowed, true, "a slot frees once the oldest hit leaves the window");
+  assert.equal(
+    limiter.tryAcquire("k", win).allowed,
+    true,
+    "a slot frees once the oldest hit leaves the window"
+  );
 });
 
 test("retryAfterMs reflects when the oldest in-window hit expires", () => {
@@ -60,7 +67,11 @@ test("keys are isolated from one another", () => {
   const win = { requests: 1, windowMs: 1000 };
 
   assert.equal(limiter.tryAcquire("a", win).allowed, true);
-  assert.equal(limiter.tryAcquire("b", win).allowed, true, "key b is unaffected by key a being saturated");
+  assert.equal(
+    limiter.tryAcquire("b", win).allowed,
+    true,
+    "key b is unaffected by key a being saturated"
+  );
   assert.equal(limiter.tryAcquire("a", win).allowed, false);
 });
 
@@ -78,7 +89,11 @@ test("reset clears a single key's history", () => {
   assert.equal(limiter.tryAcquire("k", win).allowed, true);
   assert.equal(limiter.tryAcquire("k", win).allowed, false);
   limiter.reset("k");
-  assert.equal(limiter.tryAcquire("k", win).allowed, true, "history cleared → slot available again");
+  assert.equal(
+    limiter.tryAcquire("k", win).allowed,
+    true,
+    "history cleared → slot available again"
+  );
 });
 
 test("blocked attempts do not consume a slot (no double counting)", () => {
@@ -92,5 +107,53 @@ test("blocked attempts do not consume a slot (no double counting)", () => {
   limiter.tryAcquire("k", win);
   limiter.tryAcquire("k", win);
   clock.advance(1001);
-  assert.equal(limiter.tryAcquire("k", win).allowed, true, "only the single successful hit aged out");
+  assert.equal(
+    limiter.tryAcquire("k", win).allowed,
+    true,
+    "only the single successful hit aged out"
+  );
+});
+
+test("multi-scope acquisition is atomic", () => {
+  const clock = fakeClock();
+  const limiter = new SlidingWindowLimiter({ now: clock.now });
+  const global = { key: "global", window: { requests: 2, windowMs: 1000 } };
+  const provider = { key: "provider:openference", window: { requests: 1, windowMs: 1000 } };
+
+  const first = limiter.tryAcquireMany([global, provider]);
+  assert.equal(first.allowed, true);
+
+  const blocked = limiter.tryAcquireMany([global, provider]);
+  assert.equal(blocked.allowed, false, "the provider scope blocks the second request");
+
+  clock.advance(1001);
+  const second = limiter.tryAcquireMany([global, provider]);
+  assert.equal(second.allowed, true);
+});
+
+test("releasing an un-dispatched multi-scope lease returns every scope", () => {
+  const limiter = new SlidingWindowLimiter();
+  const scopes = [
+    { key: "global", window: { requests: 1, windowMs: 1000 } },
+    { key: "provider", window: { requests: 1, windowMs: 1000 } },
+  ];
+
+  const result = limiter.tryAcquireMany(scopes);
+  assert.equal(result.allowed, true);
+  result.lease?.release();
+  assert.equal(limiter.tryAcquireMany(scopes).allowed, true);
+});
+
+test("rolling leases do not reset as a burst at a fixed boundary", () => {
+  const clock = fakeClock();
+  const limiter = new SlidingWindowLimiter({ now: clock.now });
+  const window = { key: "global", window: { requests: 2, windowMs: 1000 } };
+
+  assert.equal(limiter.tryAcquireMany([window]).allowed, true); // t=0
+  clock.advance(900);
+  assert.equal(limiter.tryAcquireMany([window]).allowed, true); // t=900
+  assert.equal(limiter.tryAcquireMany([window]).allowed, false);
+  clock.advance(100);
+  assert.equal(limiter.tryAcquireMany([window]).allowed, true, "only the t=0 lease returned");
+  assert.equal(limiter.tryAcquireMany([window]).allowed, false, "the t=900 lease remains active");
 });