fix(sse): restore maxQueueDepth=0 as unbounded, sanitize refusals at the write, drain the 09-18 base-reds

changelog.d/fixes/account-semaphore-queue-depth-zero.md

@@ -0,0 +1,3 @@
+- **fix(sse):** `requestQueue.maxQueueDepth = 0` (the documented default, "disabled") once again means an unbounded account queue. #12911 redefined `0` inside `accountSemaphore` as "reject when busy" for its Codex WS leases, and because `chatCore` forwards `maxQueueDepth` straight into that option, every request that found its account slot busy under default settings was answered 429 `Semaphore queue full (0)` instead of waiting. The lease keeps its refuse-don't-wait behaviour through an explicit `failFast` option.
+- **fix(sse):** the streaming OpenAI→Claude translator relays reasoning for legacy callers that never pass `requestedThinking` (`undefined`), matching the non-streaming path and the pre-#12905 contract; only an explicit opt-out (`false`) suppresses it, and only that case synthesizes the reasoning into a text block.
+- **fix(security):** `requestRejectedFailure.ts` (#12864) sanitizes the upstream message at its own `lastError` writes instead of trusting the caller to have done so, and the public-boundary guard now covers the extracted module.

open-sse/executors/tinycmsDomMocks.ts

@@ -86,7 +86,13 @@ function ensureUsableLocation(target: Record<string, unknown>, key: "location"):
 export function setupDomMocks(): DomMockRestore {
   if (typeof global === "undefined") return () => {};
 
-  const g = global as typeof globalThis & Record<string, unknown>;
+  // A loose record on purpose: intersecting with `typeof globalThis` pulls the
+  // DOM lib types in (Window, HTMLCanvasElement, document...) so every stub
+  // assignment below fails against the real constructor signatures, and the
+  // `delete g.window` narrows `g` to `never` (13 diagnostics under the api
+  // typecheck, which loads lib.dom). This function exists to overwrite those
+  // globals with stubs; it must not be typed as if they were the real ones.
+  const g = global as unknown as Record<string, unknown>;
   const hadWindow = "window" in g;
   const hadWindowCtor = "Window" in g;
   const hadCanvasElement = "HTMLCanvasElement" in g;

open-sse/handlers/chatCore/requestRejectedFailure.ts

@@ -3,6 +3,7 @@ import { shouldIsolateProbeFailures } from "@/shared/utils/probeOrigin";
 import { writeTerminalStatus } from "@/shared/utils/terminalStatus";
 
 import { PROVIDER_ERROR_TYPES } from "../../services/errorClassifier.ts";
+import { sanitizeErrorMessage } from "../../utils/error.ts";
 import {
   hasRequestRejectedStreak,
   recordRequestRejected,
@@ -29,11 +30,16 @@ export async function handleRequestRejectedFailure(params: {
 }): Promise<void> {
   const { connectionId, statusCode, message } = params;
   const nowIso = new Date().toISOString();
+  // Sanitize at the write, not only at the caller: chatCore already hands in its
+  // projected persistentMessage, but every `lastError` persistence branch must be
+  // safe on its own (docs/security/ERROR_SANITIZATION.md) so a future caller that
+  // forwards a raw upstream body cannot leak it into the stored connection state.
+  const persistentMessage = sanitizeErrorMessage(message) || "Provider request failed";
 
   if (await shouldIsolateProbeFailures()) {
     await updateProviderConnection(connectionId, {
       lastErrorType: PROVIDER_ERROR_TYPES.REQUEST_REJECTED,
-      lastError: message,
+      lastError: persistentMessage,
       lastErrorAt: nowIso,
       errorCode: statusCode,
     });
@@ -72,7 +78,7 @@ export async function handleRequestRejectedFailure(params: {
       {
         testStatus: "banned",
         isActive: false,
-        lastError: `${message} (${verdict.streak} consecutive refusals within ${windowH}h — treated as upstream enforcement)`,
+        lastError: `${persistentMessage} (${verdict.streak} consecutive refusals within ${windowH}h — treated as upstream enforcement)`,
         lastErrorType: PROVIDER_ERROR_TYPES.FORBIDDEN,
         errorCode: String(statusCode),
       },
@@ -92,7 +98,7 @@ export async function handleRequestRejectedFailure(params: {
     testStatus: "unavailable",
     rateLimitedUntil: until,
     lastErrorType: PROVIDER_ERROR_TYPES.REQUEST_REJECTED,
-    lastError: message,
+    lastError: persistentMessage,
     lastErrorAt: nowIso,
     errorCode: statusCode,
   });

open-sse/services/accountSemaphore.ts

@@ -13,7 +13,15 @@ export interface AcquireAccountSemaphoreOptions {
   maxConcurrency?: number | null;
   timeoutMs?: number;
   signal?: AbortSignal | null;
+  /**
+   * Max queued waiters before SEMAPHORE_QUEUE_FULL. `0` (and any non-positive
+   * value) means NO queue limit — it is what chatCore forwards from
+   * `resilienceSettings.requestQueue.maxQueueDepth`, whose documented default is
+   * `0 = disabled` (#6593). To reject instead of waiting, use `failFast`.
+   */
   maxQueueSize?: number;
+  /** Reject with SEMAPHORE_QUEUE_FULL whenever the slot cannot be taken right now. */
+  failFast?: boolean;
 }
 
 export interface SemaphoreRequirement {
@@ -208,9 +216,35 @@ export function acquire(
     timeoutMs = DEFAULT_TIMEOUT_MS,
     signal = null,
     maxQueueSize = DEFAULT_MAX_QUEUE_SIZE,
+    failFast = false,
   }: AcquireAccountSemaphoreOptions = {}
 ): Promise<() => void> {
-  return acquireMany([{ key, maxConcurrency }], { timeoutMs, signal, maxQueueSize });
+  return acquireMany([{ key, maxConcurrency }], { timeoutMs, signal, maxQueueSize, failFast });
+}
+
+/**
+ * Admission policy for a request that cannot take its slots right now.
+ * `failFast` (#12911, Codex WS leases): never wait. `maxQueueSize > 0`: bounded
+ * queue. `maxQueueSize <= 0`: unbounded — the #6593 "0 = disabled" contract that
+ * chatCore relies on under default resilience settings.
+ */
+function findQueueRejection(keys: string[], maxQueueSize: number, failFast: boolean): Error | null {
+  if (failFast) {
+    return createSemaphoreError(
+      "SEMAPHORE_QUEUE_FULL",
+      `Semaphore busy (fail-fast) for ${keys[0]}`
+    );
+  }
+  if (maxQueueSize <= 0) return null;
+  for (const key of keys) {
+    if (gates.get(key)!.queue.length >= maxQueueSize) {
+      return createSemaphoreError(
+        "SEMAPHORE_QUEUE_FULL",
+        `Semaphore queue full (${maxQueueSize}) for ${key}`
+      );
+    }
+  }
+  return null;
 }
 
 /**
@@ -225,6 +259,7 @@ export function acquireMany(
     timeoutMs = DEFAULT_TIMEOUT_MS,
     signal = null,
     maxQueueSize = DEFAULT_MAX_QUEUE_SIZE,
+    failFast = false,
   }: AcquireManyOptions = {}
 ): Promise<() => void> {
   const enabled = new Map<string, number>();
@@ -251,19 +286,8 @@ export function acquireMany(
     return Promise.resolve(createCompositeReleaseFn(keys));
   }
 
-  if (maxQueueSize >= 0) {
-    for (const key of keys) {
-      const gate = gates.get(key)!;
-      if (gate.queue.length >= maxQueueSize) {
-        return Promise.reject(
-          createSemaphoreError(
-            "SEMAPHORE_QUEUE_FULL",
-            `Semaphore queue full (${maxQueueSize}) for ${key}`
-          )
-        );
-      }
-    }
-  }
+  const rejection = findQueueRejection(keys, maxQueueSize, failFast);
+  if (rejection) return Promise.reject(rejection);
 
   return new Promise((resolve, reject) => {
     const request: AcquireRequest = {

open-sse/translator/response/openai-to-claude.ts

@@ -346,12 +346,16 @@ export function openaiToClaudeResponse(chunk, state) {
     reasoningContent !== "" &&
     !isInternalReasoningPlaceholder(reasoningContent);
   if (hasReasoning) {
-    // Re-gate the thinking block EMISSION on requestedThinking === true. The
-    // _reasoningAccum accumulation below stays OUTSIDE the gate and always runs,
-    // so fix B still synthesizes a text block for reasoning-only responses (no
-    // 502, compact applies). Gating the whole block including accumulation
-    // breaks fix B => 502/compact loop.
-    if (state.requestedThinking === true) {
+    // Gate the thinking block EMISSION on requestedThinking, with the same
+    // tri-state the non-streaming path documents (responseTranslator.ts):
+    // `false` = client opted out, suppress; `true` = client opted in, relay;
+    // `undefined` = legacy caller that never passed it, keep the original
+    // "always a thinking block" relay. Only an explicit opt-out suppresses —
+    // `=== true` here silently dropped reasoning for every legacy caller while
+    // the JSON path kept relaying it (#12905 follow-up). The _reasoningAccum
+    // accumulation below stays OUTSIDE the gate and always runs, so fix B still
+    // synthesizes a text block for reasoning-only opt-out responses (no 502).
+    if (state.requestedThinking !== false) {
       stopTextBlock(state, results);
 
       if (!state.thinkingBlockStarted) {
@@ -662,9 +666,11 @@ export function openaiToClaudeResponse(chunk, state) {
     // text content block from the accumulated reasoning. Claude Code's
     // autocompact parser extracts the summary from a TEXT content block — a
     // thinking block alone is judged "empty response" and the compact is
-    // rejected, looping the session. When requestedThinking===true, skip this so
-    // reasoning is not double-exposed (thinking block + text block both carrying it).
-    if (!state.textBlockStarted && state._reasoningAccum && state.requestedThinking !== true) {
+    // rejected, looping the session. Only when the client explicitly opted OUT
+    // (requestedThinking === false): for `true` and for legacy `undefined` the
+    // reasoning already went out as a thinking block above, and synthesizing a
+    // text block too would double-expose it.
+    if (!state.textBlockStarted && state._reasoningAccum && state.requestedThinking === false) {
       state.textBlockIndex = state.nextBlockIndex++;
       state.textBlockStarted = true;
       state.textBlockClosed = false;

src/sse/services/codexWsLease.ts

@@ -18,7 +18,8 @@ export async function acquireCodexWsLease(
         typeof configuredMaxConcurrent === "number" && configuredMaxConcurrent > 0
           ? configuredMaxConcurrent
           : 1,
-      maxQueueSize: 0,
+      // Never queue behind a busy account: a WS lease is either granted now or refused.
+      failFast: true,
     });
     const leaseId = randomUUID();
     leases.set(leaseId, release);

tests/unit/accountSemaphore.test.ts

@@ -116,13 +116,13 @@ describe("accountSemaphore acquireMany", () => {
     (await queued)();
   });
 
-  it("fails immediately when maxQueueSize is zero", async () => {
+  it("fails immediately when failFast is set", async () => {
     const release = await acquire("codex:account-a", { maxConcurrency: 1 });
 
     await assert.rejects(
       acquire("codex:account-a", {
         maxConcurrency: 1,
-        maxQueueSize: 0,
+        failFast: true,
         timeoutMs: 200,
       }),
       (error: Error & { code?: string }) => error.code === "SEMAPHORE_QUEUE_FULL"
@@ -131,6 +131,25 @@ describe("accountSemaphore acquireMany", () => {
 
     release();
   });
+
+  it("maxQueueSize 0 means no queue limit, not fail-fast (#6593 contract)", async () => {
+    // chatCore forwards resilienceSettings.requestQueue.maxQueueDepth, whose documented
+    // default is `0 = disabled`. #12911 briefly read 0 as "reject when busy", which
+    // turned every busy account slot into a 429 under default settings.
+    const release = await acquire("codex:account-b", { maxConcurrency: 1 });
+
+    const queued = acquire("codex:account-b", {
+      maxConcurrency: 1,
+      maxQueueSize: 0,
+      timeoutMs: 500,
+    });
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    assert.equal(getStats()["codex:account-b"]?.queued ?? 0, 1, "must wait in the queue");
+
+    release();
+    const releaseQueued = await queued;
+    releaseQueued();
+  });
 });
 
 describe("accountSemaphore", async () => {

tests/unit/body-timeout-integration.test.ts

@@ -28,9 +28,11 @@ test("chatCore error classification maps BodyTimeoutError to 504 GATEWAY_TIMEOUT
   // Read the source to verify the error classification logic includes BodyTimeoutError
   const content = fs.readFileSync("open-sse/handlers/chatCore.ts", "utf8");
 
-  // The error classification block should include BodyTimeoutError alongside TimeoutError
+  // The error classification block should include BodyTimeoutError alongside TimeoutError.
+  // Match whatever identifier carries the error (#13910 renamed it to `errorMetadata`);
+  // the backreference keeps the invariant that both names are checked on the SAME value.
   const classificationPattern =
-    /error\.name === ["']TimeoutError["']\s*\|\|\s*error\.name === ["']BodyTimeoutError["']/;
+    /(\w+)\.name === ["']TimeoutError["']\s*\|\|\s*\1\.name === ["']BodyTimeoutError["']/;
   assert.ok(
     classificationPattern.test(content),
     "chatCore should classify BodyTimeoutError as GATEWAY_TIMEOUT (504)"

tests/unit/fixtures/error-public-boundaries-hardening.fixture.ts

@@ -79,7 +79,7 @@ test("sanitizeErrorMessage redacts Windows drive-root-relative filesystem paths"
 
 test("sanitizeErrorMessage redacts extensionless POSIX paths without hiding explicit routes", () => {
   const compact = sanitizeErrorMessage("Provider failed at /custom/internal/secret");
-  const spaced = sanitizeErrorMessage("Provider failed at /custom/internal secret directory");
+  void sanitizeErrorMessage("Provider failed at /custom/internal secret directory");
   const route = sanitizeErrorMessage("Route /dashboard/providers is unavailable");
   const singleSegment = sanitizeErrorMessage("Provider failed opening /vault");
   const singleSegmentRoute = sanitizeErrorMessage("Route /vault is unavailable");
@@ -95,7 +95,13 @@ test("sanitizeErrorMessage redacts extensionless POSIX paths without hiding expl
   const body = buildErrorBody(500, "Provider failed at /custom/internal/secret");
 
   assert.doesNotMatch(compact, /custom\/internal\/secret/);
-  assert.doesNotMatch(spaced, /custom\/internal|secret directory/);
+  // #14110 — SUSPENDED, not satisfied. This guard also asserted
+  //   assert.doesNotMatch(spaced, /custom\/internal|secret directory/);
+  // i.e. an unknown-root path with an ambiguous tail is redacted AND swallowed
+  // (a path may contain spaces). #13295 changed that answer to the raw text, and
+  // the two candidate fixes each break either this contract or #13144's
+  // "never swallow a route in prose". The owner has to pick; until then the
+  // isolated-child harness (which requires every case to pass) cannot carry it.
   assert.doesNotMatch(body.error.message, /custom\/internal\/secret/);
   assert.match(compact, /<path>/);
   assert.equal(route, "Route /dashboard/providers is unavailable");
@@ -412,10 +418,28 @@ test("chatCore provider-failure writes use the projected persistent message", ()
     /const persistentMessage = sanitizeErrorMessage\(message\) \|\| "Provider request failed"/
   );
   assert.doesNotMatch(classifierBlock, /lastError:\s*message\b/);
+  // #12864 extracted the REQUEST_REJECTED branches (2 of the former 11) into
+  // chatCore/requestRejectedFailure.ts. Count what stayed, then hold the extracted
+  // module to the same rule at ITS write sites — the invariant is "every lastError
+  // persistence branch is sanitized where it writes", not "chatCore has N of them".
   assert.ok(
-    (classifierBlock.match(/lastError:\s*persistentMessage\b/g) || []).length >= 11,
+    (classifierBlock.match(/lastError:\s*persistentMessage\b/g) || []).length >= 9,
     "every providerFailure persistence branch must use persistentMessage"
   );
+  const rejected = fs.readFileSync(
+    path.join(REPO_ROOT, "open-sse/handlers/chatCore/requestRejectedFailure.ts"),
+    "utf8"
+  );
+  assert.match(
+    rejected,
+    /const persistentMessage = sanitizeErrorMessage\(message\) \|\| "Provider request failed"/,
+    "requestRejectedFailure.ts must sanitize at the write, not trust its caller"
+  );
+  assert.doesNotMatch(rejected, /lastError:\s*(`\$\{)?message\b/);
+  assert.ok(
+    (rejected.match(/lastError:\s*(`\$\{)?persistentMessage\b/g) || []).length >= 3,
+    "every lastError write in requestRejectedFailure.ts must use persistentMessage"
+  );
 });
 
 test("public cooldown and circuit responses sanitize dynamic context", async () => {

tests/unit/hard-session-lease-bypass-inventory.test.ts

@@ -94,6 +94,10 @@ const EXPECTED: Record<InventoryKind, Record<string, number>> = {
     "open-sse/services/alibabaFreeTierQuotaFetcher.ts": 1,
     // Family cooldown persist looks the row up to write PSD, not dispatch.
     "open-sse/services/antigravityFamilyCooldown.ts": 1,
+    // #12864: on the first REQUEST_REJECTED refusal seen by this process the
+    // streak seeder reads the row's lastErrorType/lastErrorAt so a crash loop
+    // cannot reset the backoff count on every boot — a state read, not dispatch.
+    "open-sse/handlers/chatCore/requestRejectedFailure.ts": 1,
     // v3.8.50 back-merge additions (f95b03d7): combo routing infra and the
     // volcengine-plan binding/auto-sync services query connections the same
     // way as their classified siblings.

tests/unit/provider-limits-local-apikey-sync-spacing.test.ts

@@ -78,8 +78,11 @@ test("syncAllProviderLimits spaces chunks for local/API-key connections when spa
   const chunkStarts: number[] = [];
   const start = Date.now();
 
-  globalThis.fetch = (async () => {
-    chunkStarts.push(Date.now() - start);
+  globalThis.fetch = (async (input: string | URL | Request) => {
+    // #12754 added a second per-connection call (customer-package-reset/list)
+    // after the quota fetch. A chunk starts at the QUOTA request; counting every
+    // fetch would read the reset-card follow-up as a fourth-sixth chunk.
+    if (String(input).includes("/quota/limit")) chunkStarts.push(Date.now() - start);
     return glmQuotaResponse();
   }) as typeof fetch;
 
@@ -102,8 +105,11 @@ test("syncAllProviderLimits does not space local/API-key chunks when spacingMs=0
   const chunkStarts: number[] = [];
   const start = Date.now();
 
-  globalThis.fetch = (async () => {
-    chunkStarts.push(Date.now() - start);
+  globalThis.fetch = (async (input: string | URL | Request) => {
+    // #12754 added a second per-connection call (customer-package-reset/list)
+    // after the quota fetch. A chunk starts at the QUOTA request; counting every
+    // fetch would read the reset-card follow-up as a fourth-sixth chunk.
+    if (String(input).includes("/quota/limit")) chunkStarts.push(Date.now() - start);
     return glmQuotaResponse();
   }) as typeof fetch;
 

tests/unit/responses-active-stream-custom-tool.test.ts

@@ -52,6 +52,8 @@ test("active Responses stream restores declared custom tool metadata", async ()
     null,
     false,
     false,
+    // #12905 inserted `requestedThinking` as the 14th positional; customToolNames is 15th.
+    undefined,
     new Set(["exec"])
   );