feat(ai): implemented idempotent credential refresh with CAS logic

- Added id-based credential persistence and CAS-based invalidation to prevent cross-process race conditions.
- Categorized OAuth failure responses into transient and definitive types for more resilient refresh orchestration.
- Enabled stable credential lookups during refresh to ensure integrity during peer rotation or index shifts.
- Validated new persistence and error handling logic through comprehensive concurrency and failure-mode tests.
This commit is contained in:
can1357
2026-06-25 04:31:58 +02:00
parent 61b430da9d
commit 591f1c5765
6 changed files with 299 additions and 30 deletions
+9
View File
@@ -1,14 +1,23 @@
# Changelog
## [Unreleased]
### Added
- Added `listOAuthAccounts` for retrieving a read-only list of stored OAuth account identities
- Added `getOAuthAccessAt` to resolve an OAuth token exclusively for a specific account position
### Changed
- Refactored OAuth token persistence and disable logic to use stable credential IDs instead of positional indices to prevent race conditions during concurrent updates
- Updated OAuth failure classification to treat 403 status codes, rate limits, and network errors as transient, preventing unnecessary credential invalidation
### Fixed
- Fixed Codex Responses Lite staying enabled for image prompts, which caused GPT/Codex image turns to be rejected as `Invalid value: 'input_image'`; image-bearing Codex requests now fall back to the full Responses transport. ([#3421](https://github.com/can1357/oh-my-pi/issues/3421))
- Fixed the auth-broker background refresher disabling OAuth credentials unconditionally (`disableCredentialById`) on a definitive refresh failure, so a credential another process or a fresh login rotated mid-refresh could be torn down even though the stored row already held a valid token. The definitive-failure teardown now happens inside `AuthStorage.refreshCredentialById` via the same compare-and-set the in-stream and usage-probe paths use — it disables only when the persisted row still matches the credential the refresh actually attempted, and reloads on a CAS loss; the refresher now only logs.
- Fixed OAuth refresh persisting the rotated token by a positional index captured before the refresh `await`. A concurrent disable could reorder or shrink a provider's credential array while the refresh was in flight, landing the new token on the wrong row (or silently dropping it) and leaving accounts with a stale refresh token that failed — and was then disabled — on the next cycle. Refresh persistence, selection-index resync, and CAS-disable now address the row by id across `forceRefreshCredentialById`, candidate preflight, and in-stream selection (`#replaceCredentialById` / `#disableCredentialByIdIfMatches`).
- Fixed `isDefinitiveOAuthFailure` treating a bare HTTP 403 (and generic `unauthorized` / access-token-expired wording) as a definitive credential failure, which permanently disabled healthy OAuth accounts on WAF, egress rate-limit, permission, and account-verification responses. Bare 403, rate limits (429), gateway/5xx, and more network errors (`ECONNRESET`, `ETIMEDOUT`, `EAI_AGAIN`, …) are now classified transient; only explicit dead-grant errors (`invalid_grant`, `invalid_token`, `unauthorized_client`, revoked, `refresh token … expired`) or a bare 401 tear the credential down.
## [16.1.17] - 2026-06-24
+8 -8
View File
@@ -5,9 +5,10 @@
* any whose `expires - Date.now() < refreshSkewMs`. Refresh single-flight
* lives in {@link AuthStorage} so manual and background refreshes share the
* same upstream attempt.
* Definitively-failed credentials (invalid_grant / 401 not from network blip)
* are disabled via {@link AuthStorage.disableCredentialById} so the next
* snapshot pull surfaces a clean delete on the client.
* Definitively-failed credentials (invalid_grant / bare 401, not a network
* blip) are torn down inside {@link AuthStorage.refreshCredentialById} via a
* compare-and-set disable — only when no peer/login rotated the row first — so
* the next snapshot pull surfaces a clean delete on the client.
*/
import { logger } from "@oh-my-pi/pi-utils";
import { type AuthStorage, isDefinitiveOAuthFailure } from "../auth-storage";
@@ -104,11 +105,10 @@ export class AuthBrokerRefresher {
} catch (error) {
const errorMsg = String(error);
if (isDefinitiveOAuthFailure(errorMsg)) {
logger.warn("auth-broker refresh failed definitively; disabling credential", {
id,
error: errorMsg,
});
this.#storage.disableCredentialById(id, `auth-broker refresh failed: ${errorMsg}`);
// AuthStorage.refreshCredentialById already CAS-disabled the row
// (unless a peer/login rotated it first, in which case the live
// credential is intentionally kept). Nothing to do here but record it.
logger.warn("auth-broker refresh failed definitively", { id, error: errorMsg });
} else {
logger.debug("auth-broker refresh failed (transient)", { id, error: errorMsg });
}
+121 -22
View File
@@ -568,9 +568,17 @@ const MAX_PENDING_DISABLED_EVENTS = 32;
* while streaming requests correctly tear the row down.
*/
const OAUTH_DEFINITIVE_FAILURE_REGEX =
/invalid_grant|invalid_token|revoked|unauthorized|expired.*refresh|refresh.*expired/i;
const OAUTH_TRANSIENT_FAILURE_REGEX = /timeout|network|fetch failed|ECONNREFUSED/i;
const OAUTH_HTTP_AUTH_REGEX = /\b(401|403)\b/;
/invalid_grant|invalid_token|unauthorized_client|\brevoked\b|refresh[\s_]?token.*expired/i;
// Transient: network blips, rate limits, gateway/5xx, and infra denials
// (WAF / egress 403, permission / account-verification) — block-and-retry,
// never tear the credential down for these.
const OAUTH_TRANSIENT_FAILURE_REGEX =
/timeout|network|fetch failed|ECONN(?:REFUSED|RESET)|ETIMEDOUT|EAI_AGAIN|socket hang up|\b(?:408|425|429|5\d{2})\b|rate.?limit|too many requests|temporar|unavailable|forbidden|permission_denied|cloudflare|captcha/i;
// A bare 401 from an OAuth token endpoint means the stored grant/client is
// dead. 403 is deliberately excluded: it is overwhelmingly WAF / egress
// rate-limit / permission / account-verification — none of which mean the
// refresh token itself is invalid.
const OAUTH_HTTP_AUTH_REGEX = /\b401\b/;
export function isDefinitiveOAuthFailure(errorMsg: string): boolean {
if (OAUTH_DEFINITIVE_FAILURE_REGEX.test(errorMsg)) return true;
@@ -1534,6 +1542,42 @@ export class AuthStorage {
return true;
}
/**
* Persist a refreshed credential addressed by id, not a positional index.
* A concurrent disable can reorder/shrink the provider's row array while an
* async refresh is in flight, so a pre-await index is unsafe; resolving the
* row by id at write time lands the rotated token on the correct row. Returns
* the row's current index, or -1 when it was disabled/removed mid-refresh.
*/
#replaceCredentialById(provider: string, id: number, credential: AuthCredential): number {
const entries = this.#getStoredCredentials(provider);
const index = entries.findIndex(entry => entry.id === id);
if (index === -1) return -1;
this.#store.updateAuthCredential(id, credential);
const updated = [...entries];
updated[index] = { id, credential };
this.#setStoredCredentials(provider, updated);
return index;
}
/**
* CAS-disable the row with `id`, but only if its persisted credential still
* matches `expected` — i.e. no peer/login rotated it while we refreshed.
* Addresses the row by id (re-resolved here, then matched on `data` in the
* store) so a concurrent reorder can't tear down the wrong credential.
*/
#disableCredentialByIdIfMatches(
provider: string,
id: number,
expected: AuthCredential,
disabledCause: string,
): boolean {
const entries = this.#getStoredCredentials(provider);
const index = entries.findIndex(entry => entry.id === id);
if (index === -1) return false;
return this.#tryDisableCredentialAtIfMatches(provider, index, expected, disabledCause);
}
#emitCredentialDisabled(event: CredentialDisabledEvent): void {
if (this.#credentialDisabledListeners.size === 0) {
// No subscribers — buffer for later replay. Cap the backlog so a process that runs
@@ -3293,7 +3337,12 @@ export class AuthStorage {
type: "oauth",
};
candidate.selection.credential = updated;
this.#replaceCredentialAt(provider, candidate.selection.index, updated);
if (credentialId !== undefined) {
const idx = this.#replaceCredentialById(provider, credentialId, updated);
if (idx !== -1) candidate.selection.index = idx;
} else {
this.#replaceCredentialAt(provider, candidate.selection.index, updated);
}
} catch (error) {
// Recovery for definitive failures (incl. peer rotation) lives in
// #tryOAuthCredential; log instead of swallowing silently — a bare
@@ -3497,6 +3546,11 @@ export class AuthStorage {
if (!(await this.#prepareOAuthCredentialForRequest(provider, selection, options))) {
return undefined;
}
// Capture the row id once, immediately after #prepareOAuthCredentialForRequest
// resynced selection.index from the store. A concurrent disable during the
// usage/refresh awaits below can shift positional indices, so every later
// refresh / persist / CAS-disable addresses the row by this stable id.
const credentialId = this.#getStoredCredentials(provider)[selection.index]?.id;
const requiresProModel = requiresOpenAICodexProModel(provider, options?.modelId);
const applyProFilter = enforceProRequirement ?? requiresProModel;
@@ -3539,7 +3593,7 @@ export class AuthStorage {
const refreshedCredentials = await this.#refreshOAuthCredential(
provider,
selection.credential,
this.#getStoredCredentials(provider)[selection.index]?.id,
credentialId,
options?.signal,
);
const apiKey = customProvider.getApiKey
@@ -3555,7 +3609,7 @@ export class AuthStorage {
const refreshedCredentials = await this.#refreshOAuthCredential(
provider,
selection.credential,
this.#getStoredCredentials(provider)[selection.index]?.id,
credentialId,
options?.signal,
);
const oauthCreds: Record<string, OAuthCredentials> = {
@@ -3575,7 +3629,12 @@ export class AuthStorage {
enterpriseUrl: result.newCredentials.enterpriseUrl ?? selection.credential.enterpriseUrl,
apiEndpoint: result.newCredentials.apiEndpoint ?? selection.credential.apiEndpoint,
};
this.#replaceCredentialAt(provider, selection.index, updated);
if (credentialId !== undefined) {
const idx = this.#replaceCredentialById(provider, credentialId, updated);
if (idx !== -1) selection.index = idx;
} else {
this.#replaceCredentialAt(provider, selection.index, updated);
}
if ((checkUsage && !allowBlocked) || requiresProModel) {
const sameAccount = selection.credential.accountId === updated.accountId;
if (!usageChecked || !sameAccount) {
@@ -3625,7 +3684,6 @@ export class AuthStorage {
// refresh token has changed, the peer rotation succeeded and we should pick
// up the new credential instead of soft-deleting the row that the peer just
// updated.
const credentialId = this.#getStoredCredentials(provider)[selection.index]?.id;
if (credentialId !== undefined) {
const latestRow = this.#store.listAuthCredentials(provider).find(row => row.id === credentialId);
const latestCredential = latestRow?.credential;
@@ -3643,12 +3701,20 @@ export class AuthStorage {
// Use a CAS-style disable conditioned on the row still containing the stale credential
// we tried to refresh, so a peer rotation that lands between the pre-check above and
// this disable doesn't soft-delete the freshly-rotated row.
const disabled = this.#tryDisableCredentialAtIfMatches(
provider,
selection.index,
selection.credential,
`oauth refresh failed: ${errorMsg}`,
);
const disabled =
credentialId !== undefined
? this.#disableCredentialByIdIfMatches(
provider,
credentialId,
selection.credential,
`oauth refresh failed: ${errorMsg}`,
)
: this.#tryDisableCredentialAtIfMatches(
provider,
selection.index,
selection.credential,
`oauth refresh failed: ${errorMsg}`,
);
if (!disabled) {
logger.debug("OAuth refresh disable lost CAS; reloading after peer rotation", {
provider,
@@ -4300,22 +4366,55 @@ export class AuthStorage {
if (target.credential.type !== "oauth") {
throw new Error(`Credential ${id} is not OAuth (provider=${provider}, type=${target.credential.type})`);
}
// The exact credential we are about to refresh — captured before the
// await so a definitive failure can CAS-disable the row against the
// value we actually attempted (NOT the expires:0 clone below).
const attempted = target.credential;
// Pass a clone with expires=0 so the cached not-yet-expired short-circuit
// in #refreshOAuthCredential doesn't suppress the requested refresh.
const stale: OAuthCredential = { ...target.credential, expires: 0 };
const refreshed = await this.#refreshOAuthCredential(provider as Provider, stale, id, signal);
const stale: OAuthCredential = { ...attempted, expires: 0 };
let refreshed: OAuthCredentials;
try {
refreshed = await this.#refreshOAuthCredential(provider as Provider, stale, id, signal);
} catch (error) {
// A definitively-dead grant tears the row down here, where the
// attempted credential is known. CAS on the persisted credential so a
// peer/login rotation in flight leaves the freshly-rotated row intact.
if (isDefinitiveOAuthFailure(String(error))) {
// CAS-loss (false) means a peer/login rotated the row mid-refresh, so
// our #data copy is stale — reload so the next caller serves the
// freshly-rotated credential rather than the dead token we attempted.
if (
!this.#disableCredentialByIdIfMatches(
provider,
id,
attempted,
`oauth refresh failed: ${String(error)}`,
)
) {
await this.reload();
}
}
throw error;
}
const updated: OAuthCredential = {
type: "oauth",
access: refreshed.access,
refresh: refreshed.refresh,
expires: refreshed.expires,
accountId: refreshed.accountId ?? target.credential.accountId,
email: refreshed.email ?? target.credential.email,
projectId: refreshed.projectId ?? target.credential.projectId,
enterpriseUrl: refreshed.enterpriseUrl ?? target.credential.enterpriseUrl,
apiEndpoint: refreshed.apiEndpoint ?? target.credential.apiEndpoint,
accountId: refreshed.accountId ?? attempted.accountId,
email: refreshed.email ?? attempted.email,
projectId: refreshed.projectId ?? attempted.projectId,
enterpriseUrl: refreshed.enterpriseUrl ?? attempted.enterpriseUrl,
apiEndpoint: refreshed.apiEndpoint ?? attempted.apiEndpoint,
};
this.#replaceCredentialAt(provider, index, updated);
// Persist by id: the array may have been reordered/shrunk while the
// refresh was in flight, so the pre-await positional index is unsafe. A
// -1 means the row was disabled/removed mid-refresh — surface that as a
// miss rather than implying a live row the snapshot won't contain.
if (this.#replaceCredentialById(provider, id, updated) === -1) {
throw new Error(`No credential with id=${id}`);
}
return {
id,
provider,
@@ -148,4 +148,51 @@ describe("AuthBrokerRefresher", () => {
expect(disableEvents).toHaveLength(0);
expect(storage.exportSnapshot().credentials).toHaveLength(1);
});
test("does not disable a credential a peer rotated during the refresh (CAS)", async () => {
const now = 1_700_000_000_000;
store!.saveOAuth("anthropic", {
access: "stale",
refresh: "stale-refresh",
expires: now + 60_000,
accountId: "a",
});
storage = new AuthStorage(store!);
const disableEvents: string[] = [];
storage.onCredentialDisabled(event => {
disableEvents.push(event.disabledCause);
});
await storage.reload();
const id = store!.listAuthCredentials("anthropic")[0]!.id;
// Our refresh fails with a dead-grant error, but a peer (another process /
// a fresh login) rotates the persisted row to a new token first. The
// CAS disable must see the row no longer holds the token we attempted and
// leave the freshly-rotated credential intact instead of clobbering it.
vi.spyOn(oauthUtils, "refreshOAuthToken").mockImplementation(async () => {
store!.updateAuthCredential(id, {
type: "oauth",
access: "fresh-from-peer",
refresh: "fresh-refresh-from-peer",
expires: now + 60 * 60_000,
accountId: "a",
});
throw new Error('HTTP 400 invalid_grant {"error":"invalid_grant"}');
});
const refresher = new AuthBrokerRefresher({
storage,
refreshSkewMs: 5 * 60_000,
now: () => now,
});
await refresher.tick();
expect(disableEvents).toHaveLength(0);
const rows = store!.listAuthCredentials("anthropic");
expect(rows).toHaveLength(1);
expect(rows[0]?.credential.type).toBe("oauth");
if (rows[0]?.credential.type === "oauth") {
expect(rows[0].credential.refresh).toBe("fresh-refresh-from-peer");
}
});
});
@@ -384,4 +384,51 @@ describe("AuthStorage OAuth refresh race", () => {
const retryKey = await authStorage.getApiKey("unit-oauth-rotation", sessionId);
expect(retryKey).toBe("access-b");
});
test("persists a refreshed token by id when a concurrent disable shifts indices", async () => {
if (!authStorage || !store) throw new Error("test setup failed");
const now = Date.now();
// Three distinct expired accounts → index order A, B, C by id ascending.
await authStorage.set("anthropic", [
{ type: "oauth", access: "a-acc", refresh: "a-ref", expires: now - 60_000, accountId: "acc-a", email: "a@x" },
{ type: "oauth", access: "b-acc", refresh: "b-ref", expires: now - 60_000, accountId: "acc-b", email: "b@x" },
{ type: "oauth", access: "c-acc", refresh: "c-ref", expires: now - 60_000, accountId: "acc-c", email: "c@x" },
]);
const seeded = store.listAuthCredentials("anthropic");
expect(seeded).toHaveLength(3);
const idA = seeded[0]!.id;
const idB = seeded[1]!.id;
const idC = seeded[2]!.id;
// While B refreshes, a definitive failure disables A — removing index 0 and
// shifting B from index 1 to index 0. A pre-await positional write would
// land B's rotated token on C; the id-addressed write must hit B and leave
// C untouched.
vi.spyOn(oauthUtils, "refreshOAuthToken").mockImplementation(async (_provider, credential) => {
if (credential.refresh === "b-ref") {
authStorage!.disableCredentialById(idA, "test: concurrent disable");
return {
access: "b-fresh",
refresh: "b-fresh-ref",
expires: now + 60 * 60_000,
accountId: "acc-b",
email: "b@x",
};
}
return { ...credential, expires: now + 60 * 60_000 };
});
await withEnv(SUPPRESS_ANTHROPIC_ENV, async () => {
const refreshed = await authStorage!.forceRefreshCredentialById(idB);
expect(refreshed.id).toBe(idB);
});
const after = store.listAuthCredentials("anthropic");
const bRow = after.find(row => row.id === idB);
const cRow = after.find(row => row.id === idC);
expect(bRow?.credential.type).toBe("oauth");
if (bRow?.credential.type === "oauth") expect(bRow.credential.refresh).toBe("b-fresh-ref");
expect(cRow?.credential.type).toBe("oauth");
if (cRow?.credential.type === "oauth") expect(cRow.credential.refresh).toBe("c-ref");
});
});
@@ -0,0 +1,67 @@
/**
* Contract for {@link isDefinitiveOAuthFailure} — the shared classifier that
* decides whether an OAuth refresh error tears the credential down (re-login
* required) or is a transient blip to block-and-retry. A false positive here
* permanently disables a healthy account, so the 403 / rate-limit / 5xx cases
* below are load-bearing, not cosmetic.
*/
import { describe, expect, it } from "bun:test";
import { isDefinitiveOAuthFailure } from "@oh-my-pi/pi-ai/auth-storage";
describe("isDefinitiveOAuthFailure", () => {
it("treats explicit dead-grant errors as definitive", () => {
for (const msg of [
'HTTP 400 invalid_grant {"error":"invalid_grant"}',
"invalid_token",
"OAuth refresh failed: refresh token revoked",
'invalid_grant {"error_description":"Refresh token expired"}',
"unauthorized_client",
]) {
expect(isDefinitiveOAuthFailure(msg)).toBe(true);
}
});
it("treats a bare 401 from the token endpoint as definitive", () => {
expect(isDefinitiveOAuthFailure("HTTP 401 Unauthorized")).toBe(true);
});
it("never treats a bare 403 as definitive (WAF / egress / permission, not a dead token)", () => {
// Regression: a shared broker egress IP that gets 403'd by the provider,
// or a google PERMISSION_DENIED / account-verification 403, must NOT
// permanently disable an otherwise-valid credential.
expect(isDefinitiveOAuthFailure("HTTP 403 Forbidden")).toBe(false);
expect(isDefinitiveOAuthFailure("403 PERMISSION_DENIED: account verification required")).toBe(false);
expect(isDefinitiveOAuthFailure("blocked by cloudflare (403)")).toBe(false);
});
it("treats rate-limit and server/gateway errors as transient", () => {
for (const msg of [
"429 too many requests",
"HTTP 503 Service Unavailable",
"500 internal server error",
"rate limit exceeded",
]) {
expect(isDefinitiveOAuthFailure(msg)).toBe(false);
}
});
it("treats network blips as transient (incl. ECONNRESET)", () => {
for (const msg of [
"fetch failed: ECONNRESET",
"fetch failed: ECONNREFUSED",
"ETIMEDOUT",
"socket hang up",
"network error",
"OAuth token refresh timed out for provider: anthropic",
]) {
expect(isDefinitiveOAuthFailure(msg)).toBe(false);
}
});
it("lets a transient signal override a bare 401 (rate-limited auth endpoint)", () => {
// A 401 wrapped in a rate-limit / 5xx context is the provider throttling,
// not a dead grant — block-and-retry instead of nuking the row.
expect(isDefinitiveOAuthFailure("401 unauthorized — 429 too many requests")).toBe(false);
expect(isDefinitiveOAuthFailure("502 bad gateway (was 401 upstream)")).toBe(false);
});
});