Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions scripts/test-layout/layout-replay.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
{
"responses-replay-metadata-principal.test.ts": "responses",
"responses-combo-replay-principal.test.ts": "responses"
}
2 changes: 1 addition & 1 deletion scripts/test-layout/layout.json
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
{
"version": 1,
"root": "tests",
"explicitShards": ["layout-clients.json", "layout-codex-collision.json", "layout-agent-assertion.json"],
"explicitShards": ["layout-clients.json", "layout-codex-collision.json", "layout-agent-assertion.json", "layout-replay.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",
"openai-chat-standalone-bare-block.test.ts": "adapters/openai",
"owner-registry-isolation.test.ts": "config",
Expand Down
32 changes: 18 additions & 14 deletions src/responses/reasoning-replay-cache.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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 {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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)
Expand Down Expand Up @@ -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;
}
Expand Down
3 changes: 3 additions & 0 deletions src/server/chat-completions.ts
Original file line number Diff line number Diff line change
Expand Up @@ -512,6 +512,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
Expand Down
13 changes: 10 additions & 3 deletions src/server/chat-native-reasoning.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand Down Expand Up @@ -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;
Expand Down
15 changes: 13 additions & 2 deletions src/server/chat-native.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}

/**
Expand Down Expand Up @@ -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",
Expand All @@ -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),
};
}
Expand Down Expand Up @@ -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;
Expand Down
1 change: 1 addition & 0 deletions src/server/responses/anthropic-thinking-recovery.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
8 changes: 7 additions & 1 deletion src/server/responses/compact.ts
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@ import {
ForwardAdmissionCredentialError,
contextPrincipalIdOf,
hasForwardableCodexBearer,
resolveContextPrincipal,
validateForwardAdmissionCredential,
} from "../auth-cors";
import type { DataPlaneAdmission } from "../auth-cors";
Expand Down Expand Up @@ -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`;
Expand Down
7 changes: 7 additions & 0 deletions src/server/responses/core-combo.ts
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,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
Expand Down Expand Up @@ -389,6 +390,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({
Expand Down Expand Up @@ -465,6 +470,7 @@ export async function executeComboResponses(
const unavailable = mandatoryResponsesReasoningReplayUnavailable({
body,
clientThreadId: reasoningReplayConversationId,
clientPrincipalId: replayPrincipal.clientPrincipalId,
providerName: route.providerName,
provider: route.provider,
adapterName: route.provider.adapter,
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
Expand Down Expand Up @@ -997,6 +1003,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
Expand Down
1 change: 1 addition & 0 deletions src/server/responses/core-opaque-recovery.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 } }
: {}),
Expand Down
5 changes: 5 additions & 0 deletions src/server/responses/core-options.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand Down
3 changes: 2 additions & 1 deletion src/server/responses/core-replay.ts
Original file line number Diff line number Diff line change
Expand Up @@ -344,6 +344,7 @@ function routeReasoningReplayIdentity(args: {
export function mandatoryResponsesReasoningReplayUnavailable(args: {
body: unknown;
clientThreadId: string | undefined;
clientPrincipalId?: string;
providerName: string;
provider: OcxProviderConfig;
adapterName: string;
Expand All @@ -359,7 +360,7 @@ export function mandatoryResponsesReasoningReplayUnavailable(args: {
) return false;

const current = routeReasoningReplayIdentity(args);
return reasoningReplayServingIdentityChanged({ clientThreadId, current });
return reasoningReplayServingIdentityChanged({ clientThreadId, clientPrincipalId: args.clientPrincipalId, current });
}


Expand Down
4 changes: 3 additions & 1 deletion src/server/responses/request-prepare.ts
Original file line number Diff line number Diff line change
Expand Up @@ -601,7 +601,9 @@ export async function prepareResponsesRequest(
// would become enough to read another caller's retained search result. Without a principal
// bridgeSearchReplayScope yields no scope, so nothing is recorded or restored for it. The
// field is always rewritten so an absent principal also clears one a reused holder carried.
const clientPrincipalId = resolveContextPrincipal(req, config, options.admission);
const clientPrincipalId = options.replayPrincipal
? options.replayPrincipal.clientPrincipalId
: resolveContextPrincipal(req, config, options.admission);
parsed._reasoningReplayScope = { ...parsed._reasoningReplayScope, clientPrincipalId };
}
// Prefer a pre-populated id (routed Claude) over Responses headers that may be
Expand Down
6 changes: 3 additions & 3 deletions structure/transports/anthropic-thinking-recovery.md
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,9 @@ the unsent repair uncharged. Recovery preserves the original thinking classifica

Only a delivery-validated completed response with actual text/tool output and an observed
successful terminal can memoize the rejection. Errors, incomplete/empty responses, missing
terminals, cancellation and changed serving identity cannot memoize it. The existing thread and
durable-serving-identity key and five-minute TTL in `src/responses/reasoning-replay-cache.ts` are
unchanged. Selective memos retain at most 256 payload digests within the existing cache byte cap;
terminals, cancellation and changed principal or serving identity cannot memoize it. The key in
`src/responses/reasoning-replay-cache.ts` includes the principal, thread and durable serving identity;
keyless callers have a separate namespace. The five-minute TTL and 256-payload digest cap remain;
merged sets that cannot fit fall back to the existing blanket policy. `src/server/responses/core-replay.ts`
binds a memo only for the current serving scope; a serving-identity change retains its independent
blanket strip. No prefix hashes or local prefix validator participate in replay admission.
Expand Down
Loading
Loading