Coordinate a2amcp credential refresh

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "a2amcp",
-  "version": "0.1.8",
+  "version": "0.1.10",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "a2amcp",
-      "version": "0.1.8",
+      "version": "0.1.10",
       "license": "Apache-2.0",
       "dependencies": {
         "@modelcontextprotocol/sdk": "^1.0.4",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "a2amcp",
-  "version": "0.1.8",
+  "version": "0.1.10",
   "description": "MCP gateway for a2acloud agents. Run as a local MCP server; expose any number of deployed A2A agents to Claude Code, Cursor, and other MCP clients.",
   "type": "module",
   "bin": {

src/api.ts

@@ -24,8 +24,11 @@ export class ControlPlaneClient {
   }
 
   private headers(): Record<string, string> {
-    const h: Record<string, string> = { "content-type": "application/json" };
-    if (this.token) h["authorization"] = `bearer ${this.token}`;
+    const h: Record<string, string> = {
+      accept: "application/json",
+      "content-type": "application/json",
+    };
+    if (this.token) h["authorization"] = `Bearer ${this.token}`;
     return h;
   }
 
@@ -35,18 +38,12 @@ export class ControlPlaneClient {
       headers: this.headers(),
       body: body !== undefined ? JSON.stringify(body) : undefined,
     });
+    const text = await resp.text();
     if (!resp.ok) {
-      let detail: string;
-      try {
-        const j: any = await resp.json();
-        detail = j?.detail ?? JSON.stringify(j);
-      } catch {
-        detail = await resp.text();
-      }
-      throw new ApiError(resp.status, detail);
+      throw new ApiError(resp.status, errorDetail(text, resp.statusText));
     }
     if (resp.status === 204) return undefined as T;
-    return (await resp.json()) as T;
+    return (text ? JSON.parse(text) : undefined) as T;
   }
 
   me() {
@@ -61,3 +58,16 @@ export class ControlPlaneClient {
     return this.request<AgentRow>("GET", `/v1/agents/${encodeURIComponent(name)}`);
   }
 }
