fix(codex): serialize shared-credential OAuth token refreshes

Concurrent headless runs sharing one Codex credential raced refresh-token
rotation: overlapping grants revoked each other and the loser surfaced a
bare not-found rejection. Refresh-and-persist now serializes on a lock file
next to the home-level auth store (exclusive create + in-memory tail for
same-process callers, stale takeover for crashed holders), so parallel
agents observe the persisted result instead of racing.

Refresh failures (missing profile, rejected refresh, lock timeout) throw
CodexAuthError with a re-login hint and project onto the shared
credential_failure shape, so the run surfaces a credential failure, never
a bare not-found error.

Author: TheGreatAxios

src/auth/codex/refresh-lock.test.ts

@@ -0,0 +1,144 @@
+import { mkdtemp, rm, stat, utimes, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { describe, expect, test } from "bun:test";
+import {
+  CodexRefreshLockTimeoutError,
+  withCodexRefreshLock,
+} from "./refresh-lock.js";
+
+async function tempDir(): Promise<string> {
+  return mkdtemp(join(tmpdir(), "cl8628-lock-"));
+}
+
+describe("codex refresh lock", () => {
+  test("concurrent holders serialize and the lock file is removed", async () => {
+    const dir = await tempDir();
+    try {
+      const lock = join(dir, "refresh.lock");
+      let active = 0;
+      let maxActive = 0;
+      const results = await Promise.all(
+        [0, 1, 2, 3, 4].map((i) =>
+          withCodexRefreshLock(lock, async () => {
+            active += 1;
+            maxActive = Math.max(maxActive, active);
+            // The file exists while held so a second process contends on it.
+            await stat(lock);
+            await new Promise((resolve) => setTimeout(resolve, 20));
+            active -= 1;
+            return i;
+          }),
+        ),
+      );
+      expect(results).toEqual([0, 1, 2, 3, 4]);
+      expect(maxActive).toBe(1);
+      await expect(stat(lock)).rejects.toThrow();
+    } finally {
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a foreign-held lock times out with a recovery hint", async () => {
+    const dir = await tempDir();
+    try {
+      const lock = join(dir, "refresh.lock");
+      // Simulate a lock held by another process: withCodexRefreshLock never
+      // created it, so only the file path (not the in-memory chain) applies.
+      await writeFile(lock, "");
+      let failure: unknown;
+      try {
+        await withCodexRefreshLock(lock, async () => "never", {
+          timeoutMs: 100,
+          retryMs: 10,
+        });
+      } catch (err) {
+        failure = err;
+      }
+      expect(failure).toBeInstanceOf(CodexRefreshLockTimeoutError);
+      expect((failure as CodexRefreshLockTimeoutError).lockPath).toBe(lock);
+      expect((failure as CodexRefreshLockTimeoutError).message).toContain(
+        "remove this lock file manually and retry",
+      );
+    } finally {
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a stale lock from a crashed holder is taken over", async () => {
+    const dir = await tempDir();
+    try {
+      const lock = join(dir, "refresh.lock");
+      await writeFile(lock, "");
+      const ancient = new Date(Date.now() - 60_000);
+      await utimes(lock, ancient, ancient);
+      const result = await withCodexRefreshLock(
+        lock,
+        async () => "taken-over",
+        {
+          timeoutMs: 5_000,
+          staleMs: 1_000,
+        },
+      );
+      expect(result).toBe("taken-over");
+      await expect(stat(lock)).rejects.toThrow();
+    } finally {
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a lock held by another process blocks acquisition until released", async () => {
+    const dir = await tempDir();
+    const holderPath = new URL(
+      "../../../tests/fixtures/codex-refresh-lock/hold-lock.ts",
+      import.meta.url,
+    ).pathname;
+    const lock = join(dir, "refresh.lock");
+    const proc = Bun.spawn(["bun", "run", holderPath, lock, "1500"], {
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    try {
+      // Wait until the holder process actually holds the file lock.
+      if (proc.stdout === null) throw new Error("holder has no stdout pipe");
+      const reader = proc.stdout.getReader();
+      const decoder = new TextDecoder();
+      let output = "";
+      const deadline = Date.now() + 10_000;
+      try {
+        while (!output.includes("held")) {
+          if (Date.now() > deadline)
+            throw new Error("lock holder never acquired the lock");
+          const { value, done } = await reader.read();
+          if (done) break;
+          output += decoder.decode(value, { stream: true });
+        }
+      } finally {
+        reader.releaseLock();
+      }
+      expect(output).toContain("held");
+
+      // While the other process holds it, acquisition times out instead of
+      // overlapping the grant.
+      let failure: unknown;
+      try {
+        await withCodexRefreshLock(lock, async () => "never", {
+          timeoutMs: 300,
+          retryMs: 10,
+        });
+      } catch (err) {
+        failure = err;
+      }
+      expect(failure).toBeInstanceOf(CodexRefreshLockTimeoutError);
+
+      // Once the holder exits and releases, the lock is acquirable again.
+      await proc.exited;
+      expect(await withCodexRefreshLock(lock, async () => "acquired")).toBe(
+        "acquired",
+      );
+    } finally {
+      proc.kill();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+});

src/auth/codex/refresh-lock.ts

@@ -0,0 +1,114 @@
+import { mkdir, open, stat, unlink } from "node:fs/promises";
+import { dirname } from "node:path";
+
+// Serializes Codex OAuth refresh-and-persist sections across processes that
+// share one credential store (concurrent headless runs). Same-process
+// callers chain on an in-memory tail; cross-process callers contend on an
+// exclusive lock file. At most one refresh grant is ever in flight, so a
+// second refresher observes the persisted result instead of racing the
+// authorization server's refresh-token rotation and revoking its sibling.
+const tails = new Map<string, Promise<void>>();
+
+// Default stale horizon: crashed-holder recovery only. Kept far above any
+// legitimate hold and any contender timeout so a live holder is never
+// declared stale mid-wait (see the staleMs field docs).
+const DEFAULT_STALE_MS = 120_000;
+
+export interface CodexRefreshLockOptions {
+  /** Maximum wait for the lock before giving up. Defaults to 30_000. */
+  timeoutMs?: number;
+  /** Poll interval while contending. Defaults to 25. */
+  retryMs?: number;
+  /**
+   * Lock-file age at which the holder is presumed crashed and the lock is
+   * taken over. Defaults to 120_000. This must stay far above the longest
+   * legitimate hold (the token endpoint timeout plus store I/O) — and, in
+   * particular, far above any contender's own timeout — otherwise a waiter
+   * declares the live holder stale mid-wait, steals the lock, and the two
+   * overlapping refresh grants revoke each other under token rotation.
+   */
+  staleMs?: number;
+}
+
+// The lock could not be acquired in time: either a refresh is genuinely
+// stalled past the timeout, or a crashed holder's lock survived takeover.
+// Carries the lock path so the message can tell the operator how to recover.
+export class CodexRefreshLockTimeoutError extends Error {
+  readonly lockPath: string;
+
+  constructor(lockPath: string, timeoutMs: number) {
+    super(
+      `Timed out after ${String(timeoutMs)}ms waiting for the Codex refresh lock ` +
+        `at ${lockPath}. If no refresh is running, remove this lock file manually and retry.`,
+    );
+    this.name = "CodexRefreshLockTimeoutError";
+    this.lockPath = lockPath;
+  }
+}
+
+async function acquireLockFile(
+  lockPath: string,
+  timeoutMs: number,
+  retryMs: number,
+  staleMs: number,
+): Promise<void> {
+  await mkdir(dirname(lockPath), { recursive: true, mode: 0o700 });
+  const start = Date.now();
+  for (;;) {
+    try {
+      const handle = await open(lockPath, "wx", 0o600);
+      await handle.close();
+      return;
+    } catch (err) {
+      if ((err as NodeJS.ErrnoException)?.code !== "EEXIST") throw err;
+    }
+    try {
+      // A crashed holder never releases: once its lock is older than the
+      // stale horizon, unlink and take over rather than brick refreshes.
+      const info = await stat(lockPath);
+      if (Date.now() - info.mtimeMs > staleMs) {
+        await unlink(lockPath).catch(() => undefined);
+        continue;
+      }
+    } catch {
+      // The lock vanished (or was never stat-able) between attempts: loop
+      // around and contend for it again.
+    }
+    if (Date.now() - start >= timeoutMs)
+      throw new CodexRefreshLockTimeoutError(lockPath, timeoutMs);
+    await new Promise((resolve) => setTimeout(resolve, retryMs));
+  }
+}
+
+/**
+ * Runs `work` while holding the refresh lock at `lockPath`. Concurrent
+ * same-process callers queue in memory; concurrent processes queue on the
+ * lock file. The lock file is always removed afterwards.
+ */
+export async function withCodexRefreshLock<T>(
+  lockPath: string,
+  work: () => Promise<T>,
+  options: CodexRefreshLockOptions = {},
+): Promise<T> {
+  const timeoutMs = options.timeoutMs ?? 30_000;
+  const retryMs = options.retryMs ?? 25;
+  const staleMs = options.staleMs ?? DEFAULT_STALE_MS;
+  const previous = tails.get(lockPath) ?? Promise.resolve();
+  let releaseTail!: () => void;
+  const current = new Promise<void>((resolve) => {
+    releaseTail = resolve;
+  });
+  tails.set(lockPath, current);
+  try {
+    await previous.catch(() => undefined);
+    await acquireLockFile(lockPath, timeoutMs, retryMs, staleMs);
+    try {
+      return await work();
+    } finally {
+      await unlink(lockPath).catch(() => undefined);
+    }
+  } finally {
+    if (tails.get(lockPath) === current) tails.delete(lockPath);
+    releaseTail();
+  }
+}

src/auth/codex/session-failure.test.ts

@@ -0,0 +1,117 @@
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { describe, expect, test } from "bun:test";
+import { saveCodexProfile } from "../../config/oauth-stores.js";
+import {
+  codexAuthFailureDiagnostic,
+  CodexAuthError,
+  getValidCodexToken,
+} from "./session.js";
+
+async function tempHome(): Promise<string> {
+  return mkdtemp(join(tmpdir(), "cl8628-failure-"));
+}
+
+describe("codex auth failure surface", () => {
+  test("a missing profile rejects with a re-login hint, never a bare not-found", async () => {
+    const home = await tempHome();
+    try {
+      let failure: unknown;
+      try {
+        await getValidCodexToken("ghost", Date.now(), home);
+      } catch (err) {
+        failure = err;
+      }
+      expect(failure).toBeInstanceOf(CodexAuthError);
+      const auth = failure as CodexAuthError;
+      expect(auth.reason).toBe("missing");
+      expect(auth.message).toContain('"ghost"');
+      expect(auth.message).toContain("Log in again");
+      expect(auth.message).not.toContain("No OAuth profile named");
+    } finally {
+      await rm(home, { recursive: true, force: true });
+    }
+  });
+
+  test("a rejected refresh rejects with a re-login hint", async () => {
+    const home = await tempHome();
+    const now = Date.now();
+    await saveCodexProfile(
+      {
+        name: "shared",
+        createdAt: now,
+        tokens: {
+          access: "access-1",
+          refresh: "refresh-1",
+          expiresAt: now - 300_000,
+          accountId: "acct-1",
+        },
+      },
+      home,
+    );
+    const originalFetch = globalThis.fetch;
+    globalThis.fetch = (async () =>
+      new Response(JSON.stringify({ error: "invalid_grant" }), {
+        status: 400,
+      })) as unknown as typeof fetch;
+    try {
+      let failure: unknown;
+      try {
+        await getValidCodexToken("shared", now, home);
+      } catch (err) {
+        failure = err;
+      }
+      expect(failure).toBeInstanceOf(CodexAuthError);
+      expect((failure as CodexAuthError).reason).toBe("refresh-failed");
+      expect((failure as CodexAuthError).message).toContain("Log in again");
+    } finally {
+      globalThis.fetch = originalFetch;
+      await rm(home, { recursive: true, force: true });
+    }
+  });
+
+  test("fresh tokens resolve without touching the network", async () => {
+    const home = await tempHome();
+    const now = Date.now();
+    await saveCodexProfile(
+      {
+        name: "shared",
+        createdAt: now,
+        tokens: {
+          access: "access-1",
+          refresh: "refresh-1",
+          expiresAt: now + 3_600_000,
+          accountId: "acct-1",
+        },
+      },
+      home,
+    );
+    const originalFetch = globalThis.fetch;
+    globalThis.fetch = (async () => {
+      throw new Error("network must not be touched for fresh tokens");
+    }) as unknown as typeof fetch;
+    try {
+      const access = await getValidCodexToken("shared", now, home);
+      expect(access.access).toBe("access-1");
+      expect(access.accountId).toBe("acct-1");
+    } finally {
+      globalThis.fetch = originalFetch;
+      await rm(home, { recursive: true, force: true });
+    }
+  });
+
+  test("codexAuthFailureDiagnostic projects auth errors onto credential_failure", () => {
+    const auth = new CodexAuthError(
+      "personal",
+      "refresh-failed",
+      'Codex profile "personal" could not be refreshed (boom). Log in again.',
+    );
+    expect(codexAuthFailureDiagnostic(auth)).toEqual({
+      category: "credential_failure",
+      message: auth.message,
+    });
+    expect(codexAuthFailureDiagnostic(new Error("boom"))).toBeNull();
+    expect(codexAuthFailureDiagnostic("boom")).toBeNull();
+  });
+});