fix(responses): scope reasoning replay by conversation and remember proven blob rejections

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<string, CacheEntry>();
 const servingIdentities = new Map<string, ServingIdentityEntry>();
+const opaqueBlobRejections = new Map<string, OpaqueBlobRejectionEntry>();
 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;
 }

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 | null | undefined>
+): 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;

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;

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<OcxReasoningReplayIdentity>;
 }
@@ -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

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

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);