From ca816cd163a47621f0022ff241971281fa58663f Mon Sep 17 00:00:00 2001 From: olddonkey Date: Fri, 21 Aug 2026 02:03:09 -0700 Subject: [PATCH 1/5] fix(responses): scope reasoning replay by conversation, not just parent thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The serving-identity record was keyed only on `x-codex-parent-thread-id`. Without that header there was no scope at all, so the record could never be written or compared: every turn stayed permanently cold, the deterministic pre-flight never fired, and each turn fell through to the opaque-blob recovery — one extra full upload of the transcript, every turn. Measured on live traffic. Across 95 xAI conversations, 70 recoveries occurred and 67 of them were in two conversations: f4be51de 86 requests 55 recoveries c14e85a7 66 requests 12 recoveries e925d065 165 requests 1 recovery <- healthy: one cold first turn Both outliers are conversations where the backend was switched mid-session, so their transcripts permanently carry foreign-minted reasoning blobs replayed on every later turn. An instrumented build showed those requests carrying no client thread id, which is why the record never warmed up. Those turns were ~150k input tokens each, sent twice. The recovery was working as designed — without it the turns would fail outright. The defect is that the deterministic path was structurally unavailable to them, so the recovery paid full price every turn instead of once. `conversationIdFromResponsesRequest` already resolves a conversation identity for the request log through a four-level fallback, so reuse it as the replay scope key when the header is absent. `_clientThreadId` is untouched: it remains the routing and continuation identity, and the header path is byte-for-byte unchanged. The scope is shared with the process-local raw-reasoning replay and the durable thought-signature replay. Widening is safe for both because they key additionally by provider, destination, adapter, model and credential, so a conversation namespace only narrows what they already isolate — and a fallback that yields no identity still produces no scope, preserving today's keep-the- blobs behaviour. Pinned by a three-turn headerless regression asserting sendCount [2, 1, 1]: recover once, then strip pre-flight. That sequence is the entire point. Co-Authored-By: Claude Fable 5 (cherry picked from commit 22375cf980ee7990f36f6d6c9d231966ff84102a) --- src/responses/reasoning-replay-cache.ts | 6 +- src/server/responses/core.ts | 22 ++- src/types/request.ts | 6 +- tests/responses-opaque-blob-recovery.test.ts | 137 ++++++++++++++++++- 4 files changed, 159 insertions(+), 12 deletions(-) diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts index 4b2d6d9166..e7551929bf 100644 --- a/src/responses/reasoning-replay-cache.ts +++ b/src/responses/reasoning-replay-cache.ts @@ -143,10 +143,10 @@ function servingIdentityFor( } /** - * Compare this request's route with the last successfully serving route for its client thread. + * Compare this request's route with the last successfully serving route for its conversation. * A live mismatch means replayed opaque reasoning was minted by another backend and must not be * forwarded to this one. Comparison deliberately does not refresh or replace the recorded route: - * a failed candidate request did not serve the thread. + * a failed candidate request did not serve the conversation. * * Serving provenance uses restart-stable destination and credential dimensions so token * generations and other volatile credential material cannot create false route changes. Missing @@ -164,7 +164,7 @@ export function reasoningReplayServingIdentityChanged( return previous !== undefined && previous.identity !== current.identity; } -/** Record the route only after it has successfully served the client thread. */ +/** Record the route only after it has successfully served the conversation. */ export function commitReasoningReplayServingIdentity( scope: OcxReasoningReplayScopeRef | undefined, ): void { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index fca48dd90c..205519868e 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -2185,7 +2185,6 @@ async function handleResponsesInner( if (providerContinuationCandidate) parsed._providerContinuationCandidate = providerContinuationCandidate; if (inboundClientThreadId) { parsed._clientThreadId = inboundClientThreadId; - parsed._reasoningReplayScope = { clientThreadId: inboundClientThreadId }; } } catch (err) { if (isTranslatorBudgetExceededError(err)) { @@ -2200,15 +2199,24 @@ async function handleResponsesInner( ...(force ? { force: true } : {}), ...(parsed._clientThreadId ? { clientThreadId: parsed._clientThreadId } : {}), }); + const resolvedConversationId = conversationIdFromResponsesRequest({ + clientThreadId: parsed._clientThreadId, + sessionIdHeader: sessionIdHeaderFromRequest(req.headers), + threadIdHeader: req.headers.get("thread-id"), + cursorConversationId: parsed._cursorConversationId, + }); + // `_clientThreadId` remains the routing/continuation identity supplied by Codex. Replay state + // only needs a conversation namespace, so headerless callers may use the same opaque fallback + // already resolved for request logs. Keep the raw parent-thread key when present so that path is + // byte-for-byte unchanged. + const reasoningReplayConversationId = parsed._clientThreadId ?? resolvedConversationId; + if (reasoningReplayConversationId) { + parsed._reasoningReplayScope = { clientThreadId: reasoningReplayConversationId }; + } // Prefer a pre-populated id (routed Claude) over Responses headers that may be // absent or synthetically injected (session_id from prompt_cache_key). if (!logCtx.conversationId) { - logCtx.conversationId = conversationIdFromResponsesRequest({ - clientThreadId: parsed._clientThreadId, - sessionIdHeader: sessionIdHeaderFromRequest(req.headers), - threadIdHeader: req.headers.get("thread-id"), - cursorConversationId: parsed._cursorConversationId, - }); + logCtx.conversationId = resolvedConversationId; } logCtx.requestedModel = parsed.modelId; logCtx.requestedEffort = parsed.options.reasoning; diff --git a/src/types/request.ts b/src/types/request.ts index c14bbf3085..7aed7a8fa5 100644 --- a/src/types/request.ts +++ b/src/types/request.ts @@ -30,6 +30,10 @@ export interface OcxReasoningReplayIdentity { * the holder, so late tool-call cache writes see the active physical identity. */ export interface OcxReasoningReplayScopeRef { + /** + * Conversation namespace for replay state. Historically this was always the Codex parent-thread + * id; headerless Responses callers use an opaque session/thread/Cursor conversation fallback. + */ readonly clientThreadId: string; current?: Readonly; } @@ -63,7 +67,7 @@ export interface OcxParsedRequest { _cursorConversationId?: string; /** Stable upstream client thread identity, used only to derive provider-scoped continuation ids. */ _clientThreadId?: string; - /** Provider/account/model-bound namespace for process-local raw-reasoning replay. */ + /** Conversation/provider/account/model-bound namespace for reasoning replay state. */ _reasoningReplayScope?: OcxReasoningReplayScopeRef; /** * Set by bindRouteReasoningReplayScope after a proven serving-identity change, or by diff --git a/tests/responses-opaque-blob-recovery.test.ts b/tests/responses-opaque-blob-recovery.test.ts index e6ee2d5775..29fd496b6b 100644 --- a/tests/responses-opaque-blob-recovery.test.ts +++ b/tests/responses-opaque-blob-recovery.test.ts @@ -112,11 +112,20 @@ function config(): OcxConfig { } function request(provider = "first", threadId = "thread-opaque-recovery"): Request { + return requestWithIdentityHeaders(provider, { + "x-codex-parent-thread-id": threadId, + }); +} + +function requestWithIdentityHeaders( + provider = "first", + identityHeaders: Record = {}, +): Request { return new Request("http://localhost/v1/responses", { method: "POST", headers: { "content-type": "application/json", - "x-codex-parent-thread-id": threadId, + ...identityHeaders, }, body: JSON.stringify({ model: `${provider}/model-a`, @@ -508,6 +517,132 @@ describe("opaque blob recovery through /v1/responses", () => { }); describe("reasoning replay serving identity commit through /v1/responses", () => { + test("a stable session identity records and detects a serving-identity change without a parent-thread header", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + for (const provider of ["first", "second"]) { + const response = await handleResponses( + requestWithIdentityHeaders(provider, { session_id: "session-without-parent-thread" }), + config(), + { model: "", provider: "" }, + ); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound).toHaveLength(2); + expect(hasBlob(outbound[0]!)).toBe(true); + expect(hasBlob(outbound[1]!)).toBe(false); + }); + + test("foreign blobs recover once, then alternating routes strip pre-flight on later headerless turns", async () => { + const outbound: Array> = []; + const turns: RequestLogContext[] = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as Record; + outbound.push(body); + if (outbound.length === 1 && hasBlob(body)) return rejection(XAI_DECODE_ERROR); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + for (const provider of ["first", "second", "first"]) { + const logCtx: RequestLogContext = { model: "", provider: "" }; + turns.push(logCtx); + const response = await handleResponses( + requestWithIdentityHeaders(provider, { session_id: "three-turn-headerless-session" }), + config(), + logCtx, + ); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound).toHaveLength(4); + expect(outbound.map(hasBlob)).toEqual([true, false, false, false]); + expect(turns.map(turn => turn.activeAttempt?.sendCount)).toEqual([2, 1, 1]); + expect(turns.map(turn => turn.activeAttempt?.recoveryKinds)).toEqual([ + ["opaque-blob-rejection"], + [], + [], + ]); + }); + + test("the parent-thread header remains authoritative over fallback identities", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + const cases = [ + ["first", "parent-thread-a", "shared-session"], + ["second", "parent-thread-a", "different-session"], + ["second", "parent-thread-b", "shared-session"], + ] as const; + for (const [provider, parentThreadId, sessionId] of cases) { + const response = await handleResponses(requestWithIdentityHeaders(provider, { + "x-codex-parent-thread-id": parentThreadId, + session_id: sessionId, + }), config(), { model: "", provider: "" }); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound).toHaveLength(3); + expect(outbound.map(hasBlob)).toEqual([true, false, true]); + }); + + test("requests without any usable conversation identity keep blobs and record nothing", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + for (const provider of ["first", "second"]) { + const response = await handleResponses( + requestWithIdentityHeaders(provider), + config(), + { model: "", provider: "" }, + ); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound).toHaveLength(2); + expect(outbound.map(hasBlob)).toEqual([true, true]); + }); + + test("distinct fallback conversation identities never share a serving record", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + const cases = [ + ["first", "conversation-a"], + ["second", "conversation-b"], + ["second", "conversation-a"], + ] as const; + for (const [provider, sessionId] of cases) { + const response = await handleResponses( + requestWithIdentityHeaders(provider, { session_id: sessionId }), + config(), + { model: "", provider: "" }, + ); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound).toHaveLength(3); + expect(outbound.map(hasBlob)).toEqual([true, true, false]); + }); + test("a failed A-to-B turn does not commit B, so the next B retry still strips A-minted blobs", async () => { const outbound: Array> = []; globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { From 7292b4e4587bb91eb8d585e8b44ff0c163574216 Mon Sep 17 00:00:00 2001 From: olddonkey Date: Fri, 21 Aug 2026 08:03:41 -0700 Subject: [PATCH 2/5] fix(responses): remember a proven opaque-blob rejection per destination MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The serving-identity record tracks which destination served the previous turn. That is the right signal for detecting a switch and the wrong one for what actually costs money, because foreign blobs stay in the client transcript forever while the switch happens only once. Measured on the deployed build — three consecutive headerless turns replaying a grok-minted blob to gpt-5.6-sol: turn 1 sends=1 recovery=[] pre-flight strips, one send turn 2 sends=2 recovery=[opaque-blob-rejection] record now says sol == sol, turn 3 sends=2 recovery=[opaque-blob-rejection] no strip, upstream rejects After the first turn commits the new destination every later comparison returns "same identity", so the pre-flight stops stripping while the grok-minted blob is still in the replayed history. Each of those turns paid a full extra upload. This is the production pathology: 86 requests / 55 recoveries and 66 / 12 in the two conversations where the backend was switched mid-session, against 165 / 1 for a healthy one, at ~150k input tokens a send. When a recovery succeeds the upstream has just proven this conversation's replayed opaque state is unusable for that destination. Remember it and pre-strip instead of rediscovering it once per turn. The memo is keyed by conversation **and** durable serving identity. Keyed by conversation alone it would strip the original destination's own valid blobs the moment the user switched back — a silent, permanent quality regression with no error to notice. It is recorded only when the blobless resend actually succeeded, so a resend that also failed teaches nothing. TTL is five minutes against the serving record's hour, and the asymmetry is deliberate: a stale memo silently degrades reasoning, while an expired one costs a single visible recovery round trip that re-establishes it. An earlier attempt at this test alternated destinations between turns, which passes for the wrong reason — the identity changes every turn, so the ordinary switch detection fires and the memo is never exercised. The regression now holds the destination constant and asserts sendCount [2, 1, 1], plus the switch-back case, a failed resend recording nothing, and expiry rechecking once before settling. Co-Authored-By: Claude Fable 5 (cherry picked from commit fe8be1ac4d00d855e98393642e1f2e796b21ca2f) --- src/responses/reasoning-replay-cache.ts | 78 +++++++++++ src/server/responses/core.ts | 16 +++ tests/reasoning-replay-identity.test.ts | 47 +++++++ tests/responses-opaque-blob-recovery.test.ts | 131 ++++++++++++++++++- 4 files changed, 268 insertions(+), 4 deletions(-) diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts index e7551929bf..f28dcc072e 100644 --- a/src/responses/reasoning-replay-cache.ts +++ b/src/responses/reasoning-replay-cache.ts @@ -29,6 +29,7 @@ import type { const MAX_ENTRIES = 64; const MAX_TOTAL_BYTES = 256 * 1024; const TTL_MS = 60 * 60 * 1000; +const OPAQUE_BLOB_REJECTION_TTL_MS = 5 * 60 * 1000; const replayIdentityKey = randomBytes(32); const CREDENTIAL_HEADER_NAMES = new Set([ "authorization", @@ -58,10 +59,17 @@ interface ServingIdentityEntry { at: number; } +interface OpaqueBlobRejectionEntry { + bytes: number; + at: number; +} + const entries = new Map(); const servingIdentities = new Map(); +const opaqueBlobRejections = new Map(); let totalBytes = 0; let servingIdentityTotalBytes = 0; +let opaqueBlobRejectionTotalBytes = 0; let clockForTests: (() => number) | null = null; const now = (): number => clockForTests?.() ?? Date.now(); @@ -142,6 +150,26 @@ function servingIdentityFor( return { threadId, identity: JSON.stringify(identityTuple) }; } +function opaqueBlobRejectionKeyFor( + scope: OcxReasoningReplayScopeRef | undefined, +): string | undefined { + const current = servingIdentityFor(scope); + return current ? JSON.stringify([current.threadId, current.identity]) : undefined; +} + +function deleteOpaqueBlobRejection(key: string): void { + const entry = opaqueBlobRejections.get(key); + if (!entry) return; + opaqueBlobRejections.delete(key); + opaqueBlobRejectionTotalBytes -= entry.bytes; +} + +function sweepExpiredOpaqueBlobRejections(at: number): void { + for (const [key, entry] of opaqueBlobRejections) { + if (at - entry.at >= OPAQUE_BLOB_REJECTION_TTL_MS) deleteOpaqueBlobRejection(key); + } +} + /** * Compare this request's route with the last successfully serving route for its conversation. * A live mismatch means replayed opaque reasoning was minted by another backend and must not be @@ -200,6 +228,54 @@ export function commitReasoningReplayServingIdentity( } } +/** + * Whether this exact conversation and durable serving identity previously rejected opaque replay. + * + * The five serving dimensions deliberately match the serving record. Missing durable destination + * or credential identity is unknown and never falls back to process-local dimensions. The five + * minute TTL is shorter than the serving record's hour: a stale memo silently degrades reasoning, + * while expiry costs one visible recovery round trip and can safely re-establish the memo. + */ +export function reasoningReplayOpaqueBlobRejectionMemoized( + scope: OcxReasoningReplayScopeRef | undefined, +): boolean { + const key = opaqueBlobRejectionKeyFor(scope); + if (!key) return false; + const at = now(); + sweepExpiredOpaqueBlobRejections(at); + return opaqueBlobRejections.has(key); +} + +/** Record only after a blobless retry succeeded for this durable serving identity. */ +export function rememberReasoningReplayOpaqueBlobRejection( + scope: OcxReasoningReplayScopeRef | undefined, +): void { + const key = opaqueBlobRejectionKeyFor(scope); + if (!key) return; + const bytes = Buffer.byteLength(key, "utf8"); + if (bytes > MAX_TOTAL_BYTES) return; + const at = now(); + sweepExpiredOpaqueBlobRejections(at); + if (opaqueBlobRejections.has(key)) deleteOpaqueBlobRejection(key); + opaqueBlobRejections.set(key, { bytes, at }); + opaqueBlobRejectionTotalBytes += bytes; + while ( + (opaqueBlobRejectionTotalBytes > MAX_TOTAL_BYTES || opaqueBlobRejections.size > MAX_ENTRIES) + && opaqueBlobRejections.size > 1 + ) { + let oldestKey: string | undefined; + let oldestAt = Infinity; + for (const [candidateKey, entry] of opaqueBlobRejections) { + if (entry.at < oldestAt) { + oldestAt = entry.at; + oldestKey = candidateKey; + } + } + if (oldestKey === undefined) break; + deleteOpaqueBlobRejection(oldestKey); + } +} + function processLocalIdentity(domain: string, material: string): string { return createHmac("sha256", replayIdentityKey) .update(domain) @@ -420,7 +496,9 @@ export function peekReasoningForCall( export function clearReasoningReplayCacheForTests(clock?: (() => number) | null): void { entries.clear(); servingIdentities.clear(); + opaqueBlobRejections.clear(); totalBytes = 0; servingIdentityTotalBytes = 0; + opaqueBlobRejectionTotalBytes = 0; clockForTests = clock ?? null; } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 205519868e..15fadeaba9 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -22,8 +22,10 @@ import { durableReplayDestinationIdentity, durableReplayCredentialIdentity, reasoningReplayKeyCredentialIdentity, + reasoningReplayOpaqueBlobRejectionMemoized, reasoningReplayOAuthCredentialIdentity, reasoningReplayServingIdentityChanged, + rememberReasoningReplayOpaqueBlobRejection, } from "../../responses/reasoning-replay-cache"; import { awaitThoughtSignatureDurability, thoughtSignatureReplaySalt } from "../../responses/thought-signature-replay"; import { buildCompactV1Output, COMPACT_PROMPT, decodeCompactionSummary, extractCompactUserMessages } from "../../responses/compaction"; @@ -514,6 +516,9 @@ function bindRouteReasoningReplayScope(args: { if (reasoningReplayServingIdentityChanged(parsed._reasoningReplayScope)) { parsed._stripReasoningEncryptedContent = true; } + if (reasoningReplayOpaqueBlobRejectionMemoized(parsed._reasoningReplayScope)) { + parsed._stripReasoningEncryptedContent = true; + } bindProviderContinuationForRoute(parsed, continuationOwner); } @@ -667,9 +672,20 @@ async function attemptOpaqueBlobRecovery( } args.guard.attempted = true; + const rejectedScope = args.parsed._reasoningReplayScope + ? { + clientThreadId: args.parsed._reasoningReplayScope.clientThreadId, + ...(args.parsed._reasoningReplayScope.current + ? { current: { ...args.parsed._reasoningReplayScope.current } } + : {}), + } + : undefined; prepareOpaqueBlobRecovery(args.parsed); try { void args.response.body?.cancel().catch(() => {}); } catch { /* already consumed/closed */ } const result = await rebuild("opaque-blob-rejection"); + if (!("failed" in result) && result.ok) { + rememberReasoningReplayOpaqueBlobRejection(rejectedScope); + } return "failed" in result ? { kind: "failed", response: result.failed } : { kind: "recovered", response: result }; diff --git a/tests/reasoning-replay-identity.test.ts b/tests/reasoning-replay-identity.test.ts index 834c23b5ea..9737895b6a 100644 --- a/tests/reasoning-replay-identity.test.ts +++ b/tests/reasoning-replay-identity.test.ts @@ -9,9 +9,11 @@ import { reasoningReplayCredentialIdentity, reasoningReplayDestinationIdentity, reasoningReplayKeyCredentialIdentity, + reasoningReplayOpaqueBlobRejectionMemoized, reasoningReplayOAuthCredentialIdentity, reasoningReplayServingIdentityChanged, rememberReasoningForCall, + rememberReasoningReplayOpaqueBlobRejection, } from "../src/responses/reasoning-replay-cache"; import type { AdapterEvent, OcxReasoningReplayScopeRef } from "../src/types"; @@ -142,6 +144,51 @@ describe("reasoning replay provider and credential identity", () => { })).toBe(false); }); + test("opaque-blob rejection memos use the durable serving identity and refuse incomplete scopes", () => { + const rejected = scope({ modelId: "destination-b-model" }); + rememberReasoningReplayOpaqueBlobRejection(rejected); + expect(reasoningReplayOpaqueBlobRejectionMemoized(rejected)).toBe(true); + expect(reasoningReplayOpaqueBlobRejectionMemoized(scope({ modelId: "destination-a-model" }))).toBe(false); + expect(reasoningReplayOpaqueBlobRejectionMemoized({ + ...rejected, + clientThreadId: "another-conversation", + })).toBe(false); + + for (const incomplete of [ + scope({ credentialDurableIdentity: undefined }), + scope({ providerDestinationDurableIdentity: undefined }), + { clientThreadId: THREAD }, + ]) { + rememberReasoningReplayOpaqueBlobRejection(incomplete); + expect(reasoningReplayOpaqueBlobRejectionMemoized(incomplete)).toBe(false); + } + }); + + test("opaque-blob rejection memos use the serving record's 64-entry bound", () => { + let clock = 1_000; + clearReasoningReplayCacheForTests(() => clock); + for (let i = 0; i < 65; i++) { + rememberReasoningReplayOpaqueBlobRejection({ + ...scope(), + clientThreadId: `memo-thread-${i}`, + }); + clock += 1; + } + + expect(reasoningReplayOpaqueBlobRejectionMemoized({ + ...scope(), + clientThreadId: "memo-thread-0", + })).toBe(false); + expect(reasoningReplayOpaqueBlobRejectionMemoized({ + ...scope(), + clientThreadId: "memo-thread-1", + })).toBe(true); + expect(reasoningReplayOpaqueBlobRejectionMemoized({ + ...scope(), + clientThreadId: "memo-thread-64", + })).toBe(true); + }); + test("expired serving identity is unknown rather than a backend change", () => { let clock = 1_000; clearReasoningReplayCacheForTests(() => clock); diff --git a/tests/responses-opaque-blob-recovery.test.ts b/tests/responses-opaque-blob-recovery.test.ts index 29fd496b6b..4f81fc50d8 100644 --- a/tests/responses-opaque-blob-recovery.test.ts +++ b/tests/responses-opaque-blob-recovery.test.ts @@ -539,21 +539,21 @@ describe("reasoning replay serving identity commit through /v1/responses", () => expect(hasBlob(outbound[1]!)).toBe(false); }); - test("foreign blobs recover once, then alternating routes strip pre-flight on later headerless turns", async () => { + test("foreign blobs recover once, then the same destination strips pre-flight on later headerless turns", async () => { const outbound: Array> = []; const turns: RequestLogContext[] = []; globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { const body = JSON.parse(String(init?.body)) as Record; outbound.push(body); - if (outbound.length === 1 && hasBlob(body)) return rejection(XAI_DECODE_ERROR); + if (hasBlob(body)) return rejection(XAI_DECODE_ERROR); return success(`resp-${outbound.length}`); }) as typeof fetch; - for (const provider of ["first", "second", "first"]) { + for (let turn = 0; turn < 3; turn++) { const logCtx: RequestLogContext = { model: "", provider: "" }; turns.push(logCtx); const response = await handleResponses( - requestWithIdentityHeaders(provider, { session_id: "three-turn-headerless-session" }), + requestWithIdentityHeaders("first", { session_id: "three-turn-headerless-session" }), config(), logCtx, ); @@ -571,6 +571,129 @@ describe("reasoning replay serving identity commit through /v1/responses", () => ]); }); + test("a destination rejection memo does not keep stripping after switching back to the blob-minting destination", async () => { + const outbound = new Map>>([ + ["first", []], + ["second", []], + ]); + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const provider = String(input).includes("first.example.test") ? "first" : "second"; + const body = JSON.parse(String(init?.body)) as Record; + outbound.get(provider)!.push(body); + if (provider === "first" && hasBlob(body)) return rejection(XAI_DECODE_ERROR); + return success(`resp-${provider}-${outbound.get(provider)!.length}`); + }) as typeof fetch; + + for (const provider of ["first", "second", "second"]) { + const response = await handleResponses( + requestWithIdentityHeaders(provider, { session_id: "switch-back-identity-session" }), + config(), + { model: "", provider: "" }, + ); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound.get("first")!.map(hasBlob)).toEqual([true, false]); + // The first switch-back turn is still stripped by the unchanged deterministic serving record. + // Once that record commits `second`, `first`'s memo must not suppress `second`'s own good blob. + expect(outbound.get("second")!.map(hasBlob)).toEqual([false, true]); + }); + + test("a failed blobless resend records no memo, so the next turn tries the blob again", async () => { + const outbound: Array> = []; + const turns: RequestLogContext[] = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as Record; + outbound.push(body); + if (hasBlob(body)) return rejection(XAI_DECODE_ERROR); + return new Response(JSON.stringify({ + error: { type: "invalid_request_error", code: "unknown_parameter", message: "retry also failed" }, + }), { + status: 400, + headers: { "content-type": "application/json" }, + }); + }) as typeof fetch; + + for (let turn = 0; turn < 2; turn++) { + const logCtx: RequestLogContext = { model: "", provider: "" }; + turns.push(logCtx); + const response = await handleResponses( + requestWithIdentityHeaders("first", { session_id: "failed-resend-session" }), + config(), + logCtx, + ); + expect(response.status).toBe(400); + await response.text(); + } + + expect(outbound.map(hasBlob)).toEqual([true, false, true, false]); + expect(turns.map(turn => turn.activeAttempt?.sendCount)).toEqual([2, 2]); + expect(turns.map(turn => turn.activeAttempt?.recoveryKinds)).toEqual([ + ["opaque-blob-rejection"], + ["opaque-blob-rejection"], + ]); + }); + + test("an expired rejection memo rechecks once, then settles back to one send", async () => { + let clock = 1_000; + clearReasoningReplayCacheForTests(() => clock); + const outbound: Array> = []; + const turns: RequestLogContext[] = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as Record; + outbound.push(body); + return hasBlob(body) ? rejection(XAI_DECODE_ERROR) : success(`resp-${outbound.length}`); + }) as typeof fetch; + + for (let turn = 0; turn < 4; turn++) { + if (turn === 2) clock += 5 * 60 * 1000 + 1; + const logCtx: RequestLogContext = { model: "", provider: "" }; + turns.push(logCtx); + const response = await handleResponses( + requestWithIdentityHeaders("first", { session_id: "memo-expiry-session" }), + config(), + logCtx, + ); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound.map(hasBlob)).toEqual([true, false, false, true, false, false]); + expect(turns.map(turn => turn.activeAttempt?.sendCount)).toEqual([2, 1, 2, 1]); + expect(turns.map(turn => turn.activeAttempt?.recoveryKinds)).toEqual([ + ["opaque-blob-rejection"], + [], + ["opaque-blob-rejection"], + [], + ]); + }); + + test("a stable conversation without a rejection memo remains unchanged", async () => { + const outbound: Array> = []; + const turns: RequestLogContext[] = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + for (let turn = 0; turn < 3; turn++) { + const logCtx: RequestLogContext = { model: "", provider: "" }; + turns.push(logCtx); + const response = await handleResponses( + requestWithIdentityHeaders("first", { session_id: "no-rejection-memo-session" }), + config(), + logCtx, + ); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound.map(hasBlob)).toEqual([true, true, true]); + expect(turns.map(turn => turn.activeAttempt?.sendCount)).toEqual([1, 1, 1]); + expect(turns.map(turn => turn.activeAttempt?.recoveryKinds)).toEqual([[], [], []]); + }); + test("the parent-thread header remains authoritative over fallback identities", async () => { const outbound: Array> = []; globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { From 4ba0b8d803187a09213608e7c93b7c710561cdb2 Mon Sep 17 00:00:00 2001 From: olddonkey Date: Fri, 21 Aug 2026 12:08:26 -0700 Subject: [PATCH 3/5] fix(responses): keep replay scope as a raw conversation identity Do not reuse the hashed request-log conversation id. Mixed parent-thread and session_id headers that carry the same conversation must hit one serving record, and a shared or synthetic session_id must not coalesce distinct thread or Cursor conversations. --- src/server/request-log-conversation.ts | 30 +++++++++++ src/server/responses/core.ts | 15 ++++-- src/types/request.ts | 3 +- tests/request-log-conversation.test.ts | 46 ++++++++++++++++ tests/responses-opaque-blob-recovery.test.ts | 55 ++++++++++++++++++++ 5 files changed, 143 insertions(+), 6 deletions(-) diff --git a/src/server/request-log-conversation.ts b/src/server/request-log-conversation.ts index 7cef86ab47..3fae35ef1a 100644 --- a/src/server/request-log-conversation.ts +++ b/src/server/request-log-conversation.ts @@ -61,6 +61,36 @@ export function sessionIdHeaderFromRequest(headers: Headers): string | null { return headers.get("session_id") ?? headers.get("session-id"); } +function firstSanitizedConversationId( + ...values: Array +): string | undefined { + for (const value of values) { + const sanitized = sanitizeConversationIdInput(value); + if (sanitized) return sanitized; + } + return undefined; +} + +/** + * Conversation namespace for reasoning replay. Unlike the persisted log id, this stays the raw + * sanitized identity so mixed headers that carry the same conversation still hit one serving + * record. Do not hash, and do not prefer session_id over a true per-conversation thread/Cursor + * identity: session_id can be synthesized from a shared prompt_cache_key. + */ +export function reasoningReplayConversationIdFromResponsesRequest(input: { + clientThreadId?: string; + threadIdHeader?: string | null; + cursorConversationId?: string; + sessionIdHeader?: string | null; +}): string | undefined { + return firstSanitizedConversationId( + input.clientThreadId, + input.threadIdHeader, + input.cursorConversationId, + input.sessionIdHeader, + ); +} + export function conversationIdFromResponsesRequest(input: { clientThreadId?: string; sessionIdHeader?: string | null; diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 15fadeaba9..c95a1148f1 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -225,6 +225,7 @@ import { import { conversationIdFromResponsesRequest, normalizeLogConversationId, + reasoningReplayConversationIdFromResponsesRequest, sessionIdHeaderFromRequest, } from "../request-log-conversation"; import type { AttemptRecoveryKind } from "../../usage/log"; @@ -2221,11 +2222,15 @@ async function handleResponsesInner( threadIdHeader: req.headers.get("thread-id"), cursorConversationId: parsed._cursorConversationId, }); - // `_clientThreadId` remains the routing/continuation identity supplied by Codex. Replay state - // only needs a conversation namespace, so headerless callers may use the same opaque fallback - // already resolved for request logs. Keep the raw parent-thread key when present so that path is - // byte-for-byte unchanged. - const reasoningReplayConversationId = parsed._clientThreadId ?? resolvedConversationId; + // _clientThreadId remains the routing/continuation identity supplied by Codex. Replay state uses + // a dedicated raw conversation namespace so mixed headers that carry the same identity still + // match, and a shared/synthetic session_id cannot coalesce distinct thread/Cursor conversations. + const reasoningReplayConversationId = reasoningReplayConversationIdFromResponsesRequest({ + clientThreadId: parsed._clientThreadId, + threadIdHeader: req.headers.get("thread-id"), + cursorConversationId: parsed._cursorConversationId, + sessionIdHeader: sessionIdHeaderFromRequest(req.headers), + }); if (reasoningReplayConversationId) { parsed._reasoningReplayScope = { clientThreadId: reasoningReplayConversationId }; } diff --git a/src/types/request.ts b/src/types/request.ts index 7aed7a8fa5..56156d8443 100644 --- a/src/types/request.ts +++ b/src/types/request.ts @@ -32,7 +32,8 @@ export interface OcxReasoningReplayIdentity { export interface OcxReasoningReplayScopeRef { /** * Conversation namespace for replay state. Historically this was always the Codex parent-thread - * id; headerless Responses callers use an opaque session/thread/Cursor conversation fallback. + * id; headerless Responses callers use a raw sanitized thread/Cursor/session fallback, never the + * hashed request-log conversation id. */ readonly clientThreadId: string; current?: Readonly; diff --git a/tests/request-log-conversation.test.ts b/tests/request-log-conversation.test.ts index 0a086498f6..624612fc89 100644 --- a/tests/request-log-conversation.test.ts +++ b/tests/request-log-conversation.test.ts @@ -6,6 +6,7 @@ import { conversationIdFromResponsesRequest, matchesLogConversationId, normalizeLogConversationId, + reasoningReplayConversationIdFromResponsesRequest, sessionIdHeaderFromRequest, summarizeConversationLogs, } from "../src/server/request-log-conversation"; @@ -94,6 +95,51 @@ describe("conversationIdFromResponsesRequest", () => { }); }); +describe("reasoningReplayConversationIdFromResponsesRequest", () => { + test("keeps the raw identity instead of the hashed log id", () => { + expect(reasoningReplayConversationIdFromResponsesRequest({ + clientThreadId: "parent-thread", + })).toBe("parent-thread"); + expect(reasoningReplayConversationIdFromResponsesRequest({ + sessionIdHeader: "session", + })).not.toBe(digest32("session")); + }); + + test("prefers parent thread, then thread-id, then cursor, then session_id", () => { + expect(reasoningReplayConversationIdFromResponsesRequest({ + clientThreadId: "parent-thread", + threadIdHeader: "thread", + cursorConversationId: "cursor", + sessionIdHeader: "session", + })).toBe("parent-thread"); + expect(reasoningReplayConversationIdFromResponsesRequest({ + threadIdHeader: "thread", + cursorConversationId: "cursor", + sessionIdHeader: "session", + })).toBe("thread"); + expect(reasoningReplayConversationIdFromResponsesRequest({ + cursorConversationId: "cursor", + sessionIdHeader: "session", + })).toBe("cursor"); + expect(reasoningReplayConversationIdFromResponsesRequest({ + sessionIdHeader: "session", + })).toBe("session"); + }); + + test("skips empty, control-bearing, and overlong fallbacks", () => { + expect(reasoningReplayConversationIdFromResponsesRequest({ + threadIdHeader: " ", + cursorConversationId: "cursor", + })).toBe("cursor"); + expect(reasoningReplayConversationIdFromResponsesRequest({ + sessionIdHeader: "bad\nid", + })).toBeUndefined(); + expect(reasoningReplayConversationIdFromResponsesRequest({ + sessionIdHeader: "x".repeat(4097), + })).toBeUndefined(); + }); +}); + describe("conversationIdFromClaudeMetadata", () => { test("hashes metadata.user_id and ignores Desktop system-hash keys", () => { expect(conversationIdFromClaudeMetadata({ user_id: "session-user" })).toBe(digest32("session-user")); diff --git a/tests/responses-opaque-blob-recovery.test.ts b/tests/responses-opaque-blob-recovery.test.ts index 4f81fc50d8..d161fe5522 100644 --- a/tests/responses-opaque-blob-recovery.test.ts +++ b/tests/responses-opaque-blob-recovery.test.ts @@ -719,6 +719,61 @@ describe("reasoning replay serving identity commit through /v1/responses", () => expect(outbound.map(hasBlob)).toEqual([true, false, true]); }); + test("a later session_id fallback continues the same raw parent-thread conversation", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + const identity = "mixed-header-conversation"; + const first = await handleResponses( + requestWithIdentityHeaders("first", { "x-codex-parent-thread-id": identity }), + config(), + { model: "", provider: "" }, + ); + expect(first.status).toBe(200); + await first.text(); + const second = await handleResponses( + requestWithIdentityHeaders("second", { session_id: identity }), + config(), + { model: "", provider: "" }, + ); + expect(second.status).toBe(200); + await second.text(); + + expect(outbound).toHaveLength(2); + expect(outbound.map(hasBlob)).toEqual([true, false]); + }); + + test("a shared session_id does not coalesce distinct thread-id conversations", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + const cases = [ + ["first", "thread-a"], + ["second", "thread-b"], + ] as const; + for (const [provider, threadId] of cases) { + const response = await handleResponses( + requestWithIdentityHeaders(provider, { + "thread-id": threadId, + session_id: "shared-cache-session", + }), + config(), + { model: "", provider: "" }, + ); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound).toHaveLength(2); + expect(outbound.map(hasBlob)).toEqual([true, true]); + }); + test("requests without any usable conversation identity keep blobs and record nothing", async () => { const outbound: Array> = []; globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { From 4813bcfb9f51d2c14fc368c190d51988f7af9a8e Mon Sep 17 00:00:00 2001 From: olddonkey Date: Fri, 21 Aug 2026 12:39:49 -0700 Subject: [PATCH 4/5] docs(responses): record the opaque-blob rejection memo Restore the architecture note for the conversation-and-serving-identity memo: five-minute TTL, successful blobless-retry admission, and later pre-flight stripping. --- structure/04_transports-and-sidecars.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 75410d38b3..94a90aca74 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -584,6 +584,15 @@ switch therefore costs one extra upstream round trip and one turn of degraded re wedging the thread; unrelated 4xx responses and requests whose outbound body carries no blob never enter this recovery. +After a self-identified opaque-blob rejection, the proxy also keeps a five-minute rejection memo. +The memo key is the resolved conversation identity plus the durable serving identity: provider, +destination, adapter, model, and credential. It is recorded only when the blobless recovery resend +succeeds. A missing durable destination or credential prevents memo creation and lookup. On a later +request with the same key, pre-flight sanitation removes opaque reasoning `encrypted_content` and +degrades compaction blobs before the first upstream send. This skips the rejected first send and +the recovery round trip. A different serving identity does not match the memo, so returning to the +blob-minting destination preserves valid blobs. Memo expiry returns to the fail-soft recovery path. + A combo target rotation between turns legitimately changes that serving identity, so the following turn drops blobs minted by the prior target. This is correct because the new target cannot decode them, but it is intentionally unobvious to the client: `pickComboTarget` keys selection state only by From 3d2bfebff363bac00946d6b75e89e776d75f21f7 Mon Sep 17 00:00:00 2001 From: Ingwannu Date: Fri, 21 Aug 2026 20:58:04 +0000 Subject: [PATCH 5/5] docs(responses): clarify replay memo route changes --- structure/04_transports-and-sidecars.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 5b5480b9ab..579f280df0 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -598,8 +598,8 @@ destination, adapter, model, and credential. It is recorded only when the bloble succeeds. A missing durable destination or credential prevents memo creation and lookup. On a later request with the same key, pre-flight sanitation removes opaque reasoning `encrypted_content` and degrades compaction blobs before the first upstream send. This skips the rejected first send and -the recovery round trip. A different serving identity does not match the memo, so returning to the -blob-minting destination preserves valid blobs. Memo expiry returns to the fail-soft recovery path. +the recovery round trip. A different serving identity does not match the memo. Route changes still +follow the normal pre-flight stripping rule. Memo expiry returns to the fail-soft recovery path. A combo target rotation between turns legitimately changes that serving identity, so the following turn drops blobs minted by the prior target. This is correct because the new target cannot decode