Coordinate a2amcp credential refresh

src/oauth.ts

@@ -1,5 +1,8 @@
 import crypto from "node:crypto";
+import { promises as fs } from "node:fs";
 import http from "node:http";
+import os from "node:os";
+import path from "node:path";
 import { spawn } from "node:child_process";
 import type { AddressInfo } from "node:net";
 
@@ -7,6 +10,7 @@ import { ControlPlaneClient } from "./api.js";
 import {
   DEFAULT_API_URL,
   type Credentials,
+  loadCredentials,
   saveCredentials,
 } from "./credentials.js";
 
@@ -15,6 +19,9 @@ export const DEFAULT_OAUTH_CLIENT_ID = "a2acloud-cli";
 export const DEFAULT_OAUTH_SCOPE = "openid email offline_access mcp:invoke agent:read";
 export const DEFAULT_REDIRECT_PORT = 41873;
 const CALLBACK_PATH = "/callback";
+const REFRESH_LOCK_WAIT_MS = 10_000;
+const REFRESH_LOCK_STALE_MS = 30_000;
+const REFRESH_LOCK_POLL_MS = 100;
 
 type OpenIdConfiguration = {
   authorization_endpoint: string;
@@ -116,20 +123,40 @@ export async function refreshCredentialsIfNeeded(
   minTtlSeconds = 60,
 ): Promise<Credentials> {
   if (!creds.refreshToken || !creds.issuer || !creds.clientId) return creds;
-  if (creds.expiresAt && creds.expiresAt - nowSeconds() > minTtlSeconds) return creds;
+  if (credentialsFresh(creds, minTtlSeconds)) return creds;
 
-  const cfg = await discoverOpenIdConfiguration(creds.issuer);
+  return withRefreshLock(async () => {
+    const latest = await loadCredentials();
+    if (latest && sameOAuthClient(latest, creds)) {
+      if (credentialsFresh(latest, minTtlSeconds)) return latest;
+      if (latest.refreshToken && latest.issuer && latest.clientId) creds = latest;
+    }
+
+    return refreshCredentials(creds);
+  });
+}
+
+async function refreshCredentials(creds: Credentials): Promise<Credentials> {
+  const cfg = await discoverOpenIdConfiguration(creds.issuer!);
   const body = new URLSearchParams({
     grant_type: "refresh_token",
-    client_id: creds.clientId,
-    refresh_token: creds.refreshToken,
+    client_id: creds.clientId!,
+    refresh_token: creds.refreshToken!,
   });
   const resp = await fetch(cfg.token_endpoint, {
     method: "POST",
     headers: { "content-type": "application/x-www-form-urlencoded" },
     body,
   });
-  const token = await parseTokenResponse(resp);
+  let token: TokenResponse;
+  try {
+    token = await parseTokenResponse(resp);
+  } catch (err) {
+    const detail = err instanceof Error ? err.message : String(err);
+    throw new Error(
+      `Login expired or revoked. Run \`a2amcp login\` and retry. (${detail})`,
+    );
+  }
   const next: Credentials = {
     ...creds,
     token: token.access_token!,
@@ -141,6 +168,74 @@ export async function refreshCredentialsIfNeeded(
   return next;
 }
 
+function credentialsFresh(creds: Credentials, minTtlSeconds: number): boolean {
+  return Boolean(creds.expiresAt && creds.expiresAt - nowSeconds() > minTtlSeconds);
+}
+
+function sameOAuthClient(a: Credentials, b: Credentials): boolean {
+  return (
+    a.apiUrl === b.apiUrl &&
+    a.issuer === b.issuer &&
+    a.clientId === b.clientId
+  );
+}
+
+async function withRefreshLock<T>(fn: () => Promise<T>): Promise<T> {
+  const release = await acquireRefreshLock();
+  try {
+    return await fn();
+  } finally {
+    await release();
+  }
+}
+
+async function acquireRefreshLock(): Promise<() => Promise<void>> {
+  const file = refreshLockFile();
+  await fs.mkdir(path.dirname(file), { recursive: true });
+  const started = Date.now();
+
+  while (true) {
+    try {
+      const handle = await fs.open(file, "wx", 0o600);
+      try {
+        await handle.writeFile(
+          JSON.stringify({ pid: process.pid, created_at: new Date().toISOString() }),
+        );
+      } finally {
+        await handle.close();
+      }
+      return async () => {
+        try {
+          await fs.unlink(file);
+        } catch (err: any) {
+          if (err?.code !== "ENOENT") throw err;
+        }
+      };
+    } catch (err: any) {
+      if (err?.code !== "EEXIST") throw err;
+      await removeStaleRefreshLock(file);
+      if (Date.now() - started > REFRESH_LOCK_WAIT_MS) {
+        throw new Error("timed out waiting for credential refresh lock");
+      }
+      await sleep(REFRESH_LOCK_POLL_MS);
+    }
+  }
+}
+
+async function removeStaleRefreshLock(file: string): Promise<void> {
+  try {
+    const stat = await fs.stat(file);
+    if (Date.now() - stat.mtimeMs <= REFRESH_LOCK_STALE_MS) return;
+    await fs.unlink(file);
+  } catch (err: any) {
+    if (err?.code !== "ENOENT") throw err;
+  }
+}
+
+function refreshLockFile(): string {
+  return path.join(os.homedir(), ".a2a", "credentials-refresh.lock");
+}
+
 export async function discoverOpenIdConfiguration(
   issuer: string,
 ): Promise<OpenIdConfiguration> {
@@ -338,6 +433,10 @@ function nowSeconds(): number {
   return Math.floor(Date.now() / 1000);
 }
 
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
 function parseJsonObject(text: string): Record<string, unknown> {
   if (!text.trim()) return {};
   try {