diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts index 4b2d6d9166..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,11 +150,31 @@ 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 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 +192,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 { @@ -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/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 fca48dd90c..c95a1148f1 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"; @@ -223,6 +225,7 @@ import { import { conversationIdFromResponsesRequest, normalizeLogConversationId, + reasoningReplayConversationIdFromResponsesRequest, sessionIdHeaderFromRequest, } from "../request-log-conversation"; import type { AttemptRecoveryKind } from "../../usage/log"; @@ -514,6 +517,9 @@ function bindRouteReasoningReplayScope(args: { if (reasoningReplayServingIdentityChanged(parsed._reasoningReplayScope)) { parsed._stripReasoningEncryptedContent = true; } + if (reasoningReplayOpaqueBlobRejectionMemoized(parsed._reasoningReplayScope)) { + parsed._stripReasoningEncryptedContent = true; + } bindProviderContinuationForRoute(parsed, continuationOwner); } @@ -667,9 +673,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 }; @@ -2185,7 +2202,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 +2216,28 @@ 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 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 }; + } // 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..56156d8443 100644 --- a/src/types/request.ts +++ b/src/types/request.ts @@ -30,6 +30,11 @@ 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 a raw sanitized thread/Cursor/session fallback, never the + * hashed request-log conversation id. + */ readonly clientThreadId: string; current?: Readonly; } @@ -63,7 +68,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/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index b4df17b72a..579f280df0 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -592,6 +592,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. 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 them, but it is intentionally unobvious to the client: `pickComboTarget` keys selection state only by 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/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 e6ee2d5775..d161fe5522 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,310 @@ 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 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 (hasBlob(body)) return rejection(XAI_DECODE_ERROR); + 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: "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("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) => { + 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("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) => { + 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) => {