diff --git a/src/agents/auth-profiles/oauth-manager.test.ts b/src/agents/auth-profiles/oauth-manager.test.ts index af367dacf67e..f1d1a8eaca55 100644 --- a/src/agents/auth-profiles/oauth-manager.test.ts +++ b/src/agents/auth-profiles/oauth-manager.test.ts @@ -561,6 +561,153 @@ describe("createOAuthManager", () => { }); }); + it("force-persists a refreshed credential after a same-identity CAS race", async () => { + await withOAuthTempRoot("oauth-manager-cas-same-identity-", async (tempRoot) => { + const agentDir = path.join(tempRoot, "agents", "main", "agent"); + await fs.mkdir(agentDir, { recursive: true }); + const profileId = "openai:oauth"; + const expired = createCredential({ + access: "expired-access", + refresh: "expired-refresh", + expires: Date.now() - 60_000, + accountId: "acct-123", + }); + saveAuthProfileStore( + { + version: 1, + profiles: { + [profileId]: expired, + }, + }, + agentDir, + { filterExternalAuthProfiles: false }, + ); + + const manager = createOAuthManager({ + buildApiKey: async (_provider, credential) => credential.access, + refreshCredential: vi.fn(async () => { + saveAuthProfileStore( + { + version: 1, + profiles: { + [profileId]: createCredential({ + access: "stale-race-access", + refresh: "consumed-race-refresh", + expires: Date.now() + 10 * 60_000, + accountId: "acct-123", + }), + }, + }, + agentDir, + { filterExternalAuthProfiles: false }, + ); + return { + access: "rotated-access", + refresh: "rotated-refresh", + expires: Date.now() + 60_000, + }; + }), + readBootstrapCredential: () => null, + isRefreshTokenReusedError: () => false, + }); + + const result = await manager.resolveOAuthAccess({ + store: ensureAuthProfileStoreWithoutExternalProfiles(agentDir, { + allowKeychainPrompt: false, + }), + profileId, + credential: expired, + agentDir, + }); + + expect(result?.apiKey).toBe("rotated-access"); + const persisted = ensureAuthProfileStoreWithoutExternalProfiles(agentDir, { + allowKeychainPrompt: false, + }); + expect(persisted.profiles[profileId]).toMatchObject({ + type: "oauth", + access: "rotated-access", + refresh: "rotated-refresh", + accountId: "acct-123", + }); + }); + }); + + it("uses a different-identity stored credential after a CAS race", async () => { + await withOAuthTempRoot("oauth-manager-cas-different-identity-", async (tempRoot) => { + const mainAgentDir = path.join(tempRoot, "agents", "main", "agent"); + const agentDir = path.join(tempRoot, "agents", "sub", "agent"); + await fs.mkdir(mainAgentDir, { recursive: true }); + await fs.mkdir(agentDir, { recursive: true }); + const profileId = "openai:oauth"; + const expired = createCredential({ + access: "expired-access", + refresh: "expired-refresh", + expires: Date.now() - 60_000, + accountId: "acct-123", + }); + const relogged = createCredential({ + access: "relogged-access", + refresh: "relogged-refresh", + expires: Date.now() + 10 * 60_000, + accountId: "acct-456", + }); + saveAuthProfileStore( + { + version: 1, + profiles: { + [profileId]: expired, + }, + }, + agentDir, + { filterExternalAuthProfiles: false }, + ); + + const manager = createOAuthManager({ + buildApiKey: async (_provider, credential) => credential.access, + refreshCredential: vi.fn(async () => { + saveAuthProfileStore( + { + version: 1, + profiles: { + [profileId]: relogged, + }, + }, + agentDir, + { filterExternalAuthProfiles: false }, + ); + return { + access: "rotated-access", + refresh: "rotated-refresh", + expires: Date.now() + 60_000, + }; + }), + readBootstrapCredential: () => null, + isRefreshTokenReusedError: () => false, + }); + + const result = await manager.resolveOAuthAccess({ + store: ensureAuthProfileStoreWithoutExternalProfiles(agentDir, { + allowKeychainPrompt: false, + }), + profileId, + credential: expired, + agentDir, + }); + + expect(result?.apiKey).toBe("relogged-access"); + const persisted = ensureAuthProfileStoreWithoutExternalProfiles(agentDir, { + allowKeychainPrompt: false, + }); + expect(persisted.profiles[profileId]).toMatchObject({ + type: "oauth", + access: "relogged-access", + refresh: "relogged-refresh", + accountId: "acct-456", + }); + }); + }); + it("fails closed after managed refresh failure", async () => { await withOAuthAgentDirs("oauth-manager-refresh-fail-closed-", async ({ agentDir }) => { const profileId = "openai:user@example.com"; diff --git a/src/agents/auth-profiles/oauth-manager.ts b/src/agents/auth-profiles/oauth-manager.ts index bd66269c9fcb..c90152ef32a8 100644 --- a/src/agents/auth-profiles/oauth-manager.ts +++ b/src/agents/auth-profiles/oauth-manager.ts @@ -19,6 +19,7 @@ import { } from "./oauth-refresh-lock-errors.js"; import { areOAuthCredentialsEquivalent, + hasMatchingOAuthIdentity, hasUsableOAuthCredential, isSafeToAdoptBootstrapOAuthIdentity, isSafeToAdoptMainStoreOAuthIdentity, @@ -466,6 +467,41 @@ export function createOAuthManager(adapter: OAuthManagerAdapter) { return result !== null && saved; } + async function resolveOAuthCredentialAfterPersistMiss(params: { + agentDir?: string; + profileId: string; + refreshed: OAuthCredential; + }): Promise { + const currentStore = loadStoredOAuthRefreshStore(params.agentDir); + const current = currentStore.profiles[params.profileId]; + if (current?.type !== "oauth" || current.provider !== params.refreshed.provider) { + return null; + } + if (!hasMatchingOAuthIdentity(current, params.refreshed)) { + return hasUsableOAuthCredential(current) ? current : null; + } + + let saved = false; + const result = await updateAuthProfileStoreWithLock({ + agentDir: params.agentDir, + updater: (store) => { + const existing = store.profiles[params.profileId]; + if (existing?.type !== "oauth" || existing.provider !== params.refreshed.provider) { + return false; + } + // Refresh tokens rotate server-side before persist. Same-identity CAS + // losers must win the store or the token family is bricked. + if (hasMatchingOAuthIdentity(existing, params.refreshed)) { + store.profiles[params.profileId] = { ...params.refreshed }; + saved = true; + return true; + } + return false; + }, + }); + return result !== null && saved ? params.refreshed : null; + } + async function doRefreshOAuthTokenWithLock(params: { profileId: string; provider: string; @@ -617,7 +653,23 @@ export function createOAuthManager(adapter: OAuthManagerAdapter) { credential: refreshedCredentials, }); if (!persisted) { - throw new Error("Failed to persist refreshed OAuth credential"); + const recovered = await resolveOAuthCredentialAfterPersistMiss({ + agentDir: ownerAgentDir, + profileId: params.profileId, + refreshed: refreshedCredentials, + }); + if (!recovered) { + throw new Error("Failed to persist refreshed OAuth credential"); + } + if (recovered !== refreshedCredentials) { + return { + apiKey: await adapter.buildApiKey(recovered.provider, recovered, { + cfg: params.cfg, + agentDir: params.agentDir, + }), + credential: recovered, + }; + } } if (ownerAgentDir) { const mainPath = resolveAuthStorePath(undefined); diff --git a/src/agents/auth-profiles/profiles.ts b/src/agents/auth-profiles/profiles.ts index b6bb3acaaaa3..b378a86e8b0d 100644 --- a/src/agents/auth-profiles/profiles.ts +++ b/src/agents/auth-profiles/profiles.ts @@ -8,6 +8,7 @@ import { normalizeProviderId, } from "@openclaw/model-catalog-core/provider-id"; import { normalizeStringEntries } from "@openclaw/normalization-core/string-normalization"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; import { resolveProviderIdForAuth } from "../provider-auth-aliases.js"; import { normalizeAuthProfileCredential } from "./credential-normalize.js"; import { dedupeProfileIds, listProfilesForProvider } from "./profile-list.js"; @@ -23,6 +24,8 @@ export { resolveSubscriptionAuthModeForProfiles, } from "./profile-list.js"; +const authProfileProfilesLog = createSubsystemLogger("agent/embedded"); + // Auth profile order/lastGood keys may be stored as aliases. Resolve through // auth provider normalization before updating per-provider state. function findProviderAuthStateKey( @@ -286,11 +289,15 @@ export async function markAuthProfileSuccess(params: { store.usageStats = updated.usageStats; return; } - const profile = store.profiles[profileId]; - if (!profile || resolveProviderIdForAuth(profile.provider) !== providerKey) { - return; + if (updated === null) { + authProfileProfilesLog.warn( + "dropped auth profile bookkeeping after locked store update failed", + { + event: "auth_profile_bookkeeping_dropped", + kind: "success", + profileId, + tags: ["auth_profiles", "persistence"], + }, + ); } - store.lastGood = { ...store.lastGood, [providerKey]: profileId }; - updateSuccessfulUsageStatsEntry(store, profileId, lastUsed); - saveAuthProfileStore(store, agentDir); } diff --git a/src/agents/auth-profiles/usage.test.ts b/src/agents/auth-profiles/usage.test.ts index 4abe7358058c..aeeff74b5b9e 100644 --- a/src/agents/auth-profiles/usage.test.ts +++ b/src/agents/auth-profiles/usage.test.ts @@ -31,12 +31,12 @@ vi.mock("./store.js", () => ({ })); beforeEach(() => { - vi.clearAllMocks(); + storeMocks.saveAuthProfileStore.mockReset(); + storeMocks.updateAuthProfileStoreWithLock.mockReset(); fetchMock.mockReset(); vi.stubGlobal("fetch", fetchMock); - storeMocks.updateAuthProfileStoreWithLock.mockResolvedValue(null); + storeMocks.updateAuthProfileStoreWithLock.mockResolvedValue({ version: 1, profiles: {} }); authProfileUsageTesting.setDepsForTest({ - saveAuthProfileStore: storeMocks.saveAuthProfileStore, updateAuthProfileStoreWithLock: storeMocks.updateAuthProfileStoreWithLock, }); }); @@ -62,6 +62,16 @@ function makeStore(usageStats: AuthProfileStore["usageStats"]): AuthProfileStore }; } +function mockLockedUpdateForStore(store: AuthProfileStore): void { + storeMocks.updateAuthProfileStoreWithLock.mockImplementationOnce( + async (lockParams: { updater: (store: AuthProfileStore) => boolean }) => { + const freshStore = structuredClone(store); + lockParams.updater(freshStore); + return freshStore; + }, + ); +} + function expectProfileErrorStateCleared( stats: NonNullable[string] | undefined, ) { @@ -671,6 +681,7 @@ describe("clearAuthProfileCooldown", () => { failureCounts: { billing: 3, rate_limit: 2 }, }, }); + mockLockedUpdateForStore(store); await clearAuthProfileCooldown({ store, profileId: "anthropic:default" }); @@ -689,6 +700,7 @@ describe("clearAuthProfileCooldown", () => { lastFailureAt, }, }); + mockLockedUpdateForStore(store); await clearAuthProfileCooldown({ store, profileId: "anthropic:default" }); @@ -699,6 +711,7 @@ describe("clearAuthProfileCooldown", () => { it("no-ops for unknown profile id", async () => { const store = makeStore(undefined); + mockLockedUpdateForStore(store); await clearAuthProfileCooldown({ store, profileId: "nonexistent" }); expect(store.usageStats).toBeUndefined(); }); @@ -717,6 +730,7 @@ describe("markAuthProfileFailure — active windows do not extend on retry", () cfg?: OpenClawConfig; }): Promise { const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(params.now); + mockLockedUpdateForStore(params.store); try { await markAuthProfileFailure({ store: params.store, @@ -905,6 +919,7 @@ describe("markAuthProfileBlockedUntil", () => { blockedUntil: laterBlockedUntil, }, }); + mockLockedUpdateForStore(store); try { await markAuthProfileBlockedUntil({ store, @@ -922,6 +937,7 @@ describe("markAuthProfileBlockedUntil", () => { it("ignores blocked-until updates when the process clock is invalid", async () => { const nowSpy = vi.spyOn(Date, "now").mockReturnValue(Number.NaN); const store = makeStore({}); + mockLockedUpdateForStore(store); try { await markAuthProfileBlockedUntil({ store, @@ -939,6 +955,7 @@ describe("markAuthProfileBlockedUntil", () => { it("ignores blocked-until updates outside the valid Date range", async () => { const store = makeStore({}); + mockLockedUpdateForStore(store); await markAuthProfileBlockedUntil({ store, @@ -976,6 +993,46 @@ describe("markAuthProfileFailure — detail-less provider failures", () => { }); }); +describe("markAuthProfileFailure — locked update failure", () => { + it("drops bookkeeping without an unlocked full-store save", async () => { + const store = makeStore(undefined); + const consoleWarn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const previousTestConsole = process.env.OPENCLAW_TEST_CONSOLE; + const previousLogLevel = process.env.OPENCLAW_LOG_LEVEL; + storeMocks.updateAuthProfileStoreWithLock.mockResolvedValueOnce(null); + process.env.OPENCLAW_TEST_CONSOLE = "1"; + process.env.OPENCLAW_LOG_LEVEL = "warn"; + try { + await markAuthProfileFailure({ + store, + profileId: "anthropic:default", + reason: "rate_limit", + }); + expect(store.usageStats).toBeUndefined(); + expect(storeMocks.saveAuthProfileStore).not.toHaveBeenCalled(); + expect( + consoleWarn.mock.calls.some(([line]) => + String(line).includes( + "dropped auth profile bookkeeping after locked store update failed", + ), + ), + ).toBe(true); + } finally { + if (previousTestConsole === undefined) { + delete process.env.OPENCLAW_TEST_CONSOLE; + } else { + process.env.OPENCLAW_TEST_CONSOLE = previousTestConsole; + } + if (previousLogLevel === undefined) { + delete process.env.OPENCLAW_LOG_LEVEL; + } else { + process.env.OPENCLAW_LOG_LEVEL = previousLogLevel; + } + consoleWarn.mockRestore(); + } + }); +}); + describe("markAuthProfileFailure — WHAM-aware Codex cooldowns", () => { function mockWhamResponse(status: number, body?: unknown): void { fetchMock.mockResolvedValueOnce( @@ -990,17 +1047,11 @@ describe("markAuthProfileFailure — WHAM-aware Codex cooldowns", () => { store: ReturnType; now: number; reason?: "rate_limit" | "no_error_details" | "unknown"; - useLock?: boolean; + mockLock?: boolean; }): Promise { const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(params.now); - if (params.useLock) { - storeMocks.updateAuthProfileStoreWithLock.mockImplementationOnce( - async (lockParams: { updater: (store: AuthProfileStore) => boolean }) => { - const freshStore = structuredClone(params.store); - const changed = lockParams.updater(freshStore); - return changed ? freshStore : null; - }, - ); + if (params.mockLock !== false) { + mockLockedUpdateForStore(params.store); } try { await markAuthProfileFailure({ @@ -1124,7 +1175,7 @@ describe("markAuthProfileFailure — WHAM-aware Codex cooldowns", () => { }, ); - await markCodexFailureAt({ store, now, reason: "no_error_details" }); + await markCodexFailureAt({ store, now, reason: "no_error_details", mockLock: false }); expect(fetchMock).toHaveBeenCalledTimes(1); expect(store.usageStats).toBeUndefined(); @@ -1192,7 +1243,7 @@ describe("markAuthProfileFailure — WHAM-aware Codex cooldowns", () => { }, }); - await markCodexFailureAt({ store, now, useLock: true }); + await markCodexFailureAt({ store, now }); expect(store.usageStats?.["openai:default"]?.cooldownUntil).toBe(existingCooldownUntil); }); @@ -1244,6 +1295,7 @@ describe("markAuthProfileFailure — WHAM-aware Codex cooldowns", () => { vi.useFakeTimers(); vi.setSystemTime(now); try { + mockLockedUpdateForStore(store); await markAuthProfileFailure({ store, profileId: "anthropic:default", @@ -1276,6 +1328,7 @@ describe("markAuthProfileFailure — per-model cooldown metadata", () => { }): Promise { vi.useFakeTimers(); vi.setSystemTime(params.now); + mockLockedUpdateForStore(params.store); try { await markAuthProfileFailure({ store: params.store, @@ -1360,6 +1413,7 @@ describe("markAuthProfileFailure — per-model cooldown metadata", () => { lastFailureAt: now - 1000, }, }); + mockLockedUpdateForStore(store); await markAuthProfileFailure({ store, profileId: "github-copilot:github", @@ -1384,6 +1438,7 @@ describe("markAuthProfileFailure — per-model cooldown metadata", () => { lastFailureAt: now - 1000, }, }); + mockLockedUpdateForStore(store); await markAuthProfileFailure({ store, profileId: "github-copilot:github", diff --git a/src/agents/auth-profiles/usage.ts b/src/agents/auth-profiles/usage.ts index bc510e5f835c..39187762bae0 100644 --- a/src/agents/auth-profiles/usage.ts +++ b/src/agents/auth-profiles/usage.ts @@ -19,7 +19,7 @@ import { notifyAuthProfileFailureHook, setAuthProfileFailureHook } from "./failu import { logAuthProfileFailureStateChange } from "./state-observation.js"; const authProfileUsageLog = createSubsystemLogger("agent/embedded"); -import { saveAuthProfileStore, updateAuthProfileStoreWithLock } from "./store.js"; +import { updateAuthProfileStoreWithLock } from "./store.js"; import type { AuthProfileBlockedSource, AuthProfileCredential, @@ -41,7 +41,6 @@ export { } from "./usage-state.js"; const authProfileUsageDeps = { - saveAuthProfileStore, updateAuthProfileStoreWithLock, }; @@ -51,17 +50,23 @@ export { setAuthProfileFailureHook }; export const testing = { setDepsForTest( overrides: Partial<{ - saveAuthProfileStore: typeof saveAuthProfileStore; updateAuthProfileStoreWithLock: typeof updateAuthProfileStoreWithLock; }> | null, ) { - authProfileUsageDeps.saveAuthProfileStore = - overrides?.saveAuthProfileStore ?? saveAuthProfileStore; authProfileUsageDeps.updateAuthProfileStoreWithLock = overrides?.updateAuthProfileStoreWithLock ?? updateAuthProfileStoreWithLock; }, }; +function logDroppedAuthProfileBookkeeping(kind: string, profileId: string): void { + authProfileUsageLog.warn("dropped auth profile bookkeeping after locked store update failed", { + event: "auth_profile_bookkeeping_dropped", + kind, + profileId, + tags: ["auth_profiles", "persistence"], + }); +} + const FAILURE_REASON_PRIORITY: AuthProfileFailureReason[] = [ "auth_permanent", "auth", @@ -815,59 +820,8 @@ export async function markAuthProfileFailure(params: { } return; } - if (!store.profiles[profileId]) { - return; - } - - const currentWhamResult = - whamResult && isSameWhamCredential(profile, store.profiles[profileId]) ? whamResult : null; - if (reason === "no_error_details" && !currentWhamResult) { - return; - } - - const now = Date.now(); - const providerKey = normalizeProviderId(store.profiles[profileId]?.provider ?? ""); - const cfgResolved = resolveAuthCooldownConfig({ - cfg, - providerId: providerKey, - }); - - previousStats = store.usageStats?.[profileId]; - const computed = computeNextProfileUsageStats({ - existing: previousStats ?? {}, - now, - reason, - cfgResolved, - modelId, - }); - nextStats = currentWhamResult - ? applyWhamCooldownResult({ - existing: previousStats ?? {}, - computed, - now, - whamResult: currentWhamResult, - }) - : computed; - updateUsageStatsEntry(store, profileId, () => nextStats ?? computed); - authProfileUsageDeps.saveAuthProfileStore(store, agentDir); - logAuthProfileFailureStateChange({ - runId, - profileId, - provider: store.profiles[profileId]?.provider ?? profile.provider, - reason, - previous: previousStats, - next: nextStats, - now, - }); - try { - notifyAuthProfileFailureHook(); - } catch (err) { - // Hook errors must not break failure recording; log and continue. - authProfileUsageLog.warn("auth profile failure hook threw", { - event: "auth_profile_failure_hook_error", - tags: ["error_handling", "auth_profiles"], - error: err instanceof Error ? err.message : String(err), - }); + if (updated === null) { + logDroppedAuthProfileBookkeeping("failure", profileId); } } @@ -961,33 +915,9 @@ export async function markAuthProfileBlockedUntil(params: { } return; } - if (!store.profiles[profileId]) { - return; + if (updated === null) { + logDroppedAuthProfileBookkeeping("blocked_until", profileId); } - - const now = asDateTimestampMs(Date.now()); - if (now === undefined) { - return; - } - previousStats = store.usageStats?.[profileId]; - nextStats = buildBlockedProfileUsageStats({ - previousStats, - blockedUntil, - source, - modelId, - now, - }); - updateUsageStatsEntry(store, profileId, () => nextStats as ProfileUsageStats); - authProfileUsageDeps.saveAuthProfileStore(store, agentDir); - logAuthProfileFailureStateChange({ - runId, - profileId, - provider: store.profiles[profileId]?.provider ?? profile.provider, - reason: "rate_limit", - previous: previousStats, - next: nextStats, - now, - }); } /** @@ -1035,11 +965,8 @@ export async function clearAuthProfileCooldown(params: { store.usageStats = updated.usageStats; return; } - if (!store.usageStats?.[profileId]) { - return; + if (updated === null) { + logDroppedAuthProfileBookkeeping("clear_cooldown", profileId); } - - updateUsageStatsEntry(store, profileId, (existing) => resetUsageStats(existing)); - authProfileUsageDeps.saveAuthProfileStore(store, agentDir); } export { testing as __testing };