diff --git a/packages/core/src/agents/local-providers/grok.ts b/packages/core/src/agents/local-providers/grok.ts index b51822b6..53f903a5 100644 --- a/packages/core/src/agents/local-providers/grok.ts +++ b/packages/core/src/agents/local-providers/grok.ts @@ -47,6 +47,8 @@ const grokDefaultOidcIssuer = "https://auth.x.ai"; const grokOauthDefaultTimeoutMs = 8_000; const grokFallbackClientVersion = "0.2.93"; +const grokRefreshInFlight = new Map>(); + const grokBillingResetPaths = [ "$.billingPeriodEnd", "$.currentPeriod.end", @@ -182,7 +184,7 @@ const grokBillingMapping: ProviderAccountMappingConfig = { ] }; -export class GrokRefreshAuthError extends Error { +class GrokRefreshAuthError extends Error { readonly status: number; constructor(status: number, message: string) { @@ -265,11 +267,25 @@ export async function resolveGrokAuth(): Promise { if (!auth?.refreshToken || (auth.accessToken && !grokAccessTokenExpired(auth))) { return auth; } - try { - return await refreshGrokAuth(auth); - } catch (error) { - return adoptPeerRotatedGrokAuth(auth, error); + // Coalesce concurrent refreshes of the same credential: the refresh token + // rotates on every refresh, so every in-process caller must share a single + // refresh attempt (and its peer-rotation adoption outcome) instead of + // submitting the same stale token in parallel. + const key = `${auth.sourceFile}::${auth.authRecordKey ?? ""}`; + let refresh = grokRefreshInFlight.get(key); + if (!refresh) { + refresh = (async () => { + try { + return await refreshGrokAuth(auth); + } catch (error) { + return adoptPeerRotatedGrokAuth(auth, error); + } + })().finally(() => { + grokRefreshInFlight.delete(key); + }); + grokRefreshInFlight.set(key, refresh); } + return refresh; } function adoptPeerRotatedGrokAuth(auth: GrokTokenSet, error: unknown): GrokTokenSet { diff --git a/packages/core/test/unit/agents/local-agent-provider-grok.test.mjs b/packages/core/test/unit/agents/local-agent-provider-grok.test.mjs index 607f22e1..0a125331 100644 --- a/packages/core/test/unit/agents/local-agent-provider-grok.test.mjs +++ b/packages/core/test/unit/agents/local-agent-provider-grok.test.mjs @@ -403,6 +403,43 @@ test("Grok local provider does not adopt a different account record on refresh 4 }); }); +test("Grok local provider coalesces concurrent refreshes of the same credential", async (t) => { + await withGrokHome(async (grokHome) => { + writeGrokAuth(grokHome, { + key: "expired-token", + refresh_token: "stale-refresh-token", + expires_at: "2000-01-01T00:00:00Z", + oidc_client_id: "grok-client-id", + oidc_issuer: "https://auth.x.ai" + }); + writeGrokModels(grokHome); + + const previousFetch = globalThis.fetch; + process.env.GROK_OIDC_TOKEN_ENDPOINT = "http://127.0.0.1/grok/oauth/token"; + let refreshCalls = 0; + globalThis.fetch = async () => { + refreshCalls += 1; + // Keep the refresh pending so the concurrent callers must share it. + await new Promise((resolve) => setTimeout(resolve, 25)); + return new Response(JSON.stringify({ + access_token: "refreshed-grok-access-token", + expires_in: 3600, + refresh_token: "refreshed-grok-refresh-token" + }), { headers: { "content-type": "application/json" }, status: 200 }); + }; + t.after(() => { + globalThis.fetch = previousFetch; + }); + + const results = await Promise.all([resolveGrokAuth(), resolveGrokAuth(), resolveGrokAuth()]); + assert.equal(refreshCalls, 1); + for (const auth of results) { + assert.equal(auth.accessToken, "refreshed-grok-access-token"); + assert.equal(auth.refreshToken, "refreshed-grok-refresh-token"); + } + }); +}); + async function withGrokHome(run) { const previousGrokHome = process.env.GROK_HOME; const previousGrokAuthFile = process.env.GROK_AUTH_FILE;