fix(oauth): resume retry-safe Anthropic intent cleanup

Persist an attempt-scoped cleanup-pending marker before returning a definitive rejection or pre-dispatch abort. Serialize all refresh-intent mutations with the existing SQLite config transaction, refuse guard overwrites, and retry only the exact safe cleanup before provider redispatch. Preserve ordinary intents for timeouts and post-provider persistence uncertainty.

Author: luvs01

src/oauth/index.ts

@@ -4,7 +4,31 @@ import type { OcxConfig, OcxProviderConfig, RefreshPolicy } from "../types";
 import { loadConfig, resolveEnvValue, saveConfig } from "../config";
 import { maskEmail } from "../lib/privacy";
 import { KiroTokenRefreshError, environmentKiroRoutingMetadata, loginKiro, refreshKiroToken, settleKiroLoginTransaction } from "./kiro";
-import { getAccountCredential, getAccountCredentialWithStatus, getAccountSet, removeAccount, saveAccountCredential, saveCredential, setActiveAccount, getCredential, credentialGeneration, createOAuthRefreshIntentLock, mergeAccountCredential, markAccountNeedsReauthIfGeneration, readOAuthRefreshIntent, writeOAuthRefreshIntent, markOAuthRefreshIntentStaleOwner, clearOAuthRefreshIntent, normalizeAuthStoreBuffer, OAuthMutationBusyError } from "./store";
+import {
+  OAuthMutationBusyError,
+  OAuthRefreshIntentIOError,
+  clearOAuthRefreshIntent,
+  clearOAuthRefreshIntentIfMatch,
+  createOAuthRefreshIntentLock,
+  credentialGeneration,
+  getAccountCredential,
+  getAccountCredentialWithStatus,
+  getAccountSet,
+  getCredential,
+  markAccountNeedsReauthIfGeneration,
+  markOAuthRefreshIntentCleanupPending,
+  markOAuthRefreshIntentStaleOwner,
+  mergeAccountCredential,
+  normalizeAuthStoreBuffer,
+  readOAuthRefreshIntent,
+  removeAccount,
+  saveAccountCredential,
+  saveCredential,
+  setActiveAccount,
+  writeOAuthRefreshIntent,
+  type OAuthRefreshIntent,
+  type OAuthRefreshIntentCleanupPending,
+} from "./store";
 import { loginXai, refreshXaiToken, XAI_LOCAL_CLI_DETACH_WARNING, XaiTokenRequestError } from "./xai";
 import { ANTHROPIC_OAUTH_BETA, AnthropicTokenError, loginAnthropic, refreshAnthropicToken } from "./anthropic";
 import { loginKimi, refreshKimiToken } from "./kimi";
@@ -559,7 +583,7 @@ function terminal(error:unknown):boolean{
   // Local durable-write/read/cleanup failures are operational, not credential
   // death: the provider credential was never rejected or consumed. Never mark
   // the account needsReauth for broken local persistence infrastructure.
-  if (error instanceof RefreshIntentIOError) return false;
+  if (error instanceof RefreshIntentIOError || error instanceof OAuthRefreshIntentIOError) return false;
   return isTerminalRefreshError(error);
 }
 