+
+function errorDetail(text: string, fallback: string): string {
+  if (!text.trim()) return fallback;
+  try {
+    const parsed = JSON.parse(text);
+    const detail = parsed?.detail;
+    if (typeof detail === "string") return detail;
+    if (detail !== undefined) return JSON.stringify(detail);
+    return JSON.stringify(parsed);
+  } catch {
+    return text;
+  }
+}

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> {
@@ -208,9 +303,10 @@ async function exchangeAuthorizationCode(opts: {
 }
 
 async function parseTokenResponse(resp: Response): Promise<TokenResponse> {
-  const data = (await resp.json().catch(() => ({}))) as TokenResponse;
+  const text = await resp.text();
+  const data = parseJsonObject(text) as TokenResponse;
   if (!resp.ok || !data.access_token) {
-    const message = data.error_description || data.error || resp.statusText;
+    const message = data.error_description || data.error || text.trim() || resp.statusText;
     throw new Error(`Keycloak token exchange failed: ${message}`);
   }
   return data;
@@ -336,3 +432,17 @@ function expiresAt(expiresIn: number | undefined): number | undefined {
 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 {
+    const parsed = JSON.parse(text);
+    return parsed && typeof parsed === "object" ? parsed : {};
+  } catch {
+    return {};
+  }
+}

tests/api.test.ts

@@ -0,0 +1,50 @@
+import { test, beforeEach, afterEach } from "node:test";
+import assert from "node:assert/strict";
+
+import { ApiError, ControlPlaneClient } from "../src/api.js";
+
+const originalFetch = globalThis.fetch;
+
+beforeEach(() => {
+  globalThis.fetch = originalFetch;
+});
+
+afterEach(() => {
+  globalThis.fetch = originalFetch;
+});
+
+test("ControlPlaneClient reports non-JSON error bodies without rereading them", async () => {
+  globalThis.fetch = (async () =>
+    new Response("<h1>upstream unavailable</h1>", {
+      status: 502,
+      statusText: "Bad Gateway",
+      headers: { "content-type": "text/html" },
+    })) as typeof fetch;
+
+  await assert.rejects(
+    () => new ControlPlaneClient("https://api.example.test", "token").me(),
+    (err) => {
+      assert.ok(err instanceof ApiError);
+      assert.equal(err.status, 502);
+      assert.match(err.message, /upstream unavailable/);
+      assert.doesNotMatch(err.message, /Body is unusable/);
+      return true;
+    },
+  );
+});
+
+test("ControlPlaneClient forwards bearer tokens and extracts JSON error detail", async () => {
+  globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => {
+    const headers = init?.headers as Record<string, string>;
+    assert.equal(headers.authorization, "Bearer access-token");
+    return new Response(JSON.stringify({ detail: "keycloak email is not verified" }), {
+      status: 403,
+      headers: { "content-type": "application/json" },
+    });
+  }) as typeof fetch;
+
+  await assert.rejects(
+    () => new ControlPlaneClient("https://api.example.test", "access-token").me(),
+    /API 403: keycloak email is not verified/,
+  );
+});

tests/oauth.test.ts

@@ -93,3 +93,89 @@ test("refreshCredentialsIfNeeded refreshes expired Keycloak access token", async
   assert.equal((await loadCredentials())!.token, "fresh-token");
   assert.ok(calls.some((call) => call.body?.includes("grant_type=refresh_token")));
 });
+
+test("refreshCredentialsIfNeeded serializes concurrent refreshes", async () => {
+  let tokenCalls = 0;
+  globalThis.fetch = (async (input: RequestInfo | URL) => {
+    const url = String(input);
+    if (url.endsWith("/.well-known/openid-configuration")) {
+      return new Response(
+        JSON.stringify({
+          authorization_endpoint: "https://auth.example.test/auth",
+          token_endpoint: "https://auth.example.test/token",
+        }),
+        { status: 200, headers: { "content-type": "application/json" } },
+      );
+    }
+    tokenCalls += 1;
+    await new Promise((resolve) => setTimeout(resolve, 25));
+    return new Response(
+      JSON.stringify({
+        access_token: `fresh-token-${tokenCalls}`,
+        refresh_token: `fresh-refresh-${tokenCalls}`,
+        expires_in: 3600,
+        scope: "openid email mcp:invoke",
+      }),
+      { status: 200, headers: { "content-type": "application/json" } },
+    );
+  }) as typeof fetch;
+
+  await saveCredentials({
+    apiUrl: "https://api.example.test",
+    token: "old-token",
+    refreshToken: "refresh-token",
+    expiresAt: 1,
+    email: "dev@example.test",
+    issuer: "https://auth.example.test",
+    clientId: "a2acloud-cli",
+  });
+
+  const stale = (await loadCredentials())!;
+  const [first, second] = await Promise.all([
+    refreshCredentialsIfNeeded(stale),
+    refreshCredentialsIfNeeded({ ...stale }),
+  ]);
+
+  assert.equal(tokenCalls, 1);
+  assert.equal(first.token, "fresh-token-1");
+  assert.equal(second.token, "fresh-token-1");
+  assert.equal((await loadCredentials())!.refreshToken, "fresh-refresh-1");
+});
+
+test("refreshCredentialsIfNeeded reports stale Keycloak sessions clearly", async () => {
+  globalThis.fetch = (async (input: RequestInfo | URL) => {
+    const url = String(input);
+    if (url.endsWith("/.well-known/openid-configuration")) {
+      return new Response(
+        JSON.stringify({
+          authorization_endpoint: "https://auth.example.test/auth",
+          token_endpoint: "https://auth.example.test/token",
+        }),
+        { status: 200, headers: { "content-type": "application/json" } },
+      );
+    }
+    return new Response(
+      JSON.stringify({
+        error: "invalid_grant",
+        error_description: "Session doesn't have required client",
+      }),
+      { status: 400, headers: { "content-type": "application/json" } },
+    );
+  }) as typeof fetch;
+
+  await saveCredentials({
+    apiUrl: "https://api.example.test",
+    token: "old-token",
+    refreshToken: "refresh-token",
+    expiresAt: 1,
+    email: "dev@example.test",
+    issuer: "https://auth.example.test",
+    clientId: "a2acloud-cli",
+  });
+
+  await assert.rejects(
+    refreshCredentialsIfNeeded((await loadCredentials())!),
+    /Login expired or revoked\. Run `a2amcp login` and retry\. .*Session doesn't have required client/,
+  );
+  assert.equal((await loadCredentials())!.token, "old-token");
+});