fix(chatcore): block duplicate turn execution with 409 turn_in_progress (#12912)

* fix(chatcore): block duplicate turn execution with 409 turn_in_progress

* test(sse): align turn-execution-guard 409 body expectation with buildErrorBody reason field

* fix(errors): preserve duplicate turn classification

* test(turn-execution-guard): assert ageMs range instead of exact 0

Comparing ageMs to an exact 0 was flaky under real scheduling
latency between the two synchronous calls in the test.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* docs(changelog): add fragment for turn execution guard fix (#12912)

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* chore(quality): rebaseline chatCore.ts file-size for the turn-execution guard

The guard logic lives in the new open-sse/handlers/chatCore/turnExecutionGuard.ts
leaf; what grows chatCore.ts is the irreducible call-site wiring at the single
execution chokepoint (acquire, the 409 turn_in_progress early return, the
release/handoff bookkeeping and the try wrapper that scopes it).

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* fix(sse): keep endpointPath outside the turn-guard try so the failure-usage closure can reach it

The try/finally that scopes the duplicate-turn guard block-scoped the
resolveChatCoreRequestFormat destructuring, but persistFailureUsage is defined
above the try and closes over endpointPath — every failure-usage write would
have thrown ReferenceError. Moved the destructuring above the guard (it is a
pure derivation from the request, so nothing else changes) and narrowed the
acquire result with an explicit === false, which the workspace tsconfig
(strict: false) needs to see the non-acquired arm's retryCount/ageMs.

check:open-sse-typecheck goes from 4 errors to 0; typecheck:core, eslint,
prettier and the PR's 4 turn-execution-guard tests stay green.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

---------

Co-authored-by: Jihyun Son <jihyun.son@sk.com>
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
Co-authored-by: initguru <initguru@users.noreply.github.com>

changelog.d/fixes/12912-turn-execution-guard-409.md

@@ -0,0 +1 @@
+- **fix(chatcore):** block a client's own duplicate retry (same idempotency key) from opening a second upstream turn while the first is still in flight, returning `409 turn_in_progress` instead of wasting quota on a redundant execution

open-sse/handlers/chatCore.ts

@@ -27,6 +27,7 @@ export {
   relocateDirectiveOnlyMessages,
 } from "./chatCore/claudeSystemRole.ts";
 import { checkIdempotencyCache } from "./chatCore/idempotency.ts";
+import { acquireTurnExecution, createTurnInProgressResult } from "./chatCore/turnExecutionGuard.ts";
 import { checkSemanticCache } from "./chatCore/semanticCache.ts";
 import { checkLifecycle, resolveLifecycle } from "./chatCore/modelLifecyclePolicy.ts";
 import {
@@ -704,6 +705,19 @@ export async function handleChatCore({
     transport?: string,
     failureDetail?: string
   ): void => recordKeyHealthStatusFor(status, creds, log, transport, failureDetail);
+  // Endpoint/format resolution extracted to chatCore/requestFormat.ts (#3501); pure derivation
+  // from the request. OUTSIDE the try below — persistFailureUsage closes over endpointPath.
+  const {
+    endpointPath,
+    sourceFormat,
+    isResponsesEndpoint,
+    nativeCodexPassthrough,
+    nativeXaiResponsesPassthrough,
+    isDroidCLI,
+    isOpencodeClient,
+    copilotCompatibleReasoning,
+    clientResponseFormat,
+  } = resolveChatCoreRequestFormat({ clientRawRequest, body, provider, userAgent });
   // ── Phase 9.2: Idempotency check ──
   // Resolve the idempotency key once here and reuse it at the Phase 9.2 save site below,
   // rather than re-deriving it. (#3821-review LEDGER-6)
@@ -722,24 +736,27 @@ export async function handleChatCore({
   if (idempotencyHit) {
     return idempotencyHit;
   }
-  // T07: Inject connectionId into credentials so executors can rotate API keys
+
+  const turnExecution = acquireTurnExecution(idempotencyKey);
+  if (turnExecution.acquired === false) {
+    const duplicate = createTurnInProgressResult(turnExecution.retryCount);
+    log?.warn?.(
+      "TURN_GUARD",
+      `duplicate blocked cid=${traceId} retry=${turnExecution.retryCount} ageMs=${turnExecution.ageMs}`
+    );
+    return duplicate.result;
+  }
+  const releaseTurnExecution = turnExecution.release;
+  let turnExecutionHandedOffToStream = false;
+
+  // Preserve chatCore's canonical formatting while the guarded body remains byte-stable.
+  // prettier-ignore
+  try {
+    // T07: Inject connectionId into credentials so executors can rotate API keys
   // using providerSpecificData.extraApiKeys (API Key Round-Robin feature)
   if (connectionId && credentials && !credentials.connectionId) {
     credentials.connectionId = connectionId;
   }
-  // Endpoint/format resolution extracted to chatCore/requestFormat.ts (#3501); pure derivation
-  // from the inbound request, destructured so every downstream use stays byte-identical.
-  const {
-    endpointPath,
-    sourceFormat,
-    isResponsesEndpoint,
-    nativeCodexPassthrough,
-    nativeXaiResponsesPassthrough,
-    isDroidCLI,
-    isOpencodeClient,
-    copilotCompatibleReasoning,
-    clientResponseFormat,
-  } = resolveChatCoreRequestFormat({ clientRawRequest, body, provider, userAgent });
   let clientRequestedResponsesStream = false;
   const nativeOpenAICompatibleResponsesPassthrough =
     shouldUseNativeOpenAICompatibleResponsesPassthrough({
@@ -6157,23 +6174,27 @@ export async function handleChatCore({
     );
   }
 
-  const finalStream = assembleStreamingPipeline({
-    providerResponse,
-    transformStream,
-    streamController,
-    createPiiTransform,
-    clientRawRequestHeaders: clientRawRequest?.headers,
-    clientResponseFormat,
-    echoModel,
-    responseHeaders,
-    // Same adaptive budget the pre-handoff readiness gate above just used —
-    // reasoning models that legitimately take a while to say anything keep
-    // that same patience for their first REAL content, not just their first
-    // lifecycle frame. See pipeWithDisconnect's own doc comment.
-    contentStallTimeoutMs: streamReadinessPolicy.timeoutMs,
-  });
+    const finalStream = assembleStreamingPipeline({
+      providerResponse,
+      transformStream,
+      streamController,
+      createPiiTransform,
+      clientRawRequestHeaders: clientRawRequest?.headers,
+      clientResponseFormat,
+      echoModel,
+      responseHeaders,
+      // Same adaptive budget the pre-handoff readiness gate above just used —
+      // reasoning models that legitimately take a while to say anything keep
+      // that same patience for their first REAL content, not just their first
+      // lifecycle frame. See pipeWithDisconnect's own doc comment.
+      contentStallTimeoutMs: streamReadinessPolicy.timeoutMs,
+    });
+    const clientFacingStream = wrapReadableStreamWithFinalize(
+      finalStream,
+      releaseTurnExecution
+    );
 
-  // ── Gamification event (fire-and-forget) ──
+    // ── Gamification event (fire-and-forget) ──
   await emitRequestGamificationEvent({ apiKeyId: apiKeyInfo?.id, model, provider });
 
   // ── Plugin onResponse hook (fire-and-forget) ──
@@ -6187,12 +6208,19 @@ export async function handleChatCore({
     response: { status: 200, streamed: true },
   });
 
-  return {
-    success: true,
-    response: new Response(finalStream, {
+    const response = new Response(clientFacingStream, {
       headers: responseHeaders,
-    }),
-  };
+    });
+    turnExecutionHandedOffToStream = true;
+    return {
+      success: true,
+      response,
+    };
+  } finally {
+    if (!turnExecutionHandedOffToStream) {
+      releaseTurnExecution();
+    }
+  }
 }
 export function isTokenExpiringSoon(expiresAt, bufferMs = 5 * 60 * 1000) {
   if (!expiresAt) return false;

open-sse/handlers/chatCore/turnExecutionGuard.ts

@@ -0,0 +1,80 @@
+import { buildErrorBody } from "../../utils/error.ts";
+
+type ActiveTurn = {
+  startedAt: number;
+  retryCount: number;
+};
+
+export type TurnExecutionAcquireResult =
+  | { acquired: true; release: () => void }
+  | { acquired: false; retryCount: number; ageMs: number; release: () => void };
+
+export const TURN_IN_PROGRESS_MESSAGE = "An identical request is already in progress";
+
+export function createTurnInProgressResult(retryCount: number) {
+  const body = buildErrorBody(409, TURN_IN_PROGRESS_MESSAGE, undefined, {
+    type: "turn_in_progress",
+    code: "turn_in_progress",
+  });
+  return {
+    body,
+    result: {
+      success: false as const,
+      status: 409,
+      error: TURN_IN_PROGRESS_MESSAGE,
+      errorType: "turn_in_progress",
+      errorCode: "turn_in_progress",
+      response: new Response(JSON.stringify(body), {
+        status: 409,
+        headers: {
+          "Content-Type": "application/json",
+          "Retry-After": "1",
+          "X-OmniRoute-Turn-Retry": String(retryCount),
+        },
+      }),
+    },
+  };
+}
+
+const activeTurns = new Map<string, ActiveTurn>();
+
+/**
+ * Ensures a client retry carrying the same effective idempotency key cannot
+ * open another upstream execution while the original turn is still streaming.
+ */
+export function acquireTurnExecution(key: string | null): TurnExecutionAcquireResult {
+  if (!key) return { acquired: true, release: () => {} };
+
+  const existing = activeTurns.get(key);
+  if (existing) {
+    existing.retryCount += 1;
+    return {
+      acquired: false,
+      retryCount: existing.retryCount,
+      ageMs: Math.max(0, Date.now() - existing.startedAt),
+      release: () => {},
+    };
+  }
+
+  const entry: ActiveTurn = { startedAt: Date.now(), retryCount: 0 };
+  activeTurns.set(key, entry);
+  let released = false;
+  return {
+    acquired: true,
+    release: () => {
+      if (released) return;
+      released = true;
+      if (activeTurns.get(key) === entry) activeTurns.delete(key);
+    },
+  };
+}
+
+export function getTurnExecutionSnapshot(key: string) {
+  const active = activeTurns.get(key);
+  if (!active) return null;
+  return { retryCount: active.retryCount, ageMs: Math.max(0, Date.now() - active.startedAt) };
+}
+
+export function clearTurnExecutionsForTesting(): void {
+  activeTurns.clear();
+}

open-sse/utils/error.ts

@@ -277,6 +277,7 @@ const SAFE_PUBLIC_ERROR_IDENTIFIERS = new Set([
   "token_required",
   "tool_calling_not_supported",
   "tools",
+  "turn_in_progress",
   "uc_auth_error",
   "uc_generation_failed",
   "uc_message_limit_exceeded",

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

@@ -0,0 +1,76 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import {
+  acquireTurnExecution,
+  clearTurnExecutionsForTesting,
+  createTurnInProgressResult,
+  getTurnExecutionSnapshot,
+} from "../../open-sse/handlers/chatCore/turnExecutionGuard.ts";
+
+test.afterEach(() => {
+  clearTurnExecutionsForTesting();
+});
+
+test("turn execution guard rejects a concurrent duplicate and reports its retry count", () => {
+  const first = acquireTurnExecution("turn-guard-key");
+  const duplicate = acquireTurnExecution("turn-guard-key");
+
+  assert.equal(first.acquired, true);
+  assert.equal(duplicate.acquired, false);
+  assert.equal(duplicate.retryCount, 1);
+  const snapshot = getTurnExecutionSnapshot("turn-guard-key");
+  assert.equal(snapshot?.retryCount, 1);
+  assert.ok(
+    snapshot && snapshot.ageMs >= 0 && snapshot.ageMs < 1000,
+    `unexpected ageMs: ${snapshot?.ageMs}`
+  );
+
+  first.release();
+  const next = acquireTurnExecution("turn-guard-key");
+  assert.equal(next.acquired, true);
+  next.release();
+});
+
+test("turn execution guard release is idempotent", () => {
+  const first = acquireTurnExecution("turn-guard-release");
+  assert.equal(first.acquired, true);
+
+  first.release();
+  first.release();
+
+  const next = acquireTurnExecution("turn-guard-release");
+  assert.equal(next.acquired, true);
+  next.release();
+});
+
+test("turn execution guard bypasses missing idempotency keys", () => {
+  const first = acquireTurnExecution(null);
+  const second = acquireTurnExecution(null);
+
+  assert.equal(first.acquired, true);
+  assert.equal(second.acquired, true);
+  first.release();
+  second.release();
+});
+
+test("duplicate turn result uses a sanitized 409 body and retry headers", async () => {
+  const duplicate = createTurnInProgressResult(3);
+
+  assert.deepEqual(duplicate.body, {
+    error: {
+      message: "An identical request is already in progress",
+      type: "turn_in_progress",
+      code: "turn_in_progress",
+      reason: undefined,
+    },
+  });
+  assert.equal(duplicate.result.status, 409);
+  assert.equal(duplicate.result.errorType, "turn_in_progress");
+  assert.equal(duplicate.result.errorCode, "turn_in_progress");
+  assert.equal(duplicate.result.response.headers.get("Retry-After"), "1");
+  assert.equal(duplicate.result.response.headers.get("X-OmniRoute-Turn-Retry"), "3");
+  assert.deepEqual(
+    await duplicate.result.response.json(),
+    JSON.parse(JSON.stringify(duplicate.body))
+  );
+});