fix(qoder): coalesce concurrent job token exchanges (#5254)

Integrated into release/v3.8.40

open-sse/services/qoderCli.ts

@@ -407,6 +407,10 @@ const QODER_JOB_TOKEN_MIN_TTL_MS = 60 * 1000;
 
 type QoderJobTokenCacheEntry = { jobToken: string; expiresAt: number };
 const qoderJobTokenCache = new Map<string, QoderJobTokenCacheEntry>();
+const qoderJobTokenPending = new Map<
+  string,
+  Promise<{ jobToken: string; expiresInMs: number } | null>
+>();
 
 type FetchLike = (input: string, init?: Record<string, unknown>) => Promise<Response>;
 
@@ -482,7 +486,14 @@ export async function resolveQoderJobToken(
   const cached = qoderJobTokenCache.get(trimmed);
   if (cached && cached.expiresAt > now) return cached.jobToken;
 
-  const exchanged = await exchangeQoderJobToken(trimmed, options);
+  let pending = qoderJobTokenPending.get(trimmed);
+  if (!pending) {
+    pending = exchangeQoderJobToken(trimmed, options).finally(() => {
+      qoderJobTokenPending.delete(trimmed);
+    });
+    qoderJobTokenPending.set(trimmed, pending);
+  }
+  const exchanged = await pending;
   if (!exchanged) return trimmed; // graceful fallback — keep prior behavior
   qoderJobTokenCache.set(trimmed, {
     jobToken: exchanged.jobToken,
@@ -494,6 +505,7 @@ export async function resolveQoderJobToken(
 /** Test-only: clear the job-token cache so unit tests don't leak state. */
 export function __clearQoderJobTokenCache(): void {
   qoderJobTokenCache.clear();
+  qoderJobTokenPending.clear();
 }
 
 export async function validateQoderCliPat({

tests/unit/qoder-jobtoken-exchange-4683.test.ts

@@ -76,6 +76,34 @@ test("#4683 resolveQoderJobToken exchanges a pt-* once and caches the jt-*", asy
   __clearQoderJobTokenCache();
 });
 
+test("#4683 resolveQoderJobToken coalesces concurrent pt-* exchanges", async () => {
+  __clearQoderJobTokenCache();
+  let fetchCount = 0;
+  let releaseExchange: (() => void) | undefined;
+  let markExchangeStarted: (() => void) | undefined;
+  const exchangeStarted = new Promise<void>((resolveStarted) => {
+    markExchangeStarted = resolveStarted;
+  });
+  const fetchImpl = async () => {
+    fetchCount += 1;
+    markExchangeStarted?.();
+    await new Promise<void>((release) => {
+      releaseExchange = release;
+    });
+    return jsonResponse({ job_token: "jt-shared", expires_in: 86400 });
+  };
+
+  const resolves = Array.from({ length: 8 }, () =>
+    resolveQoderJobToken("pt-concurrent", { fetchImpl, now: 1_000 })
+  );
+  await exchangeStarted;
+  releaseExchange?.();
+  const tokens = await Promise.all(resolves);
+  assert.deepEqual(tokens, Array(8).fill("jt-shared"));
+  assert.equal(fetchCount, 1, "concurrent resolves must share one upstream exchange");
+  __clearQoderJobTokenCache();
+});
+
 test("#4683 resolveQoderJobToken passes a jt-* through without exchanging", async () => {
   __clearQoderJobTokenCache();
   let fetchCount = 0;