From 6768b14b54eba451800345a3224bc0eac46b8d6a Mon Sep 17 00:00:00 2001 From: initguru Date: Fri, 18 Sep 2026 23:58:57 +0900 Subject: [PATCH] fix(chatcore): block duplicate turn execution with 409 turn_in_progress (#12912) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Co-authored-by: initguru --- .../fixes/12912-turn-execution-guard-409.md | 1 + open-sse/handlers/chatCore.ts | 98 ++++++++++++------- .../handlers/chatCore/turnExecutionGuard.ts | 80 +++++++++++++++ open-sse/utils/error.ts | 1 + tests/unit/turn-execution-guard.test.ts | 76 ++++++++++++++ 5 files changed, 221 insertions(+), 35 deletions(-) create mode 100644 changelog.d/fixes/12912-turn-execution-guard-409.md create mode 100644 open-sse/handlers/chatCore/turnExecutionGuard.ts create mode 100644 tests/unit/turn-execution-guard.test.ts diff --git a/changelog.d/fixes/12912-turn-execution-guard-409.md b/changelog.d/fixes/12912-turn-execution-guard-409.md new file mode 100644 index 0000000000..63c2d66526 --- /dev/null +++ b/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 diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 9447082f82..e6ed8d7da7 100644 --- a/open-sse/handlers/chatCore.ts +++ b/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; diff --git a/open-sse/handlers/chatCore/turnExecutionGuard.ts b/open-sse/handlers/chatCore/turnExecutionGuard.ts new file mode 100644 index 0000000000..d2adc1d61c --- /dev/null +++ b/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(); + +/** + * 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(); +} diff --git a/open-sse/utils/error.ts b/open-sse/utils/error.ts index 2f3b3ee868..11ef8972ef 100644 --- a/open-sse/utils/error.ts +++ b/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", diff --git a/tests/unit/turn-execution-guard.test.ts b/tests/unit/turn-execution-guard.test.ts new file mode 100644 index 0000000000..cc9730545c --- /dev/null +++ b/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)) + ); +});