fix(resilience): stop unbounded queue that hangs 6min until Aborted (#12715)

Fila sem teto que segura a request seis minutos até o cliente abortar é pior que 503 imediato: consome slot, mascara a saturação e ainda entrega erro no fim. Um orçamento `maxWaitMs` por conexão compartilhado entre gate, slot padrão do provider e fila do Bottleneck é a forma certa — o teto tem que ser um só, senão cada camada espera o seu.

O `max(perConn, upstream)` no `executionMaxWaitMs` é o detalhe que evita a correção matar request em voo, que seria trocar um defeito por outro.

Registro a atribuição: você manteve o #12635 aberto para o @Tushar49 e creditou a percepção dele (providers lentos precisam de 2min→10min por conexão) enquanto adiciona o encanamento que faltava. É o jeito certo de construir sobre PR de outra pessoa sem tomar o crédito.

Sobre o `npm run lint` desmarcado com a nota do eslint quebrado no ambiente: deixar em branco e explicar vale mais que marcar sem ter rodado. Rodei aqui: limpo.

Revalidei sobre o tip: **13/13**, typecheck:core limpo, check-file-size OK. O `file-size-baseline.json` conflitou com os rebaselines desta campanha — resolvido aditivamente, JSON revalidado com `json.load`.

changelog.d/fixes/12715-cascade-pr1-queue-bound.md

@@ -0,0 +1 @@
+- **fix(resilience):** stop unbounded queue that hangs 6min until Aborted — gate, provider slot, and Bottleneck queue now share a per-connection `maxWaitMs` budget; `fail-closed` on exhaust (503); execution backstop `executionMaxWaitMs` overridable per connection with upstream clamp ([#12715](https://github.com/diegosouzapw/OmniRoute/pull/12715)) — thanks @maxmad64bis (with thanks to @Tushar49 for surfacing the slow-provider need in #12635)

config/quality/file-size-baseline.json

@@ -1,5 +1,6 @@
 {
   "_rebaseline_2026_09_10_12828_translate_usage_chunk": "PR #12828 own growth: open-sse/utils/stream.ts 3072->3080 (+8). Translate-mode streams now send the estimated usage as the canonical trailing usage-only chunk before [DONE] when the upstream stays silent (parity with the #12151 passthrough flush), with a latch so a finish chunk that already carried the estimate is not doubled. The chunk builder is shared with the passthrough flush in open-sse/utils/usageOnlyChunk.ts (under cap); what remains is the flush-site wiring. Covered by tests/unit/stream-translate-usage-trailing.test.ts.",
+  "_rebaseline_2026_09_10_12715_queue_budget": "PR #12715 own growth: open-sse/handlers/chatCore.ts 6021->6036 (+15). Hierarchical admission now resolves the per-connection queue budget before the gates and hands withRateLimit the remaining budget, the correlation id and the executor timeout context, so gate wait, provider slot and Bottleneck queue share one bound instead of stacking. Error shaping lives in open-sse/handlers/chatCore/queueBudget.ts (under cap); what remains is irreducible call-site wiring. Covered by tests/unit/rate-limit-remaining-budget.test.ts, rate-limit-manager-queue-bound.test.ts and chatcore-hierarchical-admission.test.ts.",
   "_rebaseline_2026_09_06_runtime_quotagroup_nodemap": "Own growth: src/app/(dashboard)/dashboard/runtime/RuntimePageClient.tsx 1201->1222 (+21, check-file-size split-newline). QuotaGroup is a module-level sibling and was reading nodeMap from RuntimePageClient's closure; that identifier is not in scope, so a quota monitor with status error/exhausted/alerting throws ReferenceError. Fix threads nodeMap as a prop (3 call sites + parameter + ProviderNodeEntry import). Prettier wraps the long import and the three QuotaGroup JSX tags. Covered by tests/unit/ui/runtime-page-client.test.tsx (empty monitors stay green; error+exhausted fixtures mount QuotaGroup).",
   "_rebaseline_2026_09_05_claude_extra_usage_preflight": "Own growth: open-sse/services/combo.ts 4080->4084 (+4). buildAutoCandidates now forwards connection.providerSpecificData into evaluateQuotaCutoff so a Claude account with blockExtraUsage=false is not dropped at the 5h bar. Irreducible at the existing cutoff call site; the helper lives in claudeExtraUsage.ts (under cap). Covered by tests/unit/quota-preflight.test.ts.",
   "_rebaseline_2026_09_04_12697_combo_pin_allowlist": "PR #12697 own growth: src/sse/handlers/chat.ts 2454->2458 (+4). checkModelAvailable preflight and handleSingleModelChat now call comboPinAllowlist so a pin-only combo step cannot scan the provider pool after 502/429. Helper lives in src/lib/combos/steps.ts under cap. Covered by tests/unit/combo-pin-implicit-allowlist.test.ts (11/11).",
@@ -426,6 +427,7 @@
     "open-sse/executors/cursor.ts": 1759,
     "open-sse/executors/muse-spark-web.ts": 1405,
     "open-sse/handlers/chatCore.ts": 6026,
+    "open-sse/handlers/chatCore.ts": 6036,
     "open-sse/handlers/imageGeneration.ts": 3259,
     "open-sse/handlers/search.ts": 1789,
     "open-sse/mcp-server/schemas/tools.ts": 1621,

open-sse/handlers/chatCore.ts

@@ -354,8 +354,10 @@ import {
   updateFromHeaders,
   updateFromResponseBody,
   initializeRateLimits,
+  resolveRequestQueueMaxWaitMs,
 } from "../services/rateLimitManager.ts";
 import * as localLimiterErrors from "../services/rateLimitManager/errors.ts";
+import { rethrowAdmissionError, remainingQueueBudgetMs } from "./chatCore/queueBudget.ts";
 import {
   acquireMany as acquireConcurrencyGates,
   markBlocked as markAccountSemaphoreBlocked,
@@ -3092,6 +3094,12 @@ export async function handleChatCore({
                 stage: "waiting_account_slot",
               });
             }
+            const maxWaitMs = resolveRequestQueueMaxWaitMs(
+              provider,
+              undefined,
+              attemptConnectionId ?? undefined
+            );
+            const gateStartedAt = Date.now();
             const releaseAccountSemaphore = await acquireConcurrencyGates(
               [
                 {
@@ -3108,12 +3116,13 @@ export async function handleChatCore({
                 },
               ],
               {
-                timeoutMs: resilienceSettings.requestQueue.maxWaitMs,
+                timeoutMs: maxWaitMs,
                 maxQueueSize: resilienceSettings.requestQueue.maxQueueDepth,
                 signal: streamController.signal,
               }
-            );
-            trace("post_semaphore");
+            ).catch(rethrowAdmissionError);
+            const remainingAfterGate = remainingQueueBudgetMs(maxWaitMs, gateStartedAt);
+            trace("post_semaphore", { maxWaitMs, remainingAfterGate });
             updatePendingScope(pendingScope, {
               stage: "waiting_rate_limit",
             });
@@ -3162,7 +3171,13 @@ export async function handleChatCore({
                       ),
                   });
                 },
-                streamController.signal
+                streamController.signal,
+                remainingAfterGate,
+                correlationId ?? undefined,
+                {
+                  executor: executor as unknown as { getTimeoutMs?: () => unknown },
+                  providerSpecificData: execCreds?.providerSpecificData,
+                }
               );
               const res = normalizeExecutorResult(rawExecutorResult);
               trace("post_executor", { status: res?.response?.status });

open-sse/handlers/chatCore/queueBudget.ts

@@ -0,0 +1,25 @@
+import { isLocalStreamLifecycleError } from "@/shared/utils/circuitBreaker.ts";
+import {
+  markLocalRateLimitError,
+  LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE,
+} from "../../services/rateLimitManager/errors.ts";
+
+/**
+ * Rethrow a hierarchical-admission failure. Aborts, stream-lifecycle errors and coded
+ * semaphore errors pass through unchanged (SEMAPHORE_QUEUE_FULL is a 429 admission signal
+ * the combo cascade reads); only a bare time-budget timeout becomes the LEGACY queue 503.
+ */
+export function rethrowAdmissionError(error: unknown): never {
+  const err = error as { name?: unknown; code?: unknown } | null;
+  if (err?.name === "AbortError" || err?.code === "ABORT_ERR") throw error;
+  if (isLocalStreamLifecycleError(error)) throw error;
+  if (err?.code === undefined) {
+    throw markLocalRateLimitError(error as Error, LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE);
+  }
+  throw error;
+}
+
+/** What is left of a queue-wait budget that started at `startedAt`. */
+export function remainingQueueBudgetMs(budgetMs: number, startedAt: number): number {
+  return Math.max(0, budgetMs - (Date.now() - startedAt));
+}

open-sse/services/providerDefaultRateLimit.ts

@@ -17,6 +17,10 @@
  * atomically.
  */
 import { SlidingWindowLimiter, type RateLimitWindow } from "./slidingWindowLimiter.ts";
+import {
+  markLocalRateLimitError,
+  LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE,
+} from "./rateLimitManager/errors.ts";
 
 // Opt-in per-provider caps. Example shape (commented — add real entries as needed):
 //   "some-headerless-provider": { requests: 60, windowMs: 60_000 },
@@ -128,16 +132,36 @@ export async function awaitProviderDefaultSlot(
   provider: string,
   connectionId: string | null,
   signal: AbortSignal | null,
-  maxWaitMs?: number
+  remainingBudgetMs?: number
 ): Promise<void> {
   const cfg = getProviderDefaultRateLimit(provider);
   if (!cfg) return;
-  const budget = Math.max(cfg.windowMs, maxWaitMs && maxWaitMs > 0 ? maxWaitMs : 0);
+  if (typeof remainingBudgetMs === "number" && remainingBudgetMs <= 0)
+    throw markLocalRateLimitError(
+      new Error(
+        `Queue budget exhausted before provider-default slot (remaining=${remainingBudgetMs}ms)`
+      ),
+      LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE
+    );
+  const budget =
+    typeof remainingBudgetMs === "number" && Number.isFinite(remainingBudgetMs)
+      ? Math.min(remainingBudgetMs, cfg.windowMs)
+      : Math.max(cfg.windowMs, 0);
   const start = Date.now();
   for (;;) {
     const waitMs = acquireProviderDefaultSlot(provider, connectionId);
     if (waitMs === 0) return;
-    if (Date.now() - start >= budget) return; // waited the budget; let it through
-    await sleepOrAbort(Math.min(waitMs, budget), signal);
+    if (Date.now() - start >= budget)
+      throw markLocalRateLimitError(
+        new Error(`Provider-default slot wait exceeded budget ${budget}ms`),
+        LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE
+      );
+    const remainingBudget = budget - (Date.now() - start);
+    if (remainingBudget <= 0)
+      throw markLocalRateLimitError(
+        new Error(`Provider-default slot budget exhausted`),
+        LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE
+      );
+    await sleepOrAbort(Math.min(waitMs, remainingBudget), signal);
   }
 }

open-sse/services/rateLimitManager.ts

@@ -31,9 +31,14 @@ import {
   markLocalRateLimitError,
   RATE_LIMIT_EXECUTION_TIMEOUT_CODE,
   RATE_LIMIT_QUEUE_WEDGED_CODE,
+  LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE,
 } from "./rateLimitManager/errors";
 import { LimiterWedgeWatchdog, WATCHDOG_INTERVAL_MS } from "./rateLimitManager/wedgeWatchdog";
 import { toNumber } from "@/shared/utils/numeric";
+import {
+  getExecutorTimeoutMs,
+  resolveConnectionTimeoutMs,
+} from "../handlers/chatCore/upstreamTimeouts.ts";
 
 interface LearnedLimitEntry {
   provider: string;
@@ -193,9 +198,15 @@ export function resolveRequestQueueMaxWaitMs(
  * after a job leaves QUEUED; bounds execution, never queue wait. Kept strictly
  * separate from the queue-wait budget (`maxWaitMs`) so the backstop cannot
  * undercut upstream fetch-start timeouts on non-incremental gateways.
+ * Per-connection `executionMaxWaitMs` in `provider_connections.rateLimitOverrides`
+ * overrides the global setting when present (same precedence as `maxWaitMs`).
  */
-export function resolveExecutionMaxWaitMs(): number {
-  return currentRequestQueueSettings.executionMaxWaitMs;
+export function resolveExecutionMaxWaitMs(connectionId?: string): number {
+  const override = connectionId
+    ? (connectionRateLimitOverrides.get(connectionId) as Record<string, number> | undefined)
+        ?.executionMaxWaitMs
+    : undefined;
+  return resolveOverride(override, currentRequestQueueSettings.executionMaxWaitMs);
 }
 
 function buildLimiterDefaults() {
@@ -566,7 +577,18 @@ function getLimiter(provider, connectionId, model = null) {
  * @param {AbortSignal} signal - Optional abort signal to cancel waiting
  * @returns {Promise<unknown>} Result of fn()
  */
-export async function withRateLimit(provider, connectionId, model, fn, signal = null) {
+export async function withRateLimit(
+  provider,
+  connectionId,
+  model,
+  fn,
+  signal = null,
+  remainingBudgetMs = undefined,
+  correlationId = undefined,
+  opts:
+    | { executor?: { getTimeoutMs?: () => unknown }; providerSpecificData?: unknown }
+    | undefined = undefined
+) {
   if (!enabledConnections.has(connectionId)) {
     return fn();
   }
@@ -579,10 +601,36 @@ export async function withRateLimit(provider, connectionId, model, fn, signal =
     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`.
-  const maxWaitMs = resolveRequestQueueMaxWaitMs(provider, undefined, connectionId);
-  await awaitProviderDefaultSlot(provider, connectionId, signal, maxWaitMs);
+  const queueBudgetMs = resolveRequestQueueMaxWaitMs(
+    provider,
+    undefined,
+    connectionId ?? undefined
+  );
+  const budgetForSlot =
+    typeof remainingBudgetMs === "number" && Number.isFinite(remainingBudgetMs)
+      ? remainingBudgetMs
+      : queueBudgetMs;
+  if (
+    typeof remainingBudgetMs === "number" &&
+    Number.isFinite(remainingBudgetMs) &&
+    remainingBudgetMs <= 0
+  ) {
+    throw markLocalRateLimitError(
+      new Error(`Queue budget exhausted before rate-limit (remaining=${remainingBudgetMs}ms)`),
+      LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE
+    );
+  }
+  const slotStart = Date.now();
+  await awaitProviderDefaultSlot(provider, connectionId, signal, budgetForSlot);
+  const elapsedSlot = Date.now() - slotStart;
+  const remainingForQueue =
+    typeof remainingBudgetMs === "number" && Number.isFinite(remainingBudgetMs)
+      ? Math.max(0, remainingBudgetMs - elapsedSlot)
+      : queueBudgetMs;
+  if (correlationId)
+    logRateLimit(
+      `[RATE-LIMIT] cid=${correlationId} provider=${provider} remainingForQueue=${remainingForQueue}ms`
+    );
 
   const limiter = getLimiter(provider, connectionId, model);
   // Bottleneck's `expiration` starts only after a job leaves QUEUED, so it
@@ -591,7 +639,26 @@ export async function withRateLimit(provider, connectionId, model, fn, signal =
   // never by the queue-wait budget: non-incremental gateways legitimately run
   // for minutes before first bytes, and an expiration at the queue budget
   // killed them mid-flight (false 504s on opencode-go/glm-5.3-flash).
-  const executionExpirationMs = resolveExecutionMaxWaitMs();
+  // Per-connection executionMaxWaitMs wins, but never undercuts the upstream
+  // fetch-start timeout — otherwise the backstop kills a healthy mid-flight
+  // response (regression #12025 on GLM/thinking models).
+  const perConnExec = resolveExecutionMaxWaitMs(connectionId ?? undefined);
+  const upstreamMs = opts?.executor
+    ? getExecutorTimeoutMs(
+        opts.executor as unknown,
+        provider,
+        model ?? undefined,
+        resolveConnectionTimeoutMs(
+          opts.providerSpecificData as Record<string, unknown> | null | undefined
+        )
+      )
+    : undefined;
+  const executionExpirationMs = upstreamMs ? Math.max(perConnExec, upstreamMs) : perConnExec;
+  if (upstreamMs && perConnExec < upstreamMs) {
+    logRateLimit(
+      `[RATE-LIMIT] executionMaxWaitMs ${perConnExec}ms clamped to upstream ${upstreamMs}ms for ${provider}/${model ?? ""}`
+    );
+  }
   const scheduleOpts =
     executionExpirationMs && executionExpirationMs > 0 ? { expiration: executionExpirationMs } : {};
 
@@ -610,6 +677,43 @@ export async function withRateLimit(provider, connectionId, model, fn, signal =
     throw admissionErr;
   }
 
+  const queueRemainingMs = remainingForQueue;
+  let queueTimedOut = false;
+  let delayId: ReturnType<typeof setTimeout> | null = null;
+  const queueTimeoutErr = markLocalRateLimitError(
+    new Error(
+      `Request exceeded queue budget maxWaitMs=${queueRemainingMs}ms for ${provider}/${model ?? ""} — queue budget does not bound execution (executionMaxWaitMs=${executionExpirationMs}ms)`
+    ),
+    LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE
+  );
+  if (queueRemainingMs <= 0) throw queueTimeoutErr;
+  const timeoutPromise = new Promise<never>((_, reject) => {
+    delayId = setTimeout(() => {
+      queueTimedOut = true;
+      reject(queueTimeoutErr);
+    }, queueRemainingMs);
+  });
+  timeoutPromise.catch(() => {});
+  // Clear the queue-wait timer once the job leaves QUEUED and starts executing.
+  // Without this, the timer would also bound execution (queueRemainingMs ≈ 40ms
+  // would kill a 300ms execution that correctly left the queue immediately).
+  const wrappedFn = () => {
+    if (queueTimedOut) return Promise.reject(queueTimeoutErr);
+    if (delayId) {
+      clearTimeout(delayId);
+      delayId = null;
+    }
+    return (fn as unknown as (s?: AbortSignal) => Promise<unknown>)(signal ?? undefined);
+  };
+  const scheduled = limiter.schedule(scheduleOpts, wrappedFn as unknown as () => Promise<unknown>);
+  scheduled.catch(() => {});
+  // Note: if timeoutPromise wins while the job is still QUEUED (blocked by
+  // maxConcurrent), Bottleneck cannot cancel it — wrappedFn rejects only on
+  // dispatch after the slot frees. Until then counts().QUEUED stays 1 and
+  // maxQueueDepth admission sees an inflated depth transiently; this is
+  // inherent to Bottleneck (no cancelQueuedJob) and does not affect
+  // correctness since fnCalled stays false.
+
   try {
     if (signal) {
       let abortListener: (() => void) | undefined;
@@ -635,23 +739,28 @@ export async function withRateLimit(provider, connectionId, model, fn, signal =
         abortListener = onAbort;
         signal.addEventListener("abort", abortListener, { once: true });
       }
+      abortPromise.catch(() => {});
 
       try {
-        // Race the work against the abort signal. When abort wins, fn is still
-        // running inside Bottleneck's limiter — its eventual rejection must not
-        // surface as an unhandledRejection. The .catch(noop) silences only the
-        // orphaned branch; the real rejection comes from abortPromise.
-        const scheduled = limiter.schedule(scheduleOpts, fn);
-        scheduled.catch(() => {}); // prevent unhandledRejection when abort wins
-        abortPromise.catch(() => {}); // prevent unhandledRejection when scheduled wins
-        return await Promise.race([scheduled, abortPromise]);
+        return await Promise.race([scheduled, timeoutPromise, abortPromise]);
       } finally {
+        if (delayId) {
+          clearTimeout(delayId);
+          delayId = null;
+        }
         if (abortListener) {
           signal.removeEventListener("abort", abortListener);
         }
       }
     } else {
-      return await limiter.schedule(scheduleOpts, fn);
+      try {
+        return await Promise.race([scheduled, timeoutPromise]);
+      } finally {
+        if (delayId) {
+          clearTimeout(delayId);
+          delayId = null;
+        }
+      }
     }
   } catch (err) {
     // Only Bottleneck-owned failures are rewritten. Application code can throw

src/lib/resilience/settings/types.ts

@@ -28,6 +28,8 @@ export interface RequestQueueSettings {
    * only after a job leaves QUEUED). Kept separate from `maxWaitMs` because
    * non-incremental gateways legitimately take minutes before first bytes;
    * the backstop must never undercut the upstream fetch-start timeout.
+   * Per-connection `rateLimitOverrides.executionMaxWaitMs` can override this
+   * global default (bounded 0..600000 via provider schema; 0 falls through).
    */
   executionMaxWaitMs: number;
   /**

src/shared/validation/schemas/provider.ts

@@ -560,6 +560,7 @@ export const updateProviderConnectionSchema = z
         minTime: rateLimitOverrideNumber(60_000).optional(),
         maxConcurrent: rateLimitOverrideNumber(10_000).optional(),
         maxWaitMs: rateLimitOverrideNumber(120_000).optional(),
+        executionMaxWaitMs: rateLimitOverrideNumber(600_000).optional(),
       })
       .partial()
       .strict()

tests/unit/rate-limit-execution-per-conn.test.ts

@@ -0,0 +1,121 @@
+import { describe, it, beforeEach, afterEach } from "node:test";
+import assert from "node:assert/strict";
+
+import * as rateLimitManager from "../../open-sse/services/rateLimitManager.ts";
+import { updateProviderConnectionSchema } from "../../src/shared/validation/schemas/provider.ts";
+import * as resilienceSettings from "../../src/lib/resilience/settings.ts";
+
+function parse(overrides: unknown) {
+  return updateProviderConnectionSchema.safeParse({ name: "test", rateLimitOverrides: overrides });
+}
+
+describe("rateLimitOverrides executionMaxWaitMs per-connection", () => {
+  it("accepts executionMaxWaitMs per-connection", () => {
+    const r = parse({ executionMaxWaitMs: 300000 });
+    assert.ok(r.success, String((r as { error?: unknown }).error));
+    assert.equal(
+      (r as { data: { rateLimitOverrides: { executionMaxWaitMs: number } } }).data
+        .rateLimitOverrides.executionMaxWaitMs,
+      300000
+    );
+  });
+
+  it("rejects unknown key still strict", () => {
+    const r = parse({ bogus: 1 } as unknown as Record<string, unknown>);
+    assert.equal(r.success, false);
+  });
+});
+
+describe("resolveExecutionMaxWaitMs per-connection", () => {
+  beforeEach(async () => {
+    await rateLimitManager.__resetRateLimitManagerForTests();
+  });
+
+  afterEach(async () => {
+    await rateLimitManager.__resetRateLimitManagerForTests();
+  });
+
+  it("per-connection executionMaxWaitMs wins over global fallback", async () => {
+    await rateLimitManager.applyRequestQueueSettings({
+      ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+      executionMaxWaitMs: 600000,
+    });
+    rateLimitManager.refreshConnectionRateLimits("conn1", {
+      executionMaxWaitMs: 120000,
+    } as unknown as Record<string, number>);
+    assert.equal(rateLimitManager.resolveExecutionMaxWaitMs("conn1"), 120000);
+    assert.equal(rateLimitManager.resolveExecutionMaxWaitMs("conn2"), 600000);
+    assert.equal(rateLimitManager.resolveExecutionMaxWaitMs(undefined), 600000);
+  });
+
+  it("execMs is clamped to upstream timeout — never undercuts mid-flight", async () => {
+    await rateLimitManager.applyRequestQueueSettings({
+      ...resilienceSettings.DEFAULT_RESILIENCE_SETTINGS.requestQueue,
+      autoEnableApiKeyProviders: false,
+      concurrentRequests: 1,
+      requestsPerMinute: 100000,
+      minTimeBetweenRequestsMs: 0,
+      executionMaxWaitMs: 5000,
+    });
+
+    const conn = "exec-clamp-conn";
+    rateLimitManager.enableRateLimitProtection(conn);
+    rateLimitManager.refreshConnectionRateLimits(conn, {
+      executionMaxWaitMs: 10000,
+    } as unknown as Record<string, number>);
+    assert.equal(rateLimitManager.resolveExecutionMaxWaitMs(conn), 10000);
+
+    // Proof that clamp uses max(perConn, upstream): a tight per-conn budget of 50ms
+    // would kill a 150ms execution, but when upstream is 30s the effective
+    // expiration is 30s so the same 150ms execution must survive. Without the
+    // max() clamp it would fail with RATE_LIMIT_EXECUTION_TIMEOUT (504).
+    const tightPerConnMs = 50;
+    const upstreamMs = 30000;
+    rateLimitManager.refreshConnectionRateLimits(conn, {
+      executionMaxWaitMs: tightPerConnMs,
+    } as unknown as Record<string, number>);
+    assert.equal(rateLimitManager.resolveExecutionMaxWaitMs(conn), tightPerConnMs);
+    const fakeExecutor = { getTimeoutMs: () => upstreamMs };
+    const result = await rateLimitManager.withRateLimit(
+      "openai",
+      conn,
+      "gpt-4o",
+      async () => {
+        await new Promise((r) => setTimeout(r, 150));
+        return "ok";
+      },
+      null,
+      undefined,
+      undefined,
+      {
+        executor: fakeExecutor as unknown as { getTimeoutMs: () => unknown },
+        providerSpecificData: {},
+      }
+    );
+    assert.equal(
+      result,
+      "ok",
+      "150ms execution must survive: clamped expiration is 30s, not tight 50ms"
+    );
+
+    // Conversely, per-conn larger than upstream wins.
+    rateLimitManager.refreshConnectionRateLimits(conn, {
+      executionMaxWaitMs: 50000,
+    } as unknown as Record<string, number>);
+    const fakeShort = { getTimeoutMs: () => 5000 };
+    const result2 = await rateLimitManager.withRateLimit(
+      "openai",
+      conn,
+      "gpt-4o",
+      async () => "ok2",
+      null,
+      undefined,
+      undefined,
+      {
+        executor: fakeShort as unknown as { getTimeoutMs: () => unknown },
+        providerSpecificData: {},
+      }
+    );
+    assert.equal(result2, "ok2");
+  });
+});

tests/unit/rate-limit-manager-queue-bound.test.ts

@@ -0,0 +1,180 @@
+import { describe, it, beforeEach, afterEach } from "node:test";
+import assert from "node:assert/strict";
+
+import * as rateLimitManager from "../../open-sse/services/rateLimitManager.ts";
+import {
+  getTrustedLocalRateLimitError,
+  LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE,
+} from "../../open-sse/services/rateLimitManager/errors.ts";
+import { __setProviderDefaultRateLimitsForTests } from "../../open-sse/services/providerDefaultRateLimit.ts";
+
+describe("queue wait bound - anti-orphan", () => {
+  beforeEach(async () => {
+    await rateLimitManager.__resetRateLimitManagerForTests();
+    __setProviderDefaultRateLimitsForTests(null);
+  });
+
+  afterEach(async () => {
+    await rateLimitManager.__resetRateLimitManagerForTests();
+    __setProviderDefaultRateLimitsForTests(null);
+  });
+
+  it("queue wait bound - timeout before dispatch (<80ms not 200ms)", async () => {
+    // Force a single concurrency slot so the second request queues behind the first.
+    // Using a per-connection override isolates the fixture from other tests.
+    const conn = "queue-bound-conn";
+    rateLimitManager.enableRateLimitProtection(conn);
+    rateLimitManager.refreshConnectionRateLimits(conn, { maxConcurrent: 1 });
+
+    // First request occupies the only slot for 200ms.
+    const p1 = rateLimitManager.withRateLimit(
+      "openai",
+      conn,
+      "gpt-4",
+      () => new Promise<string>((resolve) => setTimeout(() => resolve("ok"), 200)),
+      null,
+      1000
+    );
+
+    // Let p1 acquire the slot.
+    await new Promise((r) => setTimeout(r, 15));
+
+    // Second request targets the same limiter (same provider/connection/model)
+    // with a tiny remaining budget (30ms). It should fail with queue timeout
+    // quickly, not wait ~200ms for p1 to finish.
+    let fnCalled = false;
+    const start = Date.now();
+    await assert.rejects(
+      () =>
+        rateLimitManager.withRateLimit(
+          "openai",
+          conn,
+          "gpt-4",
+          () => {
+            fnCalled = true;
+            return Promise.resolve("second");
+          },
+          null,
+          30
+        ),
+      (err: unknown) => {
+        const trusted = getTrustedLocalRateLimitError(err);
+        assert.equal(trusted?.code, LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE);
+        assert.equal(trusted?.status, 503);
+        return true;
+      }
+    );
+    const elapsed = Date.now() - start;
+    assert.ok(
+      elapsed < 80,
+      `queue timeout should win quickly, elapsed=${elapsed}ms (expected <80ms)`
+    );
+    assert.ok(elapsed >= 15, `should have waited close to budget, elapsed=${elapsed}ms`);
+    assert.equal(fnCalled, false, "queued function must not run after timeout");
+
+    // Drain p1 to let Bottleneck dispatch and then reject the timed-out job.
+    await p1.catch(() => {});
+    // Bottleneck moves the queued job to EXECUTING only after a slot frees,
+    // so wait briefly for the anti-orphan wrappedFn to reject.
+    await new Promise((r) => setTimeout(r, 30));
+
+    const stateAfter = await rateLimitManager.__getLimiterStateForTests("openai", conn, "gpt-4");
+    if (stateAfter) assert.equal(stateAfter.queued, 0, "no orphan QUEUED after queue timeout");
+
+    // Give the orphaned scheduled promise a chance to settle before next test.
+    await new Promise((r) => setTimeout(r, 10));
+  });
+
+  it("remaining budget zero does not schedule work", async () => {
+    const conn = "queue-remaining-zero";
+    rateLimitManager.enableRateLimitProtection(conn);
+
+    let fnCalled = false;
+    await assert.rejects(
+      () =>
+        rateLimitManager.withRateLimit(
+          "openai",
+          conn,
+          "gpt-4",
+          async () => {
+            fnCalled = true;
+            return "ok";
+          },
+          null,
+          0
+        ),
+      (err: unknown) => {
+        const trusted = getTrustedLocalRateLimitError(err);
+        return trusted?.code === LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE && trusted?.status === 503;
+      }
+    );
+    assert.equal(fnCalled, false, "function must not be called when remaining is 0");
+    const state = await rateLimitManager.__getLimiterStateForTests("openai", conn, "gpt-4");
+    if (state) assert.equal(state.queued, 0);
+  });
+
+  it("abort and queue timeout are distinguished", async () => {
+    const conn = "queue-abort-distinct";
+    rateLimitManager.enableRateLimitProtection(conn);
+    rateLimitManager.refreshConnectionRateLimits(conn, { maxConcurrent: 1 });
+
+    // Occupy the slot so the abort-tracked request actually queues.
+    const p1 = rateLimitManager.withRateLimit(
+      "openai",
+      conn,
+      "gpt-4",
+      () => new Promise<string>((r) => setTimeout(() => r("hold"), 300)),
+      null,
+      1000
+    );
+    await new Promise((r) => setTimeout(r, 10));
+
+    const ac = new AbortController();
+    setTimeout(() => ac.abort(), 15);
+
+    await assert.rejects(
+      () =>
+        rateLimitManager.withRateLimit(
+          "openai",
+          conn,
+          "gpt-4",
+          () => new Promise(() => {}),
+          ac.signal,
+          1000
+        ),
+      (err: unknown) => {
+        assert.equal((err as Error).name, "AbortError");
+        // Abort must not be branded as a local queue timeout.
+        const trusted = getTrustedLocalRateLimitError(err);
+        assert.equal(trusted, null, "AbortError must not be branded as queue timeout");
+        return true;
+      }
+    );
+
+    await p1.catch(() => {});
+  });
+
+  it("negative remaining budget is rejected without enqueuing", async () => {
+    const conn = "queue-negative-remaining";
+    rateLimitManager.enableRateLimitProtection(conn);
+
+    let fnCalled = false;
+    await assert.rejects(
+      () =>
+        rateLimitManager.withRateLimit(
+          "openai",
+          conn,
+          "gpt-4",
+          async () => {
+            fnCalled = true;
+            return 1;
+          },
+          null,
+          -5
+        ),
+      (err: unknown) =>
+        getTrustedLocalRateLimitError(err)?.code === LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE
+    );
+    assert.equal(fnCalled, false);
+  });
+});

tests/unit/rate-limit-remaining-budget.test.ts

@@ -0,0 +1,138 @@
+import { describe, it, beforeEach, afterEach } from "node:test";
+import assert from "node:assert/strict";
+
+import {
+  withRateLimit,
+  __resetRateLimitManagerForTests,
+  __getLimiterStateForTests,
+  enableRateLimitProtection,
+} from "../../open-sse/services/rateLimitManager.ts";
+import {
+  awaitProviderDefaultSlot,
+  acquireProviderDefaultSlot,
+  __setProviderDefaultRateLimitsForTests,
+} from "../../open-sse/services/providerDefaultRateLimit.ts";
+import {
+  getTrustedLocalRateLimitError,
+  LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE,
+} from "../../open-sse/services/rateLimitManager/errors.ts";
+
+describe("remaining budget shared across gate+slot+queue", () => {
+  beforeEach(async () => {
+    await __resetRateLimitManagerForTests();
+    __setProviderDefaultRateLimitsForTests(null);
+  });
+  afterEach(async () => {
+    await __resetRateLimitManagerForTests();
+    __setProviderDefaultRateLimitsForTests(null);
+  });
+
+  it("remaining budget is enforced via withRateLimit (gate budget shared to queue)", async () => {
+    // withRateLimit must honor remainingBudgetMs (per-connection queueBudgetMs minus elapsed).
+    // Before: remaining was ignored and fn ran even when budget was exhausted.
+    // After: remaining is the budget shared across gate+slot+queue.
+    const conn = "test-remaining-1";
+    enableRateLimitProtection(conn);
+    // Use a provider without a default slot (openai) to isolate the queue guard.
+    // A call with remaining 0 must throw LEGACY 503 without enqueuing.
+    await assert.rejects(
+      () => withRateLimit("openai", conn, "gpt-4", async () => "should-not-run", null, 0),
+      (err: unknown) => {
+        const trusted = getTrustedLocalRateLimitError(err);
+        assert.equal(trusted?.code, LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE);
+        assert.equal(trusted?.status, 503);
+        return true;
+      }
+    );
+    const state = await __getLimiterStateForTests("openai", conn, "gpt-4");
+    if (state) assert.equal(state.queued, 0, "remaining<=0 must not enqueue");
+  });
+
+  it("remaining<=0 does not enqueue — no QUEUED increment, no timer", async () => {
+    const conn = "test-remaining-2";
+    enableRateLimitProtection(conn);
+    // withRateLimit(..., remaining=0) must throw LEGACY 503 without limiter.schedule
+    let fnCalled = false;
+    await assert.rejects(
+      () =>
+        withRateLimit(
+          "openai",
+          conn,
+          "gpt-4",
+          async () => {
+            fnCalled = true;
+            return "ok";
+          },
+          null,
+          0
+        ),
+      (err: unknown) => {
+        const trusted = getTrustedLocalRateLimitError(err);
+        return trusted?.code === LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE;
+      }
+    );
+    assert.equal(fnCalled, false, "fn must not be called when remaining<=0");
+    const state = await __getLimiterStateForTests("openai", conn, "gpt-4");
+    if (state) assert.equal(state.queued, 0);
+  });
+
+  it("awaitProviderDefaultSlot fail-closed throws 503 instead of letting request through", async () => {
+    // Use a short test window to keep the test fast.
+    const provider = "budget-test-provider";
+    const conn = "conn-budget";
+    __setProviderDefaultRateLimitsForTests({ [provider]: { requests: 1, windowMs: 500 } });
+    try {
+      // Saturate the window - first request consumes the slot.
+      const first = acquireProviderDefaultSlot(provider, conn);
+      assert.equal(first, 0, "1st slot should be allowed");
+      const start = Date.now();
+      await assert.rejects(
+        () => awaitProviderDefaultSlot(provider, conn, null, 100),
+        (err: unknown) => {
+          const trusted = getTrustedLocalRateLimitError(err);
+          assert.equal(trusted?.code, LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE);
+          assert.equal(trusted?.status, 503);
+          return true;
+        }
+      );
+      const elapsed = Date.now() - start;
+      // Must fail-closed after ~100ms, not let it through after 500ms.
+      assert.ok(elapsed < 400, `must fail-closed within 100ms budget, elapsed=${elapsed}ms`);
+      assert.ok(elapsed >= 80, `must have waited close to budget, elapsed=${elapsed}ms`);
+    } finally {
+      __setProviderDefaultRateLimitsForTests(null);
+    }
+  });
+
+  it("withRateLimit compat — no remaining falls back to queueBudgetMs (no regression)", async () => {
+    const conn = "test-compat";
+    enableRateLimitProtection(conn);
+    const result = await withRateLimit("openai", conn, "gpt-4", async () => "ok-compat");
+    assert.equal(result, "ok-compat");
+    // also with a signal but no remaining
+    const ac = new AbortController();
+    const result2 = await withRateLimit("openai", conn, "gpt-4", async () => "ok2", ac.signal);
+    assert.equal(result2, "ok2");
+  });
+
+  it("providerDefaultSlot guard remaining<=0 throws immediate 503", async () => {
+    const provider = "budget-test-provider2";
+    __setProviderDefaultRateLimitsForTests({ [provider]: { requests: 1, windowMs: 60000 } });
+    try {
+      await assert.rejects(
+        () => awaitProviderDefaultSlot(provider, "any-conn", null, 0),
+        (err: unknown) => {
+          const t = getTrustedLocalRateLimitError(err);
+          return t?.code === LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE && t?.status === 503;
+        }
+      );
+      await assert.rejects(
+        () => awaitProviderDefaultSlot(provider, "any-conn", null, -5),
+        (err: unknown) =>
+          getTrustedLocalRateLimitError(err)?.code === LEGACY_RATE_LIMIT_QUEUE_TIMEOUT_CODE
+      );
+    } finally {
+      __setProviderDefaultRateLimitsForTests(null);
+    }
+  });
+});