fix(auth): compare-and-swap guard on OAuth refresh persist (#4038)

CHANGELOG.md

@@ -24,6 +24,7 @@ _In development — bullets added per PR; finalized at release._
 
 ### 🔧 Bug Fixes
 
+- **fix(auth): compare-and-swap guard on the OAuth refresh persist** — under multi-agent load, the per-connection refresh mutex makes `[network refresh + DB write]` atomic for **one** connection, but it does not protect against a **third** writer (a sibling request, a concurrent HealthCheck, or a replica) landing a fresher `refresh_token` rotation on the same `connection_id` between the staleness read and the persist. Overwriting that fresher row reverts the sibling's rotation; the next caller then loads the now-consumed token, Auth0/Anthropic flag it as `refresh_token_reused`, and the whole token family gets revoked (the 1352× claude/`aa5dd5cf` invalidation storm). `getAccessToken` now re-reads the row's current `refresh_token` immediately before persisting (inside the mutex) and **skips the write** when it has rotated past the token the caller presented — the caller still receives the freshly-issued access token, only the DB overwrite is skipped. Opt-in via `runWithCasGuard` (no active guard ⇒ byte-identical behavior); skip/persist counters exposed via `getCasGuardStats()`. Regression guard: `tests/unit/token-refresh-cas-guard-4038.test.ts`. ([#4038](https://github.com/diegosouzapw/OmniRoute/issues/4038) — thanks @KooshaPari for the root-cause diagnosis)
 - **mcp:** break the `schemas/tools.ts ↔ schemas/toolSearch.ts` import cycle introduced when the `tool_search` defs (#5269) were extracted into their own module — `toolSearch.ts` imported `McpToolDefinition` from `tools.ts` while `tools.ts` imported `toolSearchTool` from `toolSearch.ts`, failing `check:cycles` on `release/v3.8.40`. The shared `AuditLevel` + `McpToolDefinition` types now live in a leaf `schemas/toolDefinition.ts` that both import; `tools.ts` re-exports them for backward compatibility.
 - **compression (analytics):** record attempted-but-no-op compression runs so Stacked is no longer invisible when it saves nothing. Previously a `compression_analytics` row was written only on a net-positive saving, so a Stacked (RTK→Caveman) pipeline that ran on already-compact context produced no row — indistinguishable from "never dispatched" (`byMode.stacked.count` stayed flat while Ultra climbed). Such runs are now recorded with `skip_reason` and surfaced as a per-mode `skipped` count plus `totalSkipped`/`bySkipReason` in the analytics summary and the Mode Breakdown; the existing net-saving totals/averages are unchanged (skip rows are excluded from them) (#4268 — thanks @abdulkadirozyurt, @androw)
 - **cli (tray):** fix `omniroute server --tray` showing no tray on macOS/Linux with no error printed. The wired Unix tray path loaded `systray2` through an inline loader that called `require("module")` inside an ESM `.mjs` file (`"type":"module"`) → `ReferenceError: require is not defined`, silently swallowed (regressed in v3.8.34); even if it had loaded, `systray2` isn't in `node_modules` (it's lazily installed into `~/.omniroute/runtime`). The loader now delegates to the runtime loader, the icon path (`icon.png`) is corrected, `isTemplateIcon` is `false` (the full-color icon rendered as a white square under macOS template mode), and tray start failures are surfaced to stderr instead of being swallowed (#4605 — thanks @ProgMEM-CC)

config/quality/file-size-baseline.json

@@ -165,7 +165,8 @@
     "_rebaseline_2026_06_29_5286_memoization": "PR #5286 own growth: strategySelector.ts 899->960 (+61 = the opt-in result-memoization branches in applyCompression/applyCompressionAsync — principal+determinism gate, makeMemoKey lookup/store with model+supportsVision folded into the key, recompute-with-memo-off). Default off (memoizeCompressionResults), so zero behavior change. The memo helpers live in the leaf resultMemo.ts (<cap); the chokepoint wiring here is not extractable. Structural shrink of this hot-path file tracked in #3501.",
     "open-sse/services/compression/strategySelector.ts": 997,
     "open-sse/services/rateLimitManager.ts": 1035,
-    "open-sse/services/tokenRefresh.ts": 2103,
+    "_rebaseline_2026_06_29_4038_cas_guard": "PR (#4038) own growth: tokenRefresh.ts 2103->2181 (+78 = the compare-and-swap guard on the refresh persist — runWithCasGuard/getActiveCasGuard AsyncLocalStorage pair mirroring runWithOnPersist, casGuardShouldSkipPersist that rereads the row right before persisting and skips the write when a concurrent writer already rotated the refresh_token past the one presented, plus getCasGuardStats counters). Fixes the sibling-rotation-revert → token-family-revocation storm. Gated behind an active guard (opt-in; no guard => byte-identical). Wiring lives at the two persist chokepoints inside getAccessToken; the comparison reuses wasRefreshTokenRotated from refreshSerializer. Not extractable without splitting the refresh hot path.",
+    "open-sse/services/tokenRefresh.ts": 2181,
     "open-sse/services/usage.ts": 3454,
     "open-sse/translator/request/openai-to-gemini.ts": 906,
     "open-sse/translator/request/openai-to-kiro.ts": 842,

open-sse/handlers/chatCore.ts

@@ -82,6 +82,7 @@ import {
   refreshWithRetry,
   isUnrecoverableRefreshError,
   runWithOnPersist,
+  runWithCasGuard,
 } from "../services/tokenRefresh.ts";
 import { createRequestLogger } from "../utils/requestLogger.ts";
 import { createPreparedRequestLogger, runWithCapture } from "../utils/providerRequestLogging.ts";
@@ -2903,8 +2904,25 @@ export async function handleChatCore({
         }
       : undefined;
 
+    // #4038: build a compare-and-swap reread so getAccessToken can skip the persist if a
+    // concurrent writer (sibling request / HealthCheck / replica) already rotated this
+    // connection's refresh_token past the one we presented — overwriting would revert it
+    // and revoke the token family. No connectionId ⇒ no guard (behavior unchanged).
+    const casConnectionId =
+      typeof credentials?.connectionId === "string" ? credentials.connectionId.trim() : "";
+    const casReread = casConnectionId
+      ? async () => {
+          const latest = await getProviderConnectionById(casConnectionId);
+          return typeof latest?.refreshToken === "string" ? latest.refreshToken : null;
+        }
+      : null;
+
     const newCredentials = (await refreshWithRetry(
-      () => runWithOnPersist(persistFn, () => executor.refreshCredentials(credentials, log)),
+      () =>
+        runWithCasGuard(
+          casReread ? { expectedRefreshToken: attemptedRefreshToken, reread: casReread } : null,
+          () => runWithOnPersist(persistFn, () => executor.refreshCredentials(credentials, log))
+        ),
       3,
       log,
       provider // Explicitly pass the provider to avoid universally tripping the "unknown" circuit breaker

open-sse/services/tokenRefresh.ts

@@ -4,7 +4,7 @@ import { PROVIDERS, OAUTH_ENDPOINTS } from "../config/constants.ts";
 import { getGitHubCopilotRefreshHeaders } from "../config/providerHeaderProfiles.ts";
 import { pbkdf2Sync } from "node:crypto";
 import { runWithProxyContext } from "../utils/proxyFetch.ts";
-import { serializeRefresh } from "./refreshSerializer.ts";
+import { serializeRefresh, wasRefreshTokenRotated } from "./refreshSerializer.ts";
 import { WINDSURF_CONFIG } from "@/lib/oauth/constants/oauth";
 import { buildGitLabOAuthEndpoints, resolveGitLabOAuthBaseUrl } from "@/lib/oauth/gitlab";
 
@@ -171,6 +171,81 @@ export function getActiveOnPersist(): RefreshPersistFn | undefined {
   return onPersistStore.getStore();
 }
 
+// ── #4038: compare-and-swap (CAS) guard on the refresh persist ───────────────
+// Fix A makes [network refresh + DB write] atomic *for a single connection's
+// mutex*. It does NOT protect against a THIRD writer (a sibling process, a
+// concurrent HealthCheck, or a replica) landing a fresher rotation on the same
+// `connection_id` between the moment the caller read the row and the moment this
+// persist runs. Overwriting that fresher row reverts the sibling's rotation, the
+// next caller loads the reverted (now-consumed) refresh_token, and Auth0/Anthropic
+// revoke the whole token family (the 1352× claude/aa5dd5cf invalidation storm).
+//
+// The CAS guard carries the refresh_token the caller PRESENTED (the version token,
+// since refresh_tokens rotate on every refresh) plus a `reread` of the row's
+// current refresh_token. Right before persisting, `getAccessToken` re-reads and, if
+// a concurrent writer already rotated the row past the presented token, SKIPS the
+// persist so the DB stays at the fresher state. The caller still receives the new
+// accessToken — upstream already authenticated the request; only the DB write is
+// skipped. No active guard ⇒ behavior is byte-identical to before (opt-in).
+type CasGuard = {
+  /** The refresh_token the caller presented for this refresh (CAS version token). */
+  expectedRefreshToken: string | null;
+  /** Re-reads the CURRENT persisted refresh_token for this connection (decrypted). */
+  reread: () => Promise<string | null | undefined>;
+};
+const casGuardStore = new AsyncLocalStorage<CasGuard>();
+const casGuardStats = { skipped: 0, persisted: 0 };
+
+export function runWithCasGuard<T>(
+  guard: CasGuard | undefined | null,
+  fn: () => Promise<T>
+): Promise<T> {
+  if (!guard) return fn();
+  return casGuardStore.run(guard, fn);
+}
+
+export function getActiveCasGuard(): CasGuard | undefined {
+  return casGuardStore.getStore();
+}
+
+/** Skip/persist counters for observability + tests. */
+export function getCasGuardStats(): { skipped: number; persisted: number } {
+  return { ...casGuardStats };
+}
+
+/** Test-only: reset the CAS counters between cases. */
+export function _resetCasGuardStats(): void {
+  casGuardStats.skipped = 0;
+  casGuardStats.persisted = 0;
+}
+
+/**
+ * Returns true when the persist should be SKIPPED because a concurrent writer
+ * already rotated the row's refresh_token past the one we presented (CAS mismatch).
+ * Best-effort: any reread failure falls through to persist (never blocks recovery).
+ */
+async function casGuardShouldSkipPersist(log?: RefreshLogger): Promise<boolean> {
+  const guard = getActiveCasGuard();
+  if (!guard || !guard.expectedRefreshToken) return false;
+  let current: string | null | undefined;
+  try {
+    current = await guard.reread();
+  } catch {
+    return false; // reread failed — fall through to persist (best-effort)
+  }
+  // wasRefreshTokenRotated is true iff both are non-empty AND current !== expected.
+  if (wasRefreshTokenRotated(guard.expectedRefreshToken, current)) {
+    casGuardStats.skipped++;
+    log?.warn?.(
+      "TOKEN_REFRESH",
+      "CAS guard: skipping persist — a concurrent writer already rotated the refresh_token (#4038)"
+    );
+    return true;
+  }
+  casGuardStats.persisted++;
+  return false;
+}
+
 type RefreshLogger = {
   info?: (tag: string, message: string, data?: Record<string, unknown>) => void;
   warn?: (tag: string, message: string, data?: Record<string, unknown>) => void;
@@ -1670,6 +1745,11 @@ export async function getAccessToken(
       // Invoke onPersist INSIDE the mutex so [network call + DB write] are one atomic step.
       // This prevents a concurrent waiter from reading stale credentials before the DB is updated.
       if (result?.accessToken && effectiveOnPersist) {
+        // #4038: skip the persist if a concurrent writer already rotated this row past the
+        // refresh_token we presented (compare-and-swap) — overwriting would revert it.
+        if (await casGuardShouldSkipPersist(log)) {
+          return result;
+        }
         try {
           await effectiveOnPersist(result);
         } catch (persistErr) {
@@ -1707,6 +1787,11 @@ export async function getAccessToken(
   )
     .then(async (result) => {
       if (result?.accessToken && effectiveOnPersist) {
+        // #4038: same compare-and-swap guard as Layer 1 — skip the persist if a concurrent
+        // writer already rotated this row past the refresh_token we presented.
+        if (await casGuardShouldSkipPersist(log)) {
+          return result;
+        }
         try {
           await effectiveOnPersist(result);
         } catch (persistErr) {

tests/unit/token-refresh-cas-guard-4038.test.ts

@@ -0,0 +1,114 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import {
+  getAccessToken,
+  runWithCasGuard,
+  getCasGuardStats,
+  _resetCasGuardStats,
+  _clearTokenRotationMap,
+} from "../../open-sse/services/tokenRefresh.ts";
+
+// #4038: the per-connection mutex makes [refresh + persist] atomic for ONE connection,
+// but a THIRD writer (sibling request / HealthCheck / replica) can land a fresher
+// refresh_token rotation between our staleness read and our persist. Overwriting it
+// reverts the sibling's rotation → the next caller loads a now-consumed token → Auth0
+// revokes the whole family (the 1352× claude invalidation storm). The CAS guard
+// re-reads the row right before persisting and SKIPS the write when the row's
+// refresh_token has rotated past the one we presented.
+
+const silentLog = { info() {}, warn() {}, error() {} };
+
+// Mock the Anthropic token endpoint so claude's refresh succeeds without a network call.
+function withMockedRefresh<T>(newRefreshToken: string, fn: () => Promise<T>): Promise<T> {
+  const realFetch = globalThis.fetch;
+  globalThis.fetch = (async () =>
+    new Response(
+      JSON.stringify({
+        access_token: "NEW_ACCESS_TOKEN",
+        refresh_token: newRefreshToken,
+        expires_in: 3600,
+      }),
+      { status: 200, headers: { "content-type": "application/json" } }
+    )) as never;
+  return fn().finally(() => {
+    globalThis.fetch = realFetch;
+  });
+}
+
+test("#4038 CAS guard SKIPS the persist when a concurrent writer rotated the refresh_token", async () => {
+  _resetCasGuardStats();
+  _clearTokenRotationMap();
+  let persisted = false;
+  const onPersist = async () => {
+    persisted = true;
+  };
+
+  const result = await withMockedRefresh("ROTATED_BY_US", () =>
+    runWithCasGuard(
+      // The row's CURRENT refresh_token is NOT the one we presented (R0): a sibling
+      // already rotated it to R_CONCURRENT while our network refresh was in flight.
+      { expectedRefreshToken: "R0", reread: async () => "R_CONCURRENT" },
+      () =>
+        getAccessToken(
+          "claude",
+          { refreshToken: "R0", connectionId: "conn-cas-skip" },
+          silentLog,
+          null,
+          onPersist
+        )
+    )
+  );
+
+  assert.equal(persisted, false, "persist MUST be skipped when the row was rotated concurrently");
+  assert.equal(getCasGuardStats().skipped, 1, "the skip must be counted");
+  assert.ok(result?.accessToken, "caller still receives the freshly-issued access token");
+});
+
+test("#4038 CAS guard PERSISTS when the row is unchanged (no concurrent rotation)", async () => {
+  _resetCasGuardStats();
+  _clearTokenRotationMap();
+  let persisted = false;
+  const onPersist = async () => {
+    persisted = true;
+  };
+
+  await withMockedRefresh("NEW_REFRESH_TOKEN", () =>
+    runWithCasGuard(
+      // The row still holds R0 — the exact token we presented — so our persist is safe.
+      { expectedRefreshToken: "R0", reread: async () => "R0" },
+      () =>
+        getAccessToken(
+          "claude",
+          { refreshToken: "R0", connectionId: "conn-cas-pass" },
+          silentLog,
+          null,
+          onPersist
+        )
+    )
+  );
+
+  assert.equal(persisted, true, "persist MUST run when the row still holds the presented token");
+  assert.equal(getCasGuardStats().persisted, 1, "the pass must be counted");
+});
+
+test("#4038 no CAS guard ⇒ persist always runs (opt-in; zero behavior change)", async () => {
+  _resetCasGuardStats();
+  _clearTokenRotationMap();
+  let persisted = false;
+  const onPersist = async () => {
+    persisted = true;
+  };
+
+  await withMockedRefresh("NEW_REFRESH_TOKEN", () =>
+    getAccessToken(
+      "claude",
+      { refreshToken: "R0", connectionId: "conn-no-guard" },
+      silentLog,
+      null,
+      onPersist
+    )
+  );
+
+  assert.equal(persisted, true, "without a guard the persist always runs (unchanged behavior)");
+  assert.equal(getCasGuardStats().skipped, 0, "no guard ⇒ nothing skipped");
+});