fix(account-pool): recover from temporary OAuth refresh outages

plugins/account-pool/src/claude-adapter.ts

@@ -6,6 +6,7 @@ import {
 } from "./credentials.js";
 import type { ProviderAdapter } from "./provider-adapter.js";
 import {
+  fetchOAuthRefresh,
   filterRequestHeaders,
   mountedUpstreamUrl,
 } from "./provider-adapter.js";
@@ -101,22 +102,15 @@ export function createClaudeAdapter(options: {
       ) {
         return { secret, refreshed: false };
       }
-      const response = await context.fetch(options.refreshUrl, {
-        method: "POST",
-        headers: {
-          "content-type": "application/json",
-          accept: "application/json",
-        },
-        body: JSON.stringify({
-          grant_type: "refresh_token",
-          refresh_token: secret.refreshToken,
-          client_id: OAUTH_CLIENT_ID,
-        }),
-      });
-      if (!response.ok) {
-        throw new Error(`OAuth refresh failed with HTTP ${response.status}.`);
-      }
-      const parsed = refreshResponseSchema.parse(await response.json());
+      const parsed = refreshResponseSchema.parse(
+        JSON.parse(
+          await fetchOAuthRefresh(context, options.refreshUrl, {
+            grant_type: "refresh_token",
+            refresh_token: secret.refreshToken,
+            client_id: OAUTH_CLIENT_ID,
+          }),
+        ),
+      );
       const rawExpiresAt =
         parsed.expires_at ??
         (parsed.expires_in === undefined

plugins/account-pool/src/codex-adapter.ts

@@ -12,6 +12,7 @@ import {
 } from "./credentials.js";
 import type { ProviderAdapter } from "./provider-adapter.js";
 import {
+  fetchOAuthRefresh,
   filterRequestHeaders,
   mountedUpstreamUrl,
 } from "./provider-adapter.js";
@@ -296,21 +297,15 @@ export function createCodexAdapter(options: {
       ) {
         return { secret, refreshed: false };
       }
-      const response = await context.fetch(options.refreshUrl, {
-        method: "POST",
-        headers: {
-          "content-type": "application/json",
-          accept: "application/json",
-        },
-        body: JSON.stringify({
-          client_id: CODEX_OAUTH_CLIENT_ID,
-          grant_type: "refresh_token",
-          refresh_token: secret.refreshToken,
-        }),
-      });
-      if (!response.ok)
-        throw new Error(`OAuth refresh failed with HTTP ${response.status}.`);
-      const parsed = refreshResponseSchema.parse(await response.json());
+      const parsed = refreshResponseSchema.parse(
+        JSON.parse(
+          await fetchOAuthRefresh(context, options.refreshUrl, {
+            client_id: CODEX_OAUTH_CLIENT_ID,
+            grant_type: "refresh_token",
+            refresh_token: secret.refreshToken,
+          }),
+        ),
+      );
       const refreshed: AccountSecret = {
         kind: "oauth",
         accessToken: parsed.access_token,

plugins/account-pool/src/hub.ts

@@ -14,6 +14,7 @@ import {
 } from "./codex-adapter.js";
 import type { ProviderAdapter } from "./provider-adapter.js";
 import type { ImportedProviderAccount } from "./provider-adapter.js";
+import { TransientOAuthRefreshError } from "./provider-adapter.js";
 import type {
   ImportedClaudeCredentials,
   ImportedCodexCredentials,
@@ -32,6 +33,8 @@ const DEFAULT_USAGE_URL = "https://api.anthropic.com/api/oauth/usage";
 const DEFAULT_PROFILE_URL = "https://api.anthropic.com/api/oauth/profile";
 const DEFAULT_USAGE_REFRESH_INTERVAL_MS = 5 * 60 * 1_000;
 const MAX_INLINE_HOLD_MS = 20_000;
+const MAX_REFRESH_BACKOFF_MS = 60_000;
+const MAX_REFRESH_BACKOFFS = 1_024;
 const DROPPED_RESPONSE_HEADERS = new Set([
   "content-encoding",
   "content-length",
@@ -74,11 +77,19 @@ interface UpstreamResult {
   release: () => void;
 }
 
+interface RefreshBackoff {
+  accessToken: string;
+  retryAt: number;
+  delayMs: number;
+  error: TransientOAuthRefreshError;
+}
+
 export class AccountPoolHub {
   private accepting = false;
   private readonly inFlightByAccount = new Map<string, number>();
   private readonly activeControllers = new Set<AbortController>();
   private readonly refreshes = new Map<string, Promise<AccountSecret>>();
+  private readonly refreshBackoffs = new Map<string, RefreshBackoff>();
   private readonly usageRefreshes = new Map<string, Promise<void>>();
   private readonly lastUsageRefreshAt = new Map<string, number>();
   private readonly drainWaiters = new Set<() => void>();
@@ -247,14 +258,14 @@ export class AccountPoolHub {
     hostId: string | null,
   ): Promise<Response> {
     const attempted = new Set<string>();
+    let refreshFailure: TransientOAuthRefreshError | null = null;
     const accounts = (await this.options.accounts.list()).filter(
       (account) => account.provider === adapter.provider,
     );
     const family = adapter.modelFamily(body);
     while (attempted.size < accounts.length) {
       const selected = await this.select(adapter.provider, attempted, family);
-      if (selected === null)
-        return this.noEligibleResponse(accounts, family, adapter);
+      if (selected === null) break;
       attempted.add(selected.account.id);
       if (hostId !== null) {
         const changed = await this.options.accounts.recordUsed(
@@ -268,7 +279,11 @@ export class AccountPoolHub {
       try {
         secret = await this.freshSecret(selected.account, adapter);
       } catch (error) {
-        this.markError(selected.account.id, errorMessage(error));
+        if (error instanceof TransientOAuthRefreshError) {
+          refreshFailure = error;
+        } else {
+          this.markError(selected.account.id, errorMessage(error));
+        }
         continue;
       }
       let upstream: UpstreamResult;
@@ -360,7 +375,9 @@ export class AccountPoolHub {
       await this.captureAuthError(upstream.response, selected.account, adapter);
       return this.clientResponse(upstream);
     }
-    return this.noEligibleResponse(accounts, family, adapter);
+    return refreshFailure === null
+      ? this.noEligibleResponse(accounts, family, adapter)
+      : adapter.errorResponse(503, refreshFailure.message);
   }
 
   private async captureAuthError(
@@ -423,22 +440,76 @@ export class AccountPoolHub {
   ): Promise<AccountSecret> {
     const existing = this.refreshes.get(account.id);
     if (existing !== undefined) return existing;
-    const secret = await this.options.accounts.readSecret(account.id);
-    const refresh = adapter
-      .refreshSecret({
-        account,
-        secret,
-        accounts: this.options.accounts,
-        quotas: this.options.quotas,
-        fetch: this.options.fetch,
-        now: this.options.now,
-      })
-      .then((result) => {
-        if (result.refreshed) {
-          const quota = this.options.quotas.get(account.id);
-          this.options.quotas.put({ ...quota, error: null });
+    const refresh = this.options.accounts
+      .readSecret(account.id)
+      .then(async (secret) => {
+        let backoff = this.refreshBackoffs.get(account.id);
+        if (
+          secret.kind !== "oauth" ||
+          backoff?.accessToken !== secret.accessToken
+        ) {
+          this.refreshBackoffs.delete(account.id);
+          backoff = undefined;
+        }
+        if (backoff !== undefined && this.options.now() < backoff.retryAt) {
+          if (
+            secret.kind === "oauth" &&
+            secret.expiresAt !== null &&
+            secret.expiresAt > this.options.now()
+          ) {
+            return secret;
+          }
+          throw backoff.error;
+        }
+        try {
+          const result = await adapter.refreshSecret({
+            account,
+            secret,
+            accounts: this.options.accounts,
+            quotas: this.options.quotas,
+            fetch: this.options.fetch,
+            now: this.options.now,
+          });
+          this.refreshBackoffs.delete(account.id);
+          if (result.refreshed) {
+            const quota = this.options.quotas.get(account.id);
+            this.options.quotas.put({ ...quota, error: null });
+          }
+          return result.secret;
+        } catch (error) {
+          if (
+            !(error instanceof TransientOAuthRefreshError) ||
+            secret.kind !== "oauth"
+          ) {
+            this.refreshBackoffs.delete(account.id);
+            throw error;
+          }
+          const delayMs = Math.min(
+            MAX_REFRESH_BACKOFF_MS,
+            Math.max(
+              backoff === undefined ? 1_000 : backoff.delayMs * 2,
+              error.retryAfterMs,
+            ),
+          );
+          this.refreshBackoffs.delete(account.id);
+          this.refreshBackoffs.set(account.id, {
+            accessToken: secret.accessToken,
+            retryAt: this.options.now() + delayMs,
+            delayMs,
+            error,
+          });
+          while (this.refreshBackoffs.size > MAX_REFRESH_BACKOFFS) {
+            const oldest = this.refreshBackoffs.keys().next();
+            if (!oldest.done) this.refreshBackoffs.delete(oldest.value);
+          }
+          if (
+            secret.expiresAt !== null &&
+            secret.expiresAt > this.options.now()
+          ) {
+            return secret;
+          }
+          throw error;
         }
-        return result.secret;
       })
       .finally(() => this.refreshes.delete(account.id));
     this.refreshes.set(account.id, refresh);

plugins/account-pool/src/provider-adapter.test.ts

@@ -0,0 +1,101 @@
+import { describe, expect, it, vi } from "vitest";
+import {
+  fetchOAuthRefresh,
+  TransientOAuthRefreshError,
+} from "./provider-adapter.js";
+
+describe("OAuth refresh transport", () => {
+  it.each([
+    { status: 400, transient: false },
+    { status: 401, transient: false },
+    { status: 408, transient: true },
+    { status: 429, transient: true },
+    { status: 500, transient: true },
+    { status: 503, transient: true },
+  ])(
+    "cancels HTTP $status responses before classifying them",
+    async ({ status, transient }) => {
+      const cancel = vi.fn();
+      const response = new Response(new ReadableStream({ cancel }), {
+        status,
+        headers: { "retry-after": "120" },
+      });
+      const refresh = fetchOAuthRefresh(
+        { fetch: async () => response, now: () => 1_800_000_000_000 },
+        "https://auth.example/token",
+        { refresh_token: "refresh" },
+      );
+      await expect(refresh).rejects.toThrow(
+        `OAuth refresh failed with HTTP ${status}.`,
+      );
+      if (transient) {
+        await expect(refresh).rejects.toBeInstanceOf(
+          TransientOAuthRefreshError,
+        );
+        await expect(refresh).rejects.toMatchObject({ retryAfterMs: 120_000 });
+      } else {
+        await expect(refresh).rejects.not.toBeInstanceOf(
+          TransientOAuthRefreshError,
+        );
+      }
+      expect(cancel).toHaveBeenCalledTimes(1);
+    },
+  );
+
+  it.each(["request", "response"])(
+    "classifies a broken %s connection as transient",
+    async (stage) => {
+      const refresh = fetchOAuthRefresh(
+        {
+          fetch: async () => {
+            if (stage === "request") throw new TypeError("fetch failed");
+            return new Response(
+              new ReadableStream({
+                start(controller) {
+                  controller.error(new TypeError("socket closed"));
+                },
+              }),
+            );
+          },
+          now: () => 1_800_000_000_000,
+        },
+        "https://auth.example/token",
+        { refresh_token: "refresh" },
+      );
+      await expect(refresh).rejects.toBeInstanceOf(TransientOAuthRefreshError);
+    },
+  );
+
+  it("times out a pending OAuth request independently of its callers", async () => {
+    const controller = new AbortController();
+    const timeout = vi
+      .spyOn(AbortSignal, "timeout")
+      .mockReturnValue(controller.signal);
+    try {
+      const refresh = fetchOAuthRefresh(
+        {
+          fetch: async (_input, init) => {
+            const signal = init?.signal;
+            if (signal === null || signal === undefined)
+              return Response.json({});
+            return new Promise<Response>((_resolve, reject) => {
+              signal.addEventListener("abort", () => reject(signal.reason), {
+                once: true,
+              });
+            });
+          },
+          now: () => 1_800_000_000_000,
+        },
+        "https://auth.example/token",
+        { refresh_token: "refresh" },
+      );
+      controller.abort(
+        new DOMException("OAuth request timed out", "TimeoutError"),
+      );
+      await expect(refresh).rejects.toBeInstanceOf(TransientOAuthRefreshError);
+      expect(timeout).toHaveBeenCalledWith(15_000);
+    } finally {
+      timeout.mockRestore();
+    }
+  });
+});