fix(sse): wait out in-flight duplicate turns instead of failing them with 409

open-sse/handlers/chatCore.ts

@@ -27,7 +27,7 @@ export {
   relocateDirectiveOnlyMessages,
 } from "./chatCore/claudeSystemRole.ts";
 import { checkIdempotencyCache } from "./chatCore/idempotency.ts";
-import { acquireTurnExecution, createTurnInProgressResult } from "./chatCore/turnExecutionGuard.ts";
+import { acquireTurnExecutionWithWait, createTurnInProgressResult } from "./chatCore/turnExecutionGuard.ts";
 import { checkSemanticCache, isSemanticCacheEnabled } from "./chatCore/semanticCache.ts";
 import { checkLifecycle, resolveLifecycle } from "./chatCore/modelLifecyclePolicy.ts";
 import {
@@ -580,7 +580,7 @@ async function handleChatCoreInner({
     return idempotencyHit;
   }
 
-  const turnExecution = acquireTurnExecution(idempotencyKey);
+  const turnExecution = await acquireTurnExecutionWithWait(idempotencyKey, { log, traceId });
   if (turnExecution.acquired === false) {
     const duplicate = createTurnInProgressResult(turnExecution.retryCount);
     log?.warn?.(

open-sse/handlers/chatCore/turnExecutionGuard.ts

@@ -75,6 +75,57 @@ export function getTurnExecutionSnapshot(key: string) {
   return { retryCount: active.retryCount, ageMs: Math.max(0, Date.now() - active.startedAt) };
 }
 
+export const TURN_WAIT_TIMEOUT_MS = 60_000;
+export const TURN_WAIT_POLL_MS = 500;
+
+export type TurnExecutionWaitOptions = {
+  /** Upper bound on how long a duplicate waits for the in-flight turn to release. */
+  timeoutMs?: number;
+  /** Poll interval while waiting. */
+  pollMs?: number;
+  /** Client abort signal — a disconnected caller stops waiting immediately. */
+  signal?: AbortSignal | null;
+  /** Warn-channel logger receiving the duplicate-waiting notice. */
+  log?: { warn?: (tag: string, msg: string) => void } | null | undefined;
+  traceId?: string;
+};
+
+/**
+ * Acquires the turn slot, waiting out an in-flight duplicate instead of
+ * failing it immediately.
+ *
+ * Executing a duplicate's request path straight to a 409 turns a transient
+ * self-collision — executor/account-fallback retries re-entering chatCore
+ * with the caller's headers while the original turn is stuck on a slow
+ * upstream — into a fatal client-visible failure, and the 409 is then
+ * misread downstream as an account-level rate limit ("reset after Ns"),
+ * cooling down healthy accounts. Waiting keeps the guard's serialization
+ * guarantee without failing the duplicate; after the timeout the caller
+ * falls back to the 409 result as before.
+ */
+export async function acquireTurnExecutionWithWait(
+  key: string | null,
+  opts: TurnExecutionWaitOptions = {}
+): Promise<TurnExecutionAcquireResult> {
+  const turn = acquireTurnExecution(key);
+  if (turn.acquired) return turn;
+
+  const timeoutMs = opts.timeoutMs ?? TURN_WAIT_TIMEOUT_MS;
+  const pollMs = opts.pollMs ?? TURN_WAIT_POLL_MS;
+  const { signal, log, traceId } = opts;
+  log?.warn?.(
+    "TURN_GUARD",
+    `duplicate waiting cid=${traceId ?? "?"} retry=${turn.retryCount} ageMs=${turn.ageMs}`
+  );
+  const deadline = Date.now() + timeoutMs;
+  let current = turn;
+  while (!current.acquired && Date.now() < deadline && !signal?.aborted) {
+    await new Promise((resolve) => setTimeout(resolve, pollMs));
+    current = acquireTurnExecution(key);
+  }
+  return current;
+}
+
 export function clearTurnExecutionsForTesting(): void {
   activeTurns.clear();
 }

tests/unit/chatcore/turn-execution-guard-wait.test.ts

@@ -0,0 +1,62 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+
+import {
+  acquireTurnExecution,
+  acquireTurnExecutionWithWait,
+  clearTurnExecutionsForTesting,
+} from "../../../open-sse/handlers/chatCore/turnExecutionGuard.ts";
+
+test("acquires immediately when no turn is in flight", async () => {
+  clearTurnExecutionsForTesting();
+  const turn = await acquireTurnExecutionWithWait("k");
+  assert.equal(turn.acquired, true);
+});
+
+test("null key remains an unconditional acquire", async () => {
+  clearTurnExecutionsForTesting();
+  const turn = await acquireTurnExecutionWithWait(null);
+  assert.equal(turn.acquired, true);
+});
+
+test("waits for the in-flight turn to release, then acquires", async () => {
+  clearTurnExecutionsForTesting();
+  const first = acquireTurnExecution("k");
+  assert.equal(first.acquired, true);
+  setTimeout(() => first.release(), 120);
+  const waits: string[] = [];
+  const second = await acquireTurnExecutionWithWait("k", {
+    timeoutMs: 2_000,
+    pollMs: 40,
+    log: { warn: (tag, msg) => waits.push(`${tag}:${msg}`) },
+    traceId: "cid-1",
+  });
+  assert.equal(second.acquired, true);
+  assert.ok(waits.some((w) => w.startsWith("TURN_GUARD:duplicate waiting cid=cid-1")));
+});
+
+test("falls back to the blocked result after the timeout", async () => {
+  clearTurnExecutionsForTesting();
+  const first = acquireTurnExecution("k");
+  assert.equal(first.acquired, true);
+  const started = Date.now();
+  const second = await acquireTurnExecutionWithWait("k", { timeoutMs: 150, pollMs: 40 });
+  assert.equal(second.acquired, false);
+  assert.ok(Date.now() - started >= 140);
+  first.release();
+});
+
+test("stops waiting when the caller aborts", async () => {
+  clearTurnExecutionsForTesting();
+  const first = acquireTurnExecution("k");
+  assert.equal(first.acquired, true);
+  const controller = new AbortController();
+  setTimeout(() => controller.abort(), 80);
+  const second = await acquireTurnExecutionWithWait("k", {
+    timeoutMs: 10_000,
+    pollMs: 40,
+    signal: controller.signal,
+  });
+  assert.equal(second.acquired, false);
+  first.release();
+});