fix(replay): isolate retained reasoning state by admitted caller (carry #7523)

scripts/test-layout/layout-replay.json

@@ -0,0 +1,5 @@
+{
+  "responses-replay-metadata-principal.test.ts": "responses",
+  "responses-combo-replay-principal.test.ts": "responses",
+  "responses-translated-replay-principal.test.ts": "responses"
+}

scripts/test-layout/layout.json

@@ -1,7 +1,7 @@
 {
   "version": 1,
   "root": "tests",
-  "explicitShards": ["layout-clients.json", "layout-codex-collision.json", "layout-agent-assertion.json", "layout-chat.json", "layout-adapters.json", "layout-jev.json"],
+  "explicitShards": ["layout-clients.json", "layout-codex-collision.json", "layout-agent-assertion.json", "layout-chat.json", "layout-replay.json", "layout-adapters.json", "layout-jev.json"],
   "explicit": { "combo-description.test.ts": "routing", "combo-description-catalog.test.ts": "codex-integration", "codex-provider-routing.test.ts": "codex-integration", "config-native-reasoning-retention.test.ts": "config", "server-native-reasoning-retention-merge.test.ts": "server", "server-native-reasoning-retention.test.ts": "server", "responses-retained-reasoning-stream-recovery.test.ts": "responses", "openai-native-hosted-search-history.test.ts": "adapters/openai", "responses-native-compact-search-history.test.ts": "responses", "config-hosted-search-projection.test.ts": "config", "gcp-credential-boundary.test.ts": "adapters/google", "gcp-sa-marker.test.ts": "adapters/google", "gcp-sa-adapter.test.ts": "adapters/google", "management-gcp-credential-input.test.ts": "server", "codex-preserve-signed-in.test.ts": "codex-integration", "cli-account-preservation.test.ts": "cli", "cli-provider-list-active.test.ts": "cli", "service-state-isolation.test.ts": "service", "compaction-history-framing.test.ts": "responses", "catalog-native-max-context.test.ts": "codex-integration", "protocol-argon-thought-boundary.test.ts": "responses", "service-log.test.ts": "service", "messages-native-empty-inline.test.ts": "claude-integration", "catalog-auto-review-override.test.ts": "codex-integration", "management-auto-review-settings.test.ts": "server", "management-auto-review-auth.test.ts": "server", "prefer-model-alias-ids.test.ts": "server", "responses-mimo-envelope.test.ts": "responses", "responses-call-id-linkage.test.ts": "responses", "bridge-terminal-refusal.test.ts": "adapters", "loop-terminal-refusal.test.ts": "images", "web-search-terminal-refusal.test.ts": "web-search", "combo-terminal-refusal.test.ts": "routing", "server-terminal-refusal-wiring.test.ts": "server", "incomplete-reason-normalization.test.ts": "responses", "advisor-retention.test.ts": "advisor", "responses-native-key-cooldown.test.ts": "responses", "responses-run-turn-key-readmission.test.ts": "responses", "qoder-spawn-readmission.test.ts": "adapters", "api-key-selection-cooldown.test.ts": "providers", "api-key-auto-pause-distribution.test.ts": "providers", "explicit-key-pause-oauth-boundary.test.ts": "providers", "server-chat-native-key-selection.test.ts": "server", "compact-key-selection.test.ts": "responses", "messages-native-key-selection.test.ts": "claude-integration",
     "service-proxy-env.test.ts": "service", "openai-chat-standalone-bare-block.test.ts": "adapters/openai",
     "owner-registry-isolation.test.ts": "config",

src/responses/reasoning-replay-cache.ts

@@ -143,8 +143,9 @@ function tupleForServingIdentity(
 
 function keyFor(callId: string, scope: OcxReasoningReplayScopeRef | undefined): string | undefined {
   const identity = tupleForIdentity(scope?.current);
-  if (!nonEmpty(callId) || !nonEmpty(scope?.clientThreadId) || !identity) return undefined;
+  if (!nonEmpty(callId) || !nonEmpty(scope?.clientPrincipalId) || !nonEmpty(scope.clientThreadId) || !identity) return undefined;
   return JSON.stringify([
+    scope.clientPrincipalId,
     scope.clientThreadId,
     ...identity,
     callId,
@@ -166,18 +167,21 @@ function sweepExpiredServingIdentities(at: number): void {
 
 function servingIdentityFor(
   scope: OcxReasoningReplayScopeRef | undefined,
-): { threadId: string; identity: string } | undefined {
+): { key: string; identity: string } | undefined {
   const threadId = scope?.clientThreadId;
   const identityTuple = tupleForServingIdentity(scope?.current);
   if (!nonEmpty(threadId) || !identityTuple) return undefined;
-  return { threadId, identity: JSON.stringify(identityTuple) };
+  // Preserve physical safeguards for keyless callers in a separate namespace. A caller's
+  // thread name cannot select an authenticated principal's serving or rejection history.
+  const key = JSON.stringify([nonEmpty(scope?.clientPrincipalId) ? scope.clientPrincipalId : null, threadId]);
+  return { key, identity: JSON.stringify(identityTuple) };
 }
 
 function opaqueBlobRejectionKeyFor(
   scope: OcxReasoningReplayScopeRef | undefined,
 ): string | undefined {
   const current = servingIdentityFor(scope);
-  return current ? JSON.stringify([current.threadId, current.identity]) : undefined;
+  return current ? JSON.stringify([current.key, current.identity]) : undefined;
 }
 
 function deleteOpaqueBlobRejection(key: string): void {
@@ -224,7 +228,7 @@ export function reasoningReplayServingIdentityChange(
   if (!current) return undefined;
   const at = now();
   sweepExpiredServingIdentities(at);
-  const previous = servingIdentities.get(current.threadId);
+  const previous = servingIdentities.get(current.key);
   if (previous === undefined || previous.identity === current.identity) return undefined;
   const before = JSON.parse(previous.identity) as ReasoningReplayIdentityTuple;
   const after = JSON.parse(current.identity) as ReasoningReplayIdentityTuple;
@@ -254,7 +258,7 @@ export function reasoningReplayItemStoreChanged(
   if (!current) return false;
   const at = now();
   sweepExpiredServingIdentities(at);
-  const previous = servingIdentities.get(current.threadId);
+  const previous = servingIdentities.get(current.key);
   if (previous === undefined || previous.identity === current.identity) return false;
   const storeOwner = (identity: string): string => {
     const [, destination, , , credential] = JSON.parse(identity) as ReasoningReplayIdentityTuple;
@@ -271,16 +275,16 @@ export function commitReasoningReplayServingIdentity(
   if (!current) return;
   const at = now();
   sweepExpiredServingIdentities(at);
-  const previous = servingIdentities.get(current.threadId);
-  const { threadId, identity } = current;
-  const bytes = Buffer.byteLength(JSON.stringify([threadId, identity]), "utf8");
+  const previous = servingIdentities.get(current.key);
+  const { key, identity } = current;
+  const bytes = Buffer.byteLength(JSON.stringify([key, identity]), "utf8");
   if (bytes > MAX_TOTAL_BYTES) {
-    deleteServingIdentity(threadId);
+    deleteServingIdentity(key);
     return;
   }
 
-  if (previous) deleteServingIdentity(threadId);
-  servingIdentities.set(threadId, { identity, bytes, at });
+  if (previous) deleteServingIdentity(key);
+  servingIdentities.set(key, { identity, bytes, at });
   servingIdentityTotalBytes += bytes;
   while (
     (servingIdentityTotalBytes > MAX_TOTAL_BYTES || servingIdentities.size > MAX_ENTRIES)
@@ -517,13 +521,13 @@ export function reasoningReplayKeyCredentialIdentity(
   return reasoningReplayCredentialIdentity("key", apiKey, provider.headers);
 }
 
-/** Replace only the holder snapshot so parsed-request copies observe rotations. */
+/** Bind physical identity independently of the principal required for plaintext cache access. */
 export function bindReasoningReplayScope(
   scope: OcxReasoningReplayScopeRef | undefined,
   identity: OcxReasoningReplayIdentity | undefined,
 ): void {
   if (!scope) return;
-  if (!identity || !keyFor("binding-check", { ...scope, current: identity })) {
+  if (!identity || !nonEmpty(scope.clientThreadId) || !tupleForIdentity(identity)) {
     delete scope.current;
     return;
   }

src/server/chat-completions.ts

@@ -72,7 +72,7 @@ import {
   assertRouteAllowedByScope,
   resolveAdmissionModelScope,
 } from "./admission-model-scope";
-import type { DataPlaneAdmission } from "./auth-cors";
+import { resolveContextPrincipal, type DataPlaneAdmission } from "./auth-cors";
 import { tryClaimNativeMainProfileForTurn } from "../codex/native-main-admission";
 import {
   createTranslatorBudget,
@@ -506,6 +506,8 @@ async function handleChatCompletionsWithBudget(
     // #1686: the Chat surface translates its body and replays here, so the admission fact has
     // to ride along or a bearer-admitted Chat caller would still be refused by Direct.
     ...(logIds?.admission ? { admission: logIds.admission } : {}),
+    // The bridge request carries no caller credential, so resolve the replay principal here.
+    replayPrincipal: { clientPrincipalId: resolveContextPrincipal(req, config, logIds?.admission) },
     abortSignal: req.signal,
     // Body is Responses-shaped by now, but the client spoke Chat Completions.
     inboundWire: "chat",
@@ -520,6 +522,9 @@ async function handleChatCompletionsWithBudget(
     ...(envelope && nativeChatCombos ? {
       protocolSource: createNativeChatComboSource({
         req, config, envelope, requestedModel, requestedStream: stream, translatorBudget,
+        // Children run without logIds; the request's admission rides the source so their
+        // replay scope resolves the same caller principal as every other surface.
+        ...(logIds?.admission ? { admission: logIds.admission } : {}),
       }),
     } : {}),
     // Terminal vision-describe marker (roadmap 180): the bridge rebuilds

src/server/chat-native-reasoning.ts

@@ -7,11 +7,17 @@ import { reasoningReplayConversationIdFromResponsesRequest, sessionIdHeaderFromR
 
 type Rec = Record<string, unknown>;
 
-export function nativeChatReplayScope(headers: Headers): OcxReasoningReplayScopeRef | undefined {
+export function nativeChatReplayScope(
+  headers: Headers,
+  clientPrincipalId: string | undefined,
+): OcxReasoningReplayScopeRef | undefined {
   const clientThreadId = reasoningReplayConversationIdFromResponsesRequest({
     threadIdHeader: headers.get("thread-id"), sessionIdHeader: sessionIdHeaderFromRequest(headers),
   });
-  return clientThreadId ? { clientThreadId } : undefined;
+  return clientThreadId ? {
+    clientThreadId,
+    ...(clientPrincipalId ? { clientPrincipalId } : {}),
+  } : undefined;
 }
 
 /** Rebind before EVERY build: a key rotation must not retrieve another credential's reasoning. */
@@ -42,7 +48,8 @@ export function nativeChatReasoningCapture(
   const maxBytes = 256 * 1024;
   let bytes = 0;
   let callCount = 0;
-  let disabled = !scope?.current || !chatReasoningReplayPolicy(provider, modelId).preserve;
+  let disabled = !scope?.clientPrincipalId || !scope.current
+    || !chatReasoningReplayPolicy(provider, modelId).preserve;
   const discard = () => {
     budget.releaseRetained(bytes, accounting);
     bytes = 0;

src/server/chat-native.ts

@@ -24,6 +24,7 @@ import {
 } from "../lib/errors";
 import { reportDispatchSends, type RequestExecutionBudget, type SingleUseDispatchPermit } from "../lib/request-execution-budget";
 import type { AdmissionLease } from "../lib/admission";
+import { resolveContextPrincipal, type DataPlaneAdmission } from "./auth-cors";
 import { readBoundedResponseBody } from "../lib/bounded-body";
 import { redactSecretString } from "../lib/redact";
 import { resolveClientRetryAfter } from "../lib/retry-after";
@@ -179,7 +180,7 @@ export interface HandleNativeChatOptions {
   req: Request;
   config: OcxConfig;
   logCtx: RequestLogContext;
-  logIds?: { requestId: string; start: number; turnAdmissionLease?: AdmissionLease };
+  logIds?: { requestId: string; start: number; turnAdmissionLease?: AdmissionLease; admission?: DataPlaneAdmission };
   route: RouteResult;
   chatBody: Rec;
   requestedModel: string;
@@ -209,6 +210,12 @@ export interface NativeChatExecution extends HandleNativeChatOptions {
   onFirstOutput?: () => void;
   /** The lease a streamed body holds; defaults to `logIds.turnAdmissionLease`. */
   turnAdmissionLease?: AdmissionLease;
+  /**
+   * The request's resolved data-plane admission, when the caller carries it without `logIds`.
+   * A combo child runs on the parent's log row, so it receives the admitted identity here or
+   * its reasoning replay scope can never establish a caller principal (PF-07).
+   */
+  admission?: DataPlaneAdmission;
 }
 
 /**
@@ -246,6 +253,8 @@ export function createNativeChatComboSource(input: {
   requestedModel: string;
   requestedStream: boolean;
   translatorBudget: TranslatorBudget;
+  /** The request's data-plane admission; native children inherit it for replay scope. */
+  admission?: DataPlaneAdmission;
 }): ComboProtocolSource {
   return {
     inbound: "chat",
@@ -264,6 +273,7 @@ export function createNativeChatComboSource(input: {
       comboDispatchPermit: child.comboDispatchPermit,
       onFirstOutput: child.onFirstOutput,
       ...(child.turnAdmissionLease ? { turnAdmissionLease: child.turnAdmissionLease } : {}),
+      ...(input.admission ? { admission: input.admission } : {}),
     }, child.attemptHandle),
   };
 }
@@ -340,7 +350,8 @@ export async function runNativeChatAttempt(
   const physicalBudget = sendBudget ?? createInferenceSendBudget(req, logCtx);
   let activeAdapter: ProviderAdapter = createOpenAIChatAdapter(activeProvider);
   let activeRequest: AdapterRequest;
-  const replayScope = nativeChatReplayScope(req.headers);
+  const replayScope = nativeChatReplayScope(req.headers,
+    resolveContextPrincipal(req, config, execution.admission ?? logIds?.admission));
   let retainedRequestBytes = 0;
   const releaseRetainedRequest = () => {
     if (retainedRequestBytes === 0) return;

src/server/claude-messages.ts

@@ -100,6 +100,7 @@ import {
   isApiAuthRequired,
   isDataPlaneAdmissionSecret,
   isProxyAdmissionSecret,
+  resolveContextPrincipal,
   type RequestPolicyView,
   type DataPlaneAdmission,
 } from "./auth-cors";
@@ -1117,6 +1118,8 @@ async function handleClaudeMessagesWithBudget(
     // Routing keeps Claude-only sidecar overrides; admission policy must follow the live owner.
     codexAuthPolicy: config,
     ...(logIds?.admission ? { admission: logIds.admission } : {}),
+    // The translated request never forwards the caller credential; resolve the principal first.
+    replayPrincipal: { clientPrincipalId: resolveContextPrincipal(req, config, logIds?.admission) },
     ...(logIds?.turnAdmissionLease ? { turnAdmissionLease: logIds.turnAdmissionLease } : {}),
     abortSignal: req.signal,
     promptCacheKeyIsSharedCohort: cacheKeySource === "system",

src/server/responses/anthropic-thinking-recovery.ts

@@ -130,6 +130,7 @@ export function createAnthropicThinkingRecovery(
       || response.status !== "completed" || !Array.isArray(response.output) || response.output.length === 0
       || !rejectedScope
       || rejectedScope.clientThreadId !== parsed._reasoningReplayScope?.clientThreadId
+      || rejectedScope.clientPrincipalId !== parsed._reasoningReplayScope?.clientPrincipalId
       || JSON.stringify(rejectedScope.current) !== JSON.stringify(parsed._reasoningReplayScope?.current)) return;
     remembered = true;
     rememberReasoningReplayOpaqueBlobRejection(rejectedScope, rejectedPayloads);

src/server/responses/compact.ts

@@ -129,6 +129,7 @@ import {
   ForwardAdmissionCredentialError,
   contextPrincipalIdOf,
   hasForwardableCodexBearer,
+  resolveContextPrincipal,
   validateForwardAdmissionCredential,
 } from "../auth-cors";
 import type { DataPlaneAdmission } from "../auth-cors";
@@ -1002,7 +1003,12 @@ export async function handleResponsesCompact(
         cursorConversationId: gatewayReplayParsed._cursorConversationId,
         sessionIdHeader: sessionIdHeaderFromRequest(req.headers),
       });
-      if (clientThreadId) gatewayReplayParsed._reasoningReplayScope = { clientThreadId };
+      if (clientThreadId) {
+        gatewayReplayParsed._reasoningReplayScope = {
+          clientThreadId,
+          clientPrincipalId: resolveContextPrincipal(req, config, admission),
+        };
+      }
     }
     let compactUrl = gatewayTransport ? `${gatewayTransport.url.replace(/\/+$/, "")}/compact` : `${base}/responses/compact`;
     const compactTargetKey = `${route.providerName}|${route.modelId}|compact`;
@@ -1723,7 +1729,9 @@ export async function handleResponsesCompact(
     // The routed compaction turn is a handoff inside the same logical request, so it draws the
     // REMAINDER. Minting here is what let a native attempt spend three sends and the routed
     // fallback spend four more.
-    const response = await handleResponses(internalReq, config, logCtx, { abortSignal: req.signal, turnAdmissionLease, sendBudget, compactionRecoveryKind: "compaction-v1", compactionRoutingOverride: options.compactionRoutingOverride, ...(admission ? { admission } : {}) });
+    // internalReq keeps only FORWARD_HEADERS, so the caller principal is resolved from req.
+    const replayPrincipal = { clientPrincipalId: resolveContextPrincipal(req, config, admission) };
+    const response = await handleResponses(internalReq, config, logCtx, { abortSignal: req.signal, turnAdmissionLease, sendBudget, compactionRecoveryKind: "compaction-v1", compactionRoutingOverride: options.compactionRoutingOverride, replayPrincipal, ...(admission ? { admission } : {}) });
     if (req.signal.aborted) {
       void response.body?.cancel(req.signal.reason).catch(() => undefined);
       return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request");

src/server/responses/core-combo.ts

@@ -120,6 +120,7 @@ import { createComboProtocolLanes, dispatchNativeComboChild } from "./core-combo
 import { createJevModelInvoker } from "./jev-model-invoke";
 import { comboRequestedEffortLabel } from "./combo-requested-effort";
 import { clientWireOf } from "../inference/client-wire";
+import { resolveContextPrincipal } from "../auth-cors";
 
 /**
  * Sends one combo target may run on its own before the ladder moves on. A target is a whole
@@ -383,6 +384,10 @@ export async function executeComboResponses(
   if (!combo) {
     return formatErrorResponse(404, "invalid_request_error", `Unknown combo: ${comboId}`);
   }
+  // Resolve against the original request once; every bridge candidate receives the same trusted
+  // result after its provisional Authorization is removed, including an explicitly keyless result.
+  const replayPrincipal = options.replayPrincipal
+    ?? { clientPrincipalId: resolveContextPrincipal(req, config, options.admission) };
   // PF-07: present only for a Chat combo with `nativeChatCombos` on; otherwise every child
   // takes the bridge below exactly as before.
   const protocolLanes = createComboProtocolLanes({
@@ -459,6 +464,7 @@ export async function executeComboResponses(
       const unavailable = mandatoryResponsesReasoningReplayUnavailable({
         body,
         clientThreadId: reasoningReplayConversationId,
+        clientPrincipalId: replayPrincipal.clientPrincipalId,
         providerName: route.providerName,
         provider: route.provider,
         adapterName: route.provider.adapter,
@@ -1059,6 +1065,7 @@ export async function executeComboResponses(
         },
       }) : await requestDispatchers.handleResponses(childRequest, config, childLog, {
         ...options,
+        replayPrincipal,
         // A bridge child is a concrete route; the native source belongs to this loop only.
         ...(options.protocolSource ? { protocolSource: undefined } : {}),
         // After the spread: the child must run on THIS target's ladder, not on the holder the

src/server/responses/core-opaque-recovery.ts

@@ -547,6 +547,7 @@ export async function attemptOpaqueBlobRecovery<Failure = Response>(
   const rejectedScope = args.parsed._reasoningReplayScope
     ? {
         clientThreadId: args.parsed._reasoningReplayScope.clientThreadId,
+        clientPrincipalId: args.parsed._reasoningReplayScope.clientPrincipalId,
         ...(args.parsed._reasoningReplayScope.current
           ? { current: { ...args.parsed._reasoningReplayScope.current } }
           : {}),

src/server/responses/core-options.ts

@@ -128,6 +128,11 @@ export interface HandleResponsesOptions {
    * which is why it refused the whole env_key flow instead of serving it.
    */
   admission?: DataPlaneAdmission;
+  /**
+   * Internal caller identity captured before Combo strips provisional credentials. An existing
+   * holder with no principal preserves keyless replay; never populate it from client data.
+   */
+  replayPrincipal?: Readonly<{ clientPrincipalId: string | undefined }>;
   /** Called at most once after the complete client body is read and accepted for dispatch. */
   onRequestBodyRead?: () => void;
   /** Internal handoff for retry wrappers that must reuse the already-accounted request body. */