@@ -583,24 +607,107 @@ function definitivelyAnswered(error: unknown): boolean {
 /**
  * Intent cleanup is secondary to the refresh outcome it protects.
  *
- * A filesystem failure here must not replace a provider error, a pre-dispatch abort, or a
- * successfully persisted credential. Leaving the intent in place is the conservative fallback:
- * it blocks replay until the local persistence problem is repaired.
+ * Once a credential is already durable, cleanup remains secondary and best-effort. A known
+ * failed attempt takes the stricter path below: its retry-safe marker must become durable before
+ * the original provider error can be returned.
  */
 function clearAnthropicRefreshIntentBestEffort(
   provider: string,
   accountId: string,
-  generation: string,
+  expected: OAuthRefreshIntent,
 ): boolean {
   try {
-    return clearOAuthRefreshIntent(provider, accountId, generation);
+    return expected.attemptId
+      ? clearOAuthRefreshIntentIfMatch(provider, accountId, expected)
+      : clearOAuthRefreshIntent(provider, accountId, expected.generation);
   } catch {
     console.warn(
       "[opencodex] Anthropic refresh intent cleanup failed; preserving the durable replay guard.",
     );
     return false;
   }
 }
+
+function clearAnthropicRefreshIntentForKnownFailure(
+  provider: string,
+  accountId: string,
+  expected: OAuthRefreshIntent,
+  cleanupPending: OAuthRefreshIntentCleanupPending,
+  refreshError: unknown,
+): boolean {
+  let marked: OAuthRefreshIntent | undefined;
+  try {
+    marked = markOAuthRefreshIntentCleanupPending(
+      provider,
+      accountId,
+      expected,
+      cleanupPending,
+    );
+  } catch (cause) {
+    throw new OAuthRefreshIntentIOError(
+      "mark-cleanup-pending",
+      cause,
+      refreshError,
+    );
+  }
+  if (!marked) {
+    throw new OAuthRefreshIntentIOError(
+      "mark-cleanup-pending",
+      new Error("Anthropic refresh intent changed before safe cleanup"),
+      refreshError,
+    );
+  }
+
+  let cleared: boolean;
+  try {
+    cleared = clearOAuthRefreshIntentIfMatch(provider, accountId, marked);
+  } catch {
+    console.warn(
+      "[opencodex] Anthropic refresh intent cleanup failed; retry-safe cleanup remains pending.",
+    );
+    return false;
+  }
+  if (!cleared) {
+    throw new OAuthRefreshIntentIOError(
+      "clear-cleanup-pending",
+      new Error("Anthropic refresh intent changed during safe cleanup"),
+      refreshError,
+    );
+  }
+  return true;
+}
+
+function resumeAnthropicRefreshIntentCleanup(
+  provider: string,
+  accountId: string,
+  pendingIntent: OAuthRefreshIntent,
+): void {
+  let cleared: boolean;
+  try {
+    cleared = clearOAuthRefreshIntentIfMatch(provider, accountId, pendingIntent);
+  } catch (cause) {
+    throw new OAuthRefreshIntentIOError(
+      "resume-cleanup",
+      cause,
+    );
+  }
+  if (!cleared) {
+    throw new OAuthRefreshIntentIOError(
+      "resume-cleanup",
+      new Error("Pending Anthropic refresh intent changed before cleanup"),
+    );
+  }
+}
+
+function clearObservedAnthropicRefreshIntent(
+  provider: string,
+  accountId: string,
+  pendingIntent: OAuthRefreshIntent,
+): boolean {
+  return pendingIntent.attemptId
+    ? clearOAuthRefreshIntentIfMatch(provider, accountId, pendingIntent)
+    : clearOAuthRefreshIntent(provider, accountId, pendingIntent.generation);
+}
 function authoritative(stored:OAuthCredentials,active:boolean,now:()=>number):OAuthCredentials{if(stored.source!=="local-cli")return stored;const disk=detectGrokCliToken();if(!disk)return stored;const allowed=isSameGrokIdentity(stored,disk)||(active&&!hasComparableGrokIdentity(stored,disk));return allowed&&shouldAdoptGrokGeneration(stored,disk,now(),REFRESH_SKEW_MS)?disk:stored;}
 function merged(fresh: OAuthCredentials, previous: OAuthCredentials): OAuthCredentials {
   return {
@@ -679,18 +786,22 @@ export async function refreshAnthropicAccountWithLock(
     const account = getAccountSet(provider)?.accounts.find(candidate => candidate.id === accountId);
     const generation = credentialGeneration(stored);
     let pendingIntent = readOAuthRefreshIntent(provider, accountId);
+    if (pendingIntent?.cleanupPending && pendingIntent.generation === generation) {
+      resumeAnthropicRefreshIntentCleanup(provider, accountId, pendingIntent);
+      pendingIntent = undefined;
+    }
     const disk = newerClaudeCredential(stored, now());
     if (disk) {
       const outcome = await mergeAccountCredential(provider, accountId, disk, {
         expectedGeneration: credentialGeneration(stored),
         afterPrePersistRead: deps.afterPrePersistRead,
       });
       if (outcome.superseded) {
-        if (pendingIntent) clearOAuthRefreshIntent(provider, accountId, pendingIntent.generation);
+        if (pendingIntent) clearObservedAnthropicRefreshIntent(provider, accountId, pendingIntent);
         if (outcome.stored.expires > now() + REFRESH_SKEW_MS) return outcome.stored.access;
         throw new OAuthLoginRequiredError(provider);
       }
-      if (pendingIntent) clearOAuthRefreshIntent(provider, accountId, pendingIntent.generation);
+      if (pendingIntent) clearObservedAnthropicRefreshIntent(provider, accountId, pendingIntent);
       return disk.access;
     }
     if (!pendingIntent?.uncertain && pendingIntent?.generation === generation) {
@@ -700,15 +811,19 @@ export async function refreshAnthropicAccountWithLock(
           markOAuthRefreshIntentStaleOwner(provider, accountId, generation, deps.replacedStaleFlight.flightId);
           throw new OAuthTokenRefreshStaleError();
         }
-        clearOAuthRefreshIntent(provider, accountId, generation);
+        if (!clearObservedAnthropicRefreshIntent(provider, accountId, pendingIntent)) {
+          throw new OAuthTokenRefreshStaleError();
+        }
         pendingIntent = undefined;
       }
     }
     if (pendingIntent?.uncertain || pendingIntent?.generation === generation) {
       await markAccountNeedsReauthIfGeneration(provider, accountId, generation, writerGeneration);
       throw new OAuthLoginRequiredError(provider);
     }
-    if (pendingIntent) clearOAuthRefreshIntent(provider, accountId, pendingIntent.generation);
+    if (pendingIntent && !clearObservedAnthropicRefreshIntent(provider, accountId, pendingIntent)) {
+      throw new OAuthTokenRefreshStaleError();
+    }
     if (account?.needsReauth) {
       throw new OAuthLoginRequiredError(provider);
     }
@@ -717,8 +832,9 @@ export async function refreshAnthropicAccountWithLock(
     }
 
     let refreshMayHaveReachedProvider = false;
+    let attemptIntent: OAuthRefreshIntent | undefined;
     try {
-      writeOAuthRefreshIntent(provider, accountId, generation, now(), deps.flight?.flightId);
+      attemptIntent = writeOAuthRefreshIntent(provider, accountId, generation, now(), deps.flight?.flightId);
       if (deps.signal?.aborted) throw deps.signal.reason;
       // From this point on, even a synchronous client error is conservatively post-dispatch:
       // the provider may have received and rotated the refresh token before the caller learned
@@ -731,13 +847,13 @@ export async function refreshAnthropicAccountWithLock(
         afterPrePersistRead: deps.afterPrePersistRead,
       });
       if (outcome.superseded) {
-        clearAnthropicRefreshIntentBestEffort(provider, accountId, generation);
+        if (attemptIntent) clearAnthropicRefreshIntentBestEffort(provider, accountId, attemptIntent);
         if (outcome.stored.expires > now() + REFRESH_SKEW_MS) return outcome.stored.access;
         throw new OAuthLoginRequiredError(provider);
       }
       // The rotated credential is durable now. A cleanup failure must not turn that committed
       // success into a refresh failure; the old-generation intent remains a conservative guard.
-      clearAnthropicRefreshIntentBestEffort(provider, accountId, generation);
+      if (attemptIntent) clearAnthropicRefreshIntentBestEffort(provider, accountId, attemptIntent);
       return fresh.access;
     } catch (error) {
       if (error instanceof OAuthMutationBusyError || error instanceof OAuthTokenRefreshStaleError) throw error;
@@ -753,13 +869,19 @@ export async function refreshAnthropicAccountWithLock(
         // carries no status: the server may already have
         // rotated the token, and replaying it could trip refresh-token-reuse revocation.
         // Those outcomes keep the intent so the guard still refuses a blind replay.
-        if (!refreshMayHaveReachedProvider || definitivelyAnswered(error)) {
-          clearAnthropicRefreshIntentBestEffort(provider, accountId, generation);
+        if ((!refreshMayHaveReachedProvider || definitivelyAnswered(error)) && attemptIntent) {
+          clearAnthropicRefreshIntentForKnownFailure(
+            provider,
+            accountId,
+            attemptIntent,
+            refreshMayHaveReachedProvider ? "definitive-rejection" : "pre-dispatch",
+            error,
+          );
         }
         throw error;
       }
       await markAccountNeedsReauthIfGeneration(provider, accountId, generation, writerGeneration);
-      clearAnthropicRefreshIntentBestEffort(provider, accountId, generation);
+      if (attemptIntent) clearAnthropicRefreshIntentBestEffort(provider, accountId, attemptIntent);
       throw new OAuthLoginRequiredError(provider);
     }
   } finally {