From 93bf9a376eb607a5167f35c475f5a7e5bef2e037 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Wed, 12 Aug 2026 10:33:18 +0900 Subject: [PATCH] fix(responses): bind reasoning replay to route identity --- src/adapters/openai-chat.ts | 2 +- src/bridge.ts | 12 +- src/images/loop.ts | 2 +- src/responses/reasoning-replay-cache.ts | 166 ++++++++++++++- src/server/responses/core.ts | 135 +++++++++++- src/types.ts | 23 ++ src/web-search/loop.ts | 2 +- structure/04_transports-and-sidecars.md | 21 ++ tests/bridge-raw-reasoning-hidden.test.ts | 16 +- tests/bridge-reasoning-replay-batch.test.ts | 23 +- tests/deepseek-reasoning-replay-gaps.test.ts | 77 +++++-- tests/images/loop-reasoning-replay.test.ts | 10 + tests/reasoning-replay-identity.test.ts | 211 +++++++++++++++++++ tests/reasoning-replay-robustness.test.ts | 16 +- tests/reasoning-replay-scope-source.test.ts | 24 ++- tests/server-key-failover-e2e.test.ts | 121 +++++++++++ tests/web-search.test.ts | 20 ++ 17 files changed, 828 insertions(+), 53 deletions(-) create mode 100644 tests/reasoning-replay-identity.test.ts diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index 6350d032aa..c3eed13c3e 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -226,7 +226,7 @@ function toolResultImageChatParts(content: string | OcxContentPart[]): unknown[] function messagesToChatFormat(parsed: OcxParsedRequest, provider: OcxProviderConfig): unknown[] { const out: unknown[] = []; const { context, options } = parsed; - const replayCacheScope = parsed._clientThreadId; + const replayCacheScope = parsed._reasoningReplayScope; interface PendingToolCall { id: string; name: string } let pendingToolCalls: PendingToolCall[] = []; diff --git a/src/bridge.ts b/src/bridge.ts index 1df0aeba5c..b6a61632af 100644 --- a/src/bridge.ts +++ b/src/bridge.ts @@ -1,4 +1,10 @@ -import type { AdapterEvent, OcxMessagePhase, OcxProviderContinuationState, OcxUsage } from "./types"; +import type { + AdapterEvent, + OcxMessagePhase, + OcxProviderContinuationState, + OcxReasoningReplayScopeRef, + OcxUsage, +} from "./types"; import { adapterFailureFromMessage, classifyError, CYBER_POLICY_ERROR_CODE, isCyberPolicyCode, type OcxErrorPayload } from "./lib/errors"; import { encodeCompactionSummary } from "./responses/compaction"; import { encodeReasoningEnvelope, type ReasoningEnvelope } from "./responses/reasoning-envelope"; @@ -186,7 +192,7 @@ export function bridgeToResponsesSSE( * Provider call ids are not globally unique; scoping by thread keeps one * conversation's reasoning out of another's continuations. */ - replayCacheScope?: string; + replayCacheScope?: OcxReasoningReplayScopeRef; /** * Test seam for the wire/stall beat loop. Production omits this and uses the * global timers; injecting here must not change scheduling semantics. @@ -1362,7 +1368,7 @@ function buildResponseJSONWithBudget( onUsage?: (usage: OcxUsage | undefined) => void; translatorBudget?: TranslatorBudget; /** Conversation identity for the reasoning replay cache (issue #950). */ - replayCacheScope?: string; + replayCacheScope?: OcxReasoningReplayScopeRef; }, ): Record { const responseId = `resp_${uuid()}`; diff --git a/src/images/loop.ts b/src/images/loop.ts index 07af9e6bbc..5042aab128 100644 --- a/src/images/loop.ts +++ b/src/images/loop.ts @@ -907,7 +907,7 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise number) | null = null; const now = (): number => clockForTests?.() ?? Date.now(); -const keyFor = (callId: string, scope: string): string => `${scope}\u0000${callId}`; + +function nonEmpty(value: unknown): value is string { + return typeof value === "string" && value.trim().length > 0; +} + +function keyFor(callId: string, scope: OcxReasoningReplayScopeRef | undefined): string | undefined { + const identity = scope?.current; + if ( + !nonEmpty(callId) + || !nonEmpty(scope?.clientThreadId) + || !nonEmpty(identity?.providerName) + || !nonEmpty(identity?.providerDestinationIdentity) + || !nonEmpty(identity?.adapterName) + || !nonEmpty(identity?.modelId) + || !nonEmpty(identity?.credentialIdentity) + ) return undefined; + return JSON.stringify([ + scope.clientThreadId, + identity.providerName, + identity.providerDestinationIdentity, + identity.adapterName, + identity.modelId, + identity.credentialIdentity, + callId, + ]); +} + +function processLocalIdentity(domain: string, material: string): string { + return createHmac("sha256", replayIdentityKey) + .update(domain) + .update("\0") + .update(material) + .digest("hex"); +} + +function credentialHeaderOverrides(headers: Record | undefined): [string, string][] { + return Object.entries(headers ?? {}) + .filter(([name, value]) => CREDENTIAL_HEADER_NAMES.has(name.trim().toLowerCase()) && nonEmpty(value)) + .map(([name, value]) => [name.trim().toLowerCase(), value] as [string, string]) + .sort(([leftName, leftValue], [rightName, rightValue]) => ( + leftName.localeCompare(rightName) || leftValue.localeCompare(rightValue) + )); +} + +/** Produce a non-reversible process-local identity for an exact upstream destination. */ +export function reasoningReplayDestinationIdentity(baseUrl: string | undefined): string | undefined { + if (!nonEmpty(baseUrl)) return undefined; + const canonical = baseUrl.trim().replace(/\/+$/, ""); + return `destination:${processLocalIdentity("destination", canonical)}`; +} + +/** Produce a non-reversible process-local identity for credential material. */ +export function reasoningReplayCredentialIdentity( + kind: "key" | "oauth" | "codex", + material: string | undefined, + headers?: Record, +): string | undefined { + if (!nonEmpty(material)) return undefined; + const overrides = credentialHeaderOverrides(headers); + return `${kind}:${processLocalIdentity(`credential:${kind}`, JSON.stringify([material, overrides]))}`; +} + +/** Bind Codex forwarding to the effective bearer/account and selected physical pool slot. */ +export function reasoningReplayCodexCredentialIdentity(args: { + authorization?: string | null; + chatgptAccountId?: string | null; + accountId?: string | null; + credentialGeneration?: number | string | null; + writerGeneration?: number | string | null; + headers?: Record; +}): string | undefined { + const configuredAuthorization = credentialHeaderOverrides(args.headers) + .find(([name]) => name === "authorization")?.[1]; + const authorization = nonEmpty(args.authorization) ? args.authorization : configuredAuthorization; + if (!authorization) return undefined; + const material = JSON.stringify([ + authorization, + nonEmpty(args.chatgptAccountId) ? args.chatgptAccountId : "", + nonEmpty(args.accountId) ? args.accountId : "", + args.credentialGeneration === null || args.credentialGeneration === undefined + ? "" + : String(args.credentialGeneration), + args.writerGeneration === null || args.writerGeneration === undefined + ? "" + : String(args.writerGeneration), + ]); + return reasoningReplayCredentialIdentity("codex", material, args.headers); +} + +/** Bind OAuth replay to one persisted credential slot and exact token generation. */ +export function reasoningReplayOAuthCredentialIdentity( + snapshot: Readonly<{ accountId: string; generation: string }> | undefined, + headers?: Record, +): string | undefined { + if (!snapshot || !nonEmpty(snapshot.accountId) || !nonEmpty(snapshot.generation)) return undefined; + return reasoningReplayCredentialIdentity( + "oauth", + `${snapshot.accountId}\0${snapshot.generation}`, + headers, + ); +} + +/** Bind key-auth provider material without putting raw secrets in the replay key. */ +export function reasoningReplayKeyCredentialIdentity( + provider: Pick, +): string | undefined { + const apiKey = nonEmpty(provider.apiKey) ? provider.apiKey : undefined; + // Public/static headers do not establish a physical credential boundary. + // Header-only/key-optional providers therefore fail closed for replay. + if (!apiKey) return undefined; + return reasoningReplayCredentialIdentity("key", apiKey, provider.headers); +} + +/** Replace only the holder snapshot so parsed-request copies observe rotations. */ +export function bindReasoningReplayScope( + scope: OcxReasoningReplayScopeRef | undefined, + identity: OcxReasoningReplayIdentity | undefined, +): void { + if (!scope) return; + if (!identity || !keyFor("binding-check", { ...scope, current: identity })) { + delete scope.current; + return; + } + scope.current = { ...identity }; +} /** * Record the raw reasoning text that preceded the given tool call. @@ -42,13 +189,18 @@ const keyFor = (callId: string, scope: string): string => `${scope}\u0000${callI * Expired entries are swept on insert so the TTL bound holds even when a call * id is never read again. */ -export function rememberReasoningForCall(callId: string, text: string, scope?: string): void { +export function rememberReasoningForCall( + callId: string, + text: string, + scope?: OcxReasoningReplayScopeRef, +): void { // Never fall back to a process-wide namespace. Call ids are supplied by // clients/providers and are therefore neither unique nor trustworthy; an // unscoped entry could be recovered by an unrelated request that reuses the // same id. // Empty provider deltas are absence of new reasoning, not a request to erase a candidate. - if (!scope || !callId || typeof text !== "string" || text.length === 0) return; + const key = keyFor(callId, scope); + if (!key || typeof text !== "string" || text.length === 0) return; const bytes = Buffer.byteLength(text, "utf8"); // A single entry larger than the whole budget would immediately evict itself. if (bytes > MAX_TOTAL_BYTES) return; @@ -61,7 +213,6 @@ export function rememberReasoningForCall(callId: string, text: string, scope?: s totalBytes -= entry.bytes; } } - const key = keyFor(callId, scope); const previous = entries.get(key); if (previous) totalBytes -= previous.bytes; entries.set(key, { text, bytes, at }); @@ -88,9 +239,12 @@ export function rememberReasoningForCall(callId: string, text: string, scope?: s * Read the recorded reasoning for a call id without removing it: retries after * a failed continuation reuse the same fallback. */ -export function peekReasoningForCall(callId: string, scope?: string): string | undefined { - if (!scope || !callId) return undefined; +export function peekReasoningForCall( + callId: string, + scope?: OcxReasoningReplayScopeRef, +): string | undefined { const key = keyFor(callId, scope); + if (!key) return undefined; const entry = entries.get(key); if (!entry) return undefined; if (now() - entry.at >= TTL_MS) { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 5de97affde..f578201cab 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -8,6 +8,13 @@ import { resolveEnvValue, } from "../../config"; import { parseRequest } from "../../responses/parser"; +import { + bindReasoningReplayScope, + reasoningReplayCodexCredentialIdentity, + reasoningReplayDestinationIdentity, + reasoningReplayKeyCredentialIdentity, + reasoningReplayOAuthCredentialIdentity, +} from "../../responses/reasoning-replay-cache"; import { buildCompactV1Output, COMPACT_PROMPT, decodeCompactionSummary, extractCompactUserMessages } from "../../responses/compaction"; import { FORWARD_HEADERS, sanitizeReasoningInputContent } from "../../adapters/openai-responses"; import { @@ -255,6 +262,58 @@ export function codexLogAccountId(authCtx: CodexAuthContext): string | null { return authCtx.kind === "pool" || authCtx.kind === "main-pool" ? authCtx.accountId : null; } +function bindRouteReasoningReplayScope(args: { + parsed: OcxParsedRequest; + providerName: string; + provider: OcxProviderConfig; + adapterName: string; + oauthCredentialSnapshot?: Pick; + codexAuthContext?: CodexAuthContext; + forwardHeaders?: Headers; +}): void { + const { parsed, providerName, provider, adapterName } = args; + let credentialIdentity: string | undefined; + if (provider.authMode === "oauth") { + credentialIdentity = reasoningReplayOAuthCredentialIdentity( + args.oauthCredentialSnapshot, + provider.headers, + ); + } else if (provider.authMode === "forward") { + const poolContext = args.codexAuthContext?.kind === "pool" + || args.codexAuthContext?.kind === "main-pool" + ? args.codexAuthContext + : undefined; + credentialIdentity = reasoningReplayCodexCredentialIdentity({ + authorization: poolContext + ? `Bearer ${poolContext.accessToken}` + : args.forwardHeaders?.get("authorization"), + chatgptAccountId: poolContext?.chatgptAccountId + ?? args.forwardHeaders?.get("chatgpt-account-id"), + accountId: poolContext?.accountId, + credentialGeneration: poolContext?.kind === "pool" + ? poolContext.generation + : undefined, + writerGeneration: poolContext?.writerGeneration, + headers: provider.headers, + }); + } else if (provider.authMode !== "local") { + credentialIdentity = reasoningReplayKeyCredentialIdentity(provider); + } + const providerDestinationIdentity = reasoningReplayDestinationIdentity(provider.baseUrl); + bindReasoningReplayScope( + parsed._reasoningReplayScope, + credentialIdentity && providerDestinationIdentity + ? { + providerName, + providerDestinationIdentity, + adapterName, + modelId: parsed.modelId, + credentialIdentity, + } + : undefined, + ); +} + function isFixedCodexAccount(authCtx: CodexAuthContext): boolean { return (authCtx.kind === "pool" || authCtx.kind === "main-pool") && authCtx.fixedAccount === true; @@ -481,6 +540,14 @@ async function retryCodexPoolOnAlternateAccount( resolveWireProtocolOverride(route.providerName, route.modelId, retryProvider, inboundWire), config.cacheRetention, ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: retryProvider, + adapterName: retryAdapter.name, + codexAuthContext: retryAuthCtx, + forwardHeaders: retryHeaders, + }); const request = await retryAdapter.buildRequest(parsed, { headers: retryHeaders, translatorBudget: options.translatorBudget, @@ -1454,7 +1521,10 @@ async function handleResponsesInner( parsed._providerContinuation = previousResponseProviderState(parsed.previousResponseId); parsed._cursorConversationId = parsed._providerContinuation?.cursor?.conversationId; const clientThreadId = req.headers.get("x-codex-parent-thread-id")?.trim(); - if (clientThreadId) parsed._clientThreadId = clientThreadId; + if (clientThreadId) { + parsed._clientThreadId = clientThreadId; + parsed._reasoningReplayScope = { clientThreadId }; + } } catch (err) { if (isTranslatorBudgetExceededError(err)) { return formatErrorResponse(413, "request_too_large", "request translation buffer exceeded the safe limit", { @@ -1679,6 +1749,7 @@ async function handleResponsesInner( const isOAuth401ReplayProvider = (route.providerName === "xai" || route.providerName === "github-copilot" || route.providerName === "kiro") && route.provider.authMode === "oauth"; let sentOAuthSnapshot: OAuthAccessSnapshot | undefined; + let replayOAuthCredentialSnapshot: Pick | undefined; let anthropicPoolAccountId: string | null = null; let anthropicPoolFailovers = 0; const anthropicSessionKey = route.providerName === "anthropic" && route.provider.authMode === "oauth" @@ -1714,6 +1785,10 @@ async function handleResponsesInner( logCtx.provider = formatAnthropicProviderForLog("anthropic", selection.accountId, config); } else { const resolved = await getValidAccessTokenSnapshot(route.providerName); + replayOAuthCredentialSnapshot = { + accountId: resolved.accountId, + generation: resolved.generation, + }; if (isOAuth401ReplayProvider) sentOAuthSnapshot = resolved; route.provider = { ...route.provider, apiKey: resolved.accessToken }; if (route.providerName === "kiro") { @@ -1765,6 +1840,15 @@ async function handleResponsesInner( logCtx.provider = route.providerName; } const adapter = resolveAdapter(adapterProvider, config.cacheRetention); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: adapterProvider, + adapterName: adapter.name, + oauthCredentialSnapshot: replayOAuthCredentialSnapshot, + codexAuthContext: authCtx, + forwardHeaders: selectedForwardHeaders, + }); logCtx.providerAdapter = adapter.name; // Ordinary requests receive one durable attempt only after their final initial // adapter is resolved. Combo children own their attempt and retries keep it. @@ -2726,10 +2810,17 @@ async function handleResponsesInner( }); if (!rotated) return null; route.provider = rotated; - return resolveAdapter( + const rotatedAdapter = resolveAdapter( resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire), config.cacheRetention, ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: route.provider, + adapterName: rotatedAdapter.name, + }); + return rotatedAdapter; }, retryOn429Policy: rateLimitRetryPolicyFor(route.provider), ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), @@ -2796,10 +2887,17 @@ async function handleResponsesInner( }); if (!rotated) return null; route.provider = rotated; - return resolveAdapter( + const rotatedAdapter = resolveAdapter( resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire), config.cacheRetention, ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: route.provider, + adapterName: rotatedAdapter.name, + }); + return rotatedAdapter; }, retryOn429Policy: rateLimitRetryPolicyFor(route.provider), }); @@ -2866,7 +2964,7 @@ async function handleResponsesInner( }, 2_000, { translatorBudget, - replayCacheScope: parsed._clientThreadId, + replayCacheScope: parsed._reasoningReplayScope, ...(options.forceEmptyResponseId ? { responseId: "" } : {}), stallTimeoutSec: config.stallTimeoutSec, hideThinkingSummary: parsed.options.hideThinkingSummary, @@ -2913,7 +3011,7 @@ async function handleResponsesInner( let providerState: OcxProviderContinuationState | undefined; const json = buildResponseJSON(events, parsed._responseModelId ?? parsed.modelId, { translatorBudget, - replayCacheScope: parsed._clientThreadId, + replayCacheScope: parsed._reasoningReplayScope, hideThinkingSummary: parsed.options.hideThinkingSummary, toolNsMap, freeformToolNames, @@ -3116,6 +3214,10 @@ async function handleResponsesInner( return formatErrorResponse(401, "authentication_error", err instanceof Error ? err.message : String(err)); } sentOAuthSnapshot = refreshed; + replayOAuthCredentialSnapshot = { + accountId: refreshed.accountId, + generation: refreshed.generation, + }; if (route.providerName === "kiro") { parsed._kiroAuthContext = { ...(refreshed.kiro ?? {}) }; } @@ -3131,6 +3233,13 @@ async function handleResponsesInner( resolveWireProtocolOverride(route.providerName, route.modelId, refreshedProvider, inboundWire), config.cacheRetention, ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: refreshedProvider, + adapterName: activeAdapter.name, + oauthCredentialSnapshot: replayOAuthCredentialSnapshot, + }); const result = await rebuildAndRefetch("oauth-401"); if ("failed" in result) return result.failed; upstreamResponse = result; @@ -3196,6 +3305,12 @@ async function handleResponsesInner( resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire), config.cacheRetention, ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: route.provider, + adapterName: activeAdapter.name, + }); const result = await rebuildAndRefetch("key-429"); if ("failed" in result) return result.failed; upstreamResponse = result; @@ -3467,6 +3582,12 @@ async function handleResponsesInner( resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, inboundWire), config.cacheRetention, ); + bindRouteReasoningReplayScope({ + parsed, + providerName: route.providerName, + provider: route.provider, + adapterName: activeAdapter.name, + }); nextContinuationRecoveryKind = "key-429"; continue; } @@ -3572,7 +3693,7 @@ async function handleResponsesInner( () => upstream.abort(), 2_000, { translatorBudget, - replayCacheScope: parsed._clientThreadId, + replayCacheScope: parsed._reasoningReplayScope, ...(options.forceEmptyResponseId ? { responseId: "" } : {}), stallTimeoutSec: config.stallTimeoutSec, hideThinkingSummary: parsed.options.hideThinkingSummary, @@ -3630,7 +3751,7 @@ async function handleResponsesInner( let providerState: OcxProviderContinuationState | undefined; const json = buildResponseJSON(events, parsed._responseModelId ?? parsed.modelId, { translatorBudget, - replayCacheScope: parsed._clientThreadId, + replayCacheScope: parsed._reasoningReplayScope, hideThinkingSummary: parsed.options.hideThinkingSummary, toolNsMap, freeformToolNames, diff --git a/src/types.ts b/src/types.ts index b50d4cb2ca..3dc5c3354a 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1,5 +1,26 @@ import type { KiroOAuthMetadata } from "./oauth/types"; +/** Exact provider/credential namespace for process-local reasoning replay. */ +export interface OcxReasoningReplayIdentity { + providerName: string; + /** Opaque process-local digest of the exact upstream destination. */ + providerDestinationIdentity: string; + adapterName: string; + modelId: string; + /** Opaque process-local credential identity; never a raw token or API key. */ + credentialIdentity: string; +} + +/** + * Stable holder shared by parsed-request copies and already-created bridges. + * Credential/provider rotation replaces `current` atomically without replacing + * the holder, so late tool-call cache writes see the active physical identity. + */ +export interface OcxReasoningReplayScopeRef { + readonly clientThreadId: string; + current?: Readonly; +} + export interface OcxParsedRequest { modelId: string; /** Client-facing model selector retained for Anthropic routes after wire-model normalization. */ @@ -21,6 +42,8 @@ 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. */ + _reasoningReplayScope?: OcxReasoningReplayScopeRef; /** * Optional authenticated tenant/operator namespace for Cursor thread→conversation derivation. * When absent (single-operator local proxy), derivation stays local-scoped. diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index bf5a7e64ce..0460a98cf5 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -781,7 +781,7 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise { for (const event of events) yield event; @@ -33,7 +33,17 @@ async function collectSse(stream: ReadableStream): Promise<{ event?: }); } -const REPLAY_SCOPE = "hidden-replay-thread"; +const REPLAY_SCOPE: OcxReasoningReplayScopeRef = { + clientThreadId: "hidden-replay-thread", + current: { + providerName: "routed", + providerDestinationIdentity: "destination:provider", + adapterName: "openai-chat", + modelId: "model", + credentialIdentity: "key:test", + }, +}; +const GLOBAL_SCOPE: OcxReasoningReplayScopeRef = { ...REPLAY_SCOPE, clientThreadId: "global" }; const sseOpts = (hide: boolean) => ({ hideThinkingSummary: hide, replayCacheScope: REPLAY_SCOPE }); describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_delta)", () => { @@ -164,7 +174,7 @@ describe("hidden raw reasoning (hideThinkingSummary parity for reasoning_raw_del { type: "tool_call_end" }, { type: "done" }, ]), "routed/model", undefined, undefined, undefined, undefined, undefined, { hideThinkingSummary: true })); - expect(peekReasoningForCall("call_unscoped_stream", "global")).toBeUndefined(); + expect(peekReasoningForCall("call_unscoped_stream", GLOBAL_SCOPE)).toBeUndefined(); }); test("non-streaming hidden: raw reasoning is recorded for the following tool call", () => { diff --git a/tests/bridge-reasoning-replay-batch.test.ts b/tests/bridge-reasoning-replay-batch.test.ts index 714ad9afcd..fa288b210e 100644 --- a/tests/bridge-reasoning-replay-batch.test.ts +++ b/tests/bridge-reasoning-replay-batch.test.ts @@ -4,7 +4,7 @@ import { clearReasoningReplayCacheForTests, peekReasoningForCall, } from "../src/responses/reasoning-replay-cache"; -import type { AdapterEvent } from "../src/types"; +import type { AdapterEvent, OcxReasoningReplayScopeRef } from "../src/types"; /** * Regression for issue #950's non-streaming path: chat-completions batch @@ -18,7 +18,17 @@ import type { AdapterEvent } from "../src/types"; */ const REASONING = "I need to inspect files before answering."; -const SCOPE = "thread-batch"; +const SCOPE: OcxReasoningReplayScopeRef = { + clientThreadId: "thread-batch", + current: { + providerName: "opencode-free", + providerDestinationIdentity: "destination:provider", + adapterName: "openai-chat", + modelId: "deepseek-v4-flash-free", + credentialIdentity: "key:test", + }, +}; +const GLOBAL_SCOPE: OcxReasoningReplayScopeRef = { ...SCOPE, clientThreadId: "global" }; function batchOutput(events: AdapterEvent[]): Record { return buildResponseJSON(events, "opencode-free/deepseek-v4-flash-free", { @@ -26,7 +36,10 @@ function batchOutput(events: AdapterEvent[]): Record { }); } -async function streamFrames(events: AdapterEvent[], replayScope: string | null = SCOPE): Promise { +async function streamFrames( + events: AdapterEvent[], + replayScope: OcxReasoningReplayScopeRef | null = SCOPE, +): Promise { async function* replay(list: AdapterEvent[]): AsyncGenerator { for (const event of list) yield event; } @@ -73,7 +86,7 @@ describe("reasoning replay survives empty text deltas (both wire modes)", () => test("batch: an unscoped bridge never writes a global replay entry", () => { buildResponseJSON(toolRoundEvents(), "opencode-free/deepseek-v4-flash-free", {}); - expect(peekReasoningForCall("call_batch_1", "global")).toBeUndefined(); + expect(peekReasoningForCall("call_batch_1", GLOBAL_SCOPE)).toBeUndefined(); }); test("batch: real text between reasoning and the tool call clears the cache target", () => { @@ -90,7 +103,7 @@ describe("reasoning replay survives empty text deltas (both wire modes)", () => test("stream: an unscoped bridge never writes a global replay entry", async () => { await streamFrames(toolRoundEvents(), null); - expect(peekReasoningForCall("call_batch_1", "global")).toBeUndefined(); + expect(peekReasoningForCall("call_batch_1", GLOBAL_SCOPE)).toBeUndefined(); }); test("stream: real text between reasoning and the tool call clears the cache target", async () => { diff --git a/tests/deepseek-reasoning-replay-gaps.test.ts b/tests/deepseek-reasoning-replay-gaps.test.ts index 73226c2d09..ecbff4122d 100644 --- a/tests/deepseek-reasoning-replay-gaps.test.ts +++ b/tests/deepseek-reasoning-replay-gaps.test.ts @@ -8,7 +8,12 @@ import { rememberReasoningForCall as rememberReasoningForCallRaw, } from "../src/responses/reasoning-replay-cache"; import { routeModel } from "../src/router"; -import type { AdapterEvent, OcxConfig, OcxParsedRequest } from "../src/types"; +import type { + AdapterEvent, + OcxConfig, + OcxParsedRequest, + OcxReasoningReplayScopeRef, +} from "../src/types"; /** * Regression coverage for opencodex issue #950: OpenCode Go DeepSeek V4 Flash @@ -26,7 +31,23 @@ import type { AdapterEvent, OcxConfig, OcxParsedRequest } from "../src/types"; const MODEL = "opencode-go/deepseek-v4-flash"; const REASONING = "I need to inspect files before answering."; -const REPLAY_SCOPE = "test-thread"; +function replayScope( + clientThreadId = "test-thread", + overrides: Partial> = {}, +): OcxReasoningReplayScopeRef { + return { + clientThreadId, + current: { + providerName: "opencode-go", + providerDestinationIdentity: "destination:opencode-zen-go", + adapterName: "openai-chat", + modelId: "deepseek-v4-flash", + credentialIdentity: "key:test", + ...overrides, + }, + }; +} +const REPLAY_SCOPE = replayScope(); const rememberReasoningForCall = (callId: string, text: string, scope = REPLAY_SCOPE): void => rememberReasoningForCallRaw(callId, text, scope); const peekReasoningForCall = (callId: string, scope = REPLAY_SCOPE): string | undefined => @@ -49,10 +70,13 @@ function configFor(): OcxConfig { function wireFor( input: unknown[], - replayScope: string | null = REPLAY_SCOPE, + scope: OcxReasoningReplayScopeRef | null = REPLAY_SCOPE, ): { messages: Array> } { const parsed = parseRequest({ model: MODEL, input, stream: true }); - if (replayScope !== null) parsed._clientThreadId = replayScope; + if (scope !== null) { + parsed._clientThreadId = scope.clientThreadId; + parsed._reasoningReplayScope = scope; + } const route = routeModel(configFor(), parsed.modelId); parsed.modelId = route.modelId; const req = createOpenAIChatAdapter(route.provider).buildRequest(parsed as OcxParsedRequest); @@ -146,8 +170,10 @@ describe("issue #950 — tool-call reasoning replay invariant (openai-chat wire) }); test("cross-request replay isolates threads and rejects an unscoped producer/consumer pair", () => { - rememberReasoningForCallRaw("call_1", "thread alpha reasoning", "thread-a"); - rememberReasoningForCallRaw("call_1", "thread beta reasoning", "thread-b"); + const threadA = replayScope("thread-a"); + const threadB = replayScope("thread-b"); + rememberReasoningForCallRaw("call_1", "thread alpha reasoning", threadA); + rememberReasoningForCallRaw("call_1", "thread beta reasoning", threadB); const unscopedProducer: AdapterEvent[] = [ { type: "reasoning_raw_delta", text: "unrelated private reasoning" }, { type: "tool_call_start", id: "call_1", name: "read_file" }, @@ -162,8 +188,8 @@ describe("issue #950 — tool-call reasoning replay invariant (openai-chat wire) functionCallItem(), functionCallOutputItem(), ]; - const alpha = toolCallAssistant(wireFor(input, "thread-a").messages); - const beta = toolCallAssistant(wireFor(input, "thread-b").messages); + const alpha = toolCallAssistant(wireFor(input, threadA).messages); + const beta = toolCallAssistant(wireFor(input, threadB).messages); const unscoped = toolCallAssistant(wireFor(input, null).messages); expect(alpha?.reasoning_content).toBe("thread alpha reasoning"); expect(beta?.reasoning_content).toBe("thread beta reasoning"); @@ -230,7 +256,13 @@ describe("issue #950 — tool-call reasoning replay invariant (openai-chat wire) // reasoning still replays via preserveReasoningContentModels. const minimaxWire = (input: unknown[]) => { const parsed = parseRequest({ model: "minimax/MiniMax-M3", input, stream: true }); - parsed._clientThreadId = REPLAY_SCOPE; + const minimaxReplayScope = replayScope("test-thread-minimax", { + providerName: "minimax", + providerDestinationIdentity: "destination:minimax", + modelId: "MiniMax-M3", + }); + parsed._clientThreadId = minimaxReplayScope.clientThreadId; + parsed._reasoningReplayScope = minimaxReplayScope; const config: OcxConfig = { port: 10100, defaultProvider: "minimax", @@ -245,15 +277,19 @@ describe("issue #950 — tool-call reasoning replay invariant (openai-chat wire) const route = routeModel(config, parsed.modelId); parsed.modelId = route.modelId; const req = createOpenAIChatAdapter(route.provider).buildRequest(parsed as OcxParsedRequest); - return JSON.parse(req.body as string) as { messages: Array> }; + return { + wire: JSON.parse(req.body as string) as { messages: Array> }, + replayScope: minimaxReplayScope, + }; }; // Cache miss on the orphan-repair path: no fabricated placeholder. - const miss = toolCallAssistant(minimaxWire([userMessage(), functionCallOutputItem()]).messages); + const missResult = minimaxWire([userMessage(), functionCallOutputItem()]); + const miss = toolCallAssistant(missResult.wire.messages); expect(miss).toBeDefined(); expect(miss!["reasoning_content"]).toBeUndefined(); // Cache hit on the same path: the recorded reasoning still replays. - rememberReasoningForCall("call_1", REASONING); - const hit = toolCallAssistant(minimaxWire([userMessage(), functionCallOutputItem()]).messages); + rememberReasoningForCall("call_1", REASONING, missResult.replayScope); + const hit = toolCallAssistant(minimaxWire([userMessage(), functionCallOutputItem()]).wire.messages); expect(hit).toBeDefined(); expect(hit!["reasoning_content"]).toBe(REASONING); }); @@ -323,10 +359,12 @@ describe("issue #950 — reasoning replay cache bounds", () => { }); test("conversation scopes isolate entries with the same call id", () => { - rememberReasoningForCall("call_1", "thread alpha reasoning", "thread-a"); - rememberReasoningForCall("call_1", "thread beta reasoning", "thread-b"); - expect(peekReasoningForCall("call_1", "thread-a")).toBe("thread alpha reasoning"); - expect(peekReasoningForCall("call_1", "thread-b")).toBe("thread beta reasoning"); + const threadA = replayScope("thread-a"); + const threadB = replayScope("thread-b"); + rememberReasoningForCall("call_1", "thread alpha reasoning", threadA); + rememberReasoningForCall("call_1", "thread beta reasoning", threadB); + expect(peekReasoningForCall("call_1", threadA)).toBe("thread alpha reasoning"); + expect(peekReasoningForCall("call_1", threadB)).toBe("thread beta reasoning"); // An unscoped read must not see either scoped entry. expect(peekReasoningForCallRaw("call_1")).toBeUndefined(); }); @@ -334,7 +372,10 @@ describe("issue #950 — reasoning replay cache bounds", () => { test("unscoped entries are rejected instead of sharing a process-wide namespace", () => { rememberReasoningForCallRaw("call_collision", "private reasoning"); expect(peekReasoningForCallRaw("call_collision")).toBeUndefined(); - expect(peekReasoningForCallRaw("call_collision", "global")).toBeUndefined(); + expect(peekReasoningForCallRaw( + "call_collision", + "global" as unknown as OcxReasoningReplayScopeRef, + )).toBeUndefined(); }); test("entries expire after the TTL", () => { diff --git a/tests/images/loop-reasoning-replay.test.ts b/tests/images/loop-reasoning-replay.test.ts index 7ddb7c3901..b854b8060e 100644 --- a/tests/images/loop-reasoning-replay.test.ts +++ b/tests/images/loop-reasoning-replay.test.ts @@ -92,6 +92,16 @@ function makeParsed(): OcxParsedRequest { stream: true, options: {}, _clientThreadId: "image-replay-test", + _reasoningReplayScope: { + clientThreadId: "image-replay-test", + current: { + providerName: "test", + providerDestinationIdentity: "destination:test", + adapterName: "test", + modelId: "test-model", + credentialIdentity: "key:test", + }, + }, } as OcxParsedRequest; } diff --git a/tests/reasoning-replay-identity.test.ts b/tests/reasoning-replay-identity.test.ts new file mode 100644 index 0000000000..80cdd49aaa --- /dev/null +++ b/tests/reasoning-replay-identity.test.ts @@ -0,0 +1,211 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { bridgeToResponsesSSE } from "../src/bridge"; +import { + bindReasoningReplayScope, + clearReasoningReplayCacheForTests, + peekReasoningForCall, + reasoningReplayCodexCredentialIdentity, + reasoningReplayCredentialIdentity, + reasoningReplayDestinationIdentity, + reasoningReplayKeyCredentialIdentity, + reasoningReplayOAuthCredentialIdentity, + rememberReasoningForCall, +} from "../src/responses/reasoning-replay-cache"; +import type { AdapterEvent, OcxReasoningReplayScopeRef } from "../src/types"; + +const THREAD = "thread-identity"; +const CALL_ID = "call_identity_collision"; +const REASONING = "private reasoning"; + +function scope( + overrides: Partial> = {}, +): OcxReasoningReplayScopeRef { + return { + clientThreadId: THREAD, + current: { + providerName: "provider-a", + providerDestinationIdentity: "destination:provider-a", + adapterName: "openai-chat", + modelId: "deepseek-v4-flash", + credentialIdentity: "key:physical-a", + ...overrides, + }, + }; +} + +async function drain(events: AsyncIterable, replayScope: OcxReasoningReplayScopeRef): Promise { + const reader = bridgeToResponsesSSE( + events, + "deepseek-v4-flash", + undefined, + undefined, + undefined, + undefined, + undefined, + { replayCacheScope: replayScope }, + ).getReader(); + while (!(await reader.read()).done) { + // Drain the bridge so tool-call cache writes complete. + } +} + +describe("reasoning replay provider and credential identity", () => { + beforeEach(() => clearReasoningReplayCacheForTests()); + afterEach(() => clearReasoningReplayCacheForTests()); + + test("the same tuple replays but provider, destination, wire, model, or credential changes miss", () => { + const original = scope(); + rememberReasoningForCall(CALL_ID, REASONING, original); + expect(peekReasoningForCall(CALL_ID, scope())).toBe(REASONING); + + for (const changed of [ + scope({ providerName: "provider-b" }), + scope({ providerDestinationIdentity: "destination:provider-b" }), + scope({ adapterName: "openai-responses" }), + scope({ modelId: "deepseek-v4" }), + scope({ credentialIdentity: "key:physical-b" }), + ]) { + expect(peekReasoningForCall(CALL_ID, changed)).toBeUndefined(); + } + }); + + test("incomplete, unscoped, and legacy thread-only namespaces fail closed", () => { + const incomplete: OcxReasoningReplayScopeRef[] = [ + { clientThreadId: THREAD }, + scope({ credentialIdentity: "" }), + ]; + for (const candidate of incomplete) { + rememberReasoningForCall(CALL_ID, REASONING, candidate); + expect(peekReasoningForCall(CALL_ID, candidate)).toBeUndefined(); + } + rememberReasoningForCall(CALL_ID, REASONING); + expect(peekReasoningForCall(CALL_ID)).toBeUndefined(); + rememberReasoningForCall(CALL_ID, REASONING, THREAD as unknown as OcxReasoningReplayScopeRef); + expect(peekReasoningForCall(CALL_ID, THREAD as unknown as OcxReasoningReplayScopeRef)).toBeUndefined(); + }); + + test("invalidating a bound holder prevents writes under its stale identity", () => { + const bound = scope(); + const oldIdentity = scope(); + rememberReasoningForCall(CALL_ID, "old reasoning", oldIdentity); + + bindReasoningReplayScope(bound, undefined); + expect(bound.current).toBeUndefined(); + rememberReasoningForCall(CALL_ID, "stale overwrite", bound); + + expect(peekReasoningForCall(CALL_ID, oldIdentity)).toBe("old reasoning"); + expect(peekReasoningForCall(CALL_ID, bound)).toBeUndefined(); + }); + + test("credential and destination identities are stable, bounded, and never contain raw material", () => { + const provider = { + apiKey: "secret-key-a", + headers: { Authorization: "Bearer static-secret", "x-session-id": "session-a" }, + }; + const first = reasoningReplayKeyCredentialIdentity(provider); + const same = reasoningReplayKeyCredentialIdentity({ + headers: { Authorization: "Bearer static-secret", "x-session-id": "session-b" }, + apiKey: "secret-key-a", + }); + const different = reasoningReplayKeyCredentialIdentity({ ...provider, apiKey: "secret-key-b" }); + const overridden = reasoningReplayKeyCredentialIdentity({ + ...provider, + headers: { Authorization: "Bearer other-static-secret" }, + }); + expect(first).toBe(same); + expect(first).not.toBe(different); + expect(first).not.toBe(overridden); + expect(reasoningReplayKeyCredentialIdentity({ + headers: { "x-opencode-client": "desktop" }, + })).toBeUndefined(); + expect(first).not.toContain("secret-key-a"); + expect(first).not.toContain("static-secret"); + + const oauthA = reasoningReplayOAuthCredentialIdentity({ accountId: "slot-a", generation: "generation-a" }, { + Authorization: "Bearer override-a", + }); + const oauthOverrideB = reasoningReplayOAuthCredentialIdentity({ accountId: "slot-a", generation: "generation-a" }, { + Authorization: "Bearer override-b", + }); + const oauthGenerationB = reasoningReplayOAuthCredentialIdentity({ accountId: "slot-a", generation: "generation-b" }, { + Authorization: "Bearer override-a", + }); + const oauthAccountB = reasoningReplayOAuthCredentialIdentity({ accountId: "slot-b", generation: "generation-a" }, { + Authorization: "Bearer override-a", + }); + expect(oauthA).not.toBe(oauthOverrideB); + expect(oauthA).not.toBe(oauthGenerationB); + expect(oauthA).not.toBe(oauthAccountB); + expect(oauthA).not.toContain("slot-a"); + expect(oauthA).not.toContain("generation-a"); + expect(reasoningReplayOAuthCredentialIdentity({ accountId: "", generation: "generation-a" })).toBeUndefined(); + expect(reasoningReplayOAuthCredentialIdentity({ accountId: "slot-a", generation: "" })).toBeUndefined(); + expect(oauthA).not.toContain("override-a"); + + const forwardA = reasoningReplayCodexCredentialIdentity({ + authorization: "Bearer forward-token-a", + chatgptAccountId: "workspace-a", + }); + const forwardB = reasoningReplayCodexCredentialIdentity({ + authorization: "Bearer forward-token-b", + chatgptAccountId: "workspace-a", + }); + const poolA = reasoningReplayCodexCredentialIdentity({ + authorization: "Bearer shared-token", + chatgptAccountId: "workspace-a", + accountId: "pool-a", + credentialGeneration: 7, + writerGeneration: 11, + }); + const poolB = reasoningReplayCodexCredentialIdentity({ + authorization: "Bearer shared-token", + chatgptAccountId: "workspace-a", + accountId: "pool-b", + credentialGeneration: 7, + writerGeneration: 11, + }); + const newerPoolA = reasoningReplayCodexCredentialIdentity({ + authorization: "Bearer shared-token", + chatgptAccountId: "workspace-a", + accountId: "pool-a", + credentialGeneration: 8, + writerGeneration: 12, + }); + expect(forwardA).not.toBe(forwardB); + expect(poolA).not.toBe(poolB); + expect(poolA).not.toBe(newerPoolA); + expect(poolA).not.toBe(reasoningReplayCodexCredentialIdentity({ + authorization: "Bearer shared-token", + chatgptAccountId: "workspace-a", + accountId: "pool-a", + credentialGeneration: 7, + writerGeneration: 12, + })); + expect(forwardA).not.toContain("forward-token-a"); + expect(poolA).not.toContain("pool-a"); + expect(reasoningReplayCodexCredentialIdentity({ chatgptAccountId: "workspace-a" })).toBeUndefined(); + + const destination = reasoningReplayDestinationIdentity("https://provider.example/v1/opaque-secret"); + expect(destination).toBe(reasoningReplayDestinationIdentity("https://provider.example/v1/opaque-secret/")); + expect(destination).not.toBe(reasoningReplayDestinationIdentity("https://provider.example/v1/other-secret")); + expect(destination).not.toContain("opaque-secret"); + }); + + test("a bridge created before credential rotation writes under the holder's current identity", async () => { + const oldScope = scope(); + rememberReasoningForCall(CALL_ID, "old reasoning", oldScope); + const holder = scope(); + async function* events(): AsyncGenerator { + yield { type: "reasoning_raw_delta", text: "new reasoning" }; + holder.current = { ...holder.current!, credentialIdentity: "key:physical-b" }; + yield { type: "tool_call_start", id: CALL_ID, name: "read_file" }; + yield { type: "tool_call_delta", arguments: "{}" }; + yield { type: "tool_call_end" }; + yield { type: "done" }; + } + const pending = drain(events(), holder); + await pending; + expect(peekReasoningForCall(CALL_ID, oldScope)).toBe("old reasoning"); + expect(peekReasoningForCall(CALL_ID, holder)).toBe("new reasoning"); + }); +}); diff --git a/tests/reasoning-replay-robustness.test.ts b/tests/reasoning-replay-robustness.test.ts index c8212476bb..4672d60110 100644 --- a/tests/reasoning-replay-robustness.test.ts +++ b/tests/reasoning-replay-robustness.test.ts @@ -9,10 +9,19 @@ import { peekReasoningForCall, rememberReasoningForCall, } from "../src/responses/reasoning-replay-cache"; -import type { AdapterEvent } from "../src/types"; +import type { AdapterEvent, OcxReasoningReplayScopeRef } from "../src/types"; const REASONING = "I need to inspect files before answering."; -const SCOPE = "thread-empty-delta"; +const SCOPE: OcxReasoningReplayScopeRef = { + clientThreadId: "thread-empty-delta", + current: { + providerName: "opencode-free", + providerDestinationIdentity: "destination:provider", + adapterName: "openai-chat", + modelId: "deepseek-v4-flash-free", + credentialIdentity: "key:test", + }, +}; function reasoningToolRound(intervening: AdapterEvent): AdapterEvent[] { return [ @@ -92,7 +101,8 @@ describe("reasoning replay empty-delta robustness", () => { const moduleUrl = pathToFileURL(join(import.meta.dir, "../src/responses/reasoning-replay-cache.ts")).href; const script = [ `const cache = await import(${JSON.stringify(moduleUrl)});`, - `cache.rememberReasoningForCall("call_disk_guard", ${JSON.stringify(REASONING)}, "disk-guard");`, + `const scope = {clientThreadId:"disk-guard",current:{providerName:"test",providerDestinationIdentity:"destination:test",adapterName:"openai-chat",modelId:"model",credentialIdentity:"key:test"}};`, + `cache.rememberReasoningForCall("call_disk_guard", ${JSON.stringify(REASONING)}, scope);`, ].join("\n"); try { diff --git a/tests/reasoning-replay-scope-source.test.ts b/tests/reasoning-replay-scope-source.test.ts index c68b0d2bd1..8db776ae30 100644 --- a/tests/reasoning-replay-scope-source.test.ts +++ b/tests/reasoning-replay-scope-source.test.ts @@ -6,13 +6,25 @@ const source = (relative: string): string => readFileSync(join(import.meta.dir, "..", "src", ...relative.split("/")), "utf8"); describe("reasoning replay scope propagation", () => { - test("every production bridge call passes only the explicit client thread scope", () => { + test("every production bridge call passes the provider-bound scope holder", () => { const core = source("server/responses/core.ts"); const images = source("images/loop.ts"); const webSearch = source("web-search/loop.ts"); - expect(core.match(/replayCacheScope: parsed\._clientThreadId,/g)).toHaveLength(4); - expect(images.match(/replayCacheScope: parsed\._clientThreadId,/g)).toHaveLength(1); - expect(webSearch.match(/replayCacheScope: parsed\._clientThreadId,/g)).toHaveLength(1); + expect(core.match(/replayCacheScope: parsed\._reasoningReplayScope,/g)).toHaveLength(4); + expect(images.match(/replayCacheScope: parsed\._reasoningReplayScope,/g)).toHaveLength(1); + expect(webSearch.match(/replayCacheScope: parsed\._reasoningReplayScope,/g)).toHaveLength(1); + expect(`${core}\n${images}\n${webSearch}`).not.toContain("replayCacheScope: parsed._clientThreadId"); + expect(core).toContain("reasoningReplayDestinationIdentity(provider.baseUrl)"); + expect(core).toMatch(/reasoningReplayOAuthCredentialIdentity\(\s*args\.oauthCredentialSnapshot,\s*provider\.headers,/); + expect(core).toContain("accountId: resolved.accountId"); + expect(core).toContain("generation: resolved.generation"); + expect(core).toContain("accountId: refreshed.accountId"); + expect(core).toContain("generation: refreshed.generation"); + expect(core).toContain("reasoningReplayCodexCredentialIdentity({"); + expect(core).toContain("authorization: poolContext"); + expect(core).toContain("accountId: poolContext?.accountId"); + expect(core).toContain("credentialGeneration: poolContext?.kind === \"pool\""); + expect(core).toContain("writerGeneration: poolContext?.writerGeneration"); }); test("bridge, adapter, and cache contain no process-wide fallback", () => { @@ -20,8 +32,10 @@ describe("reasoning replay scope propagation", () => { const adapter = source("adapters/openai-chat.ts"); const cache = source("responses/reasoning-replay-cache.ts"); expect(bridge.match(/const replayCacheScope = options\?\.replayCacheScope;/g)).toHaveLength(2); - expect(adapter.match(/const replayCacheScope = parsed\._clientThreadId;/g)).toHaveLength(1); + expect(adapter.match(/const replayCacheScope = parsed\._reasoningReplayScope;/g)).toHaveLength(1); + expect(adapter).not.toContain("const replayCacheScope = parsed._clientThreadId"); expect(cache).not.toContain('scope ?? "global"'); + expect(cache).not.toContain("identity.providerBaseUrl"); expect(`${bridge}\n${adapter}`).not.toContain('replayCacheScope ?? "global"'); }); }); diff --git a/tests/server-key-failover-e2e.test.ts b/tests/server-key-failover-e2e.test.ts index 967647c62c..be0d56df7d 100644 --- a/tests/server-key-failover-e2e.test.ts +++ b/tests/server-key-failover-e2e.test.ts @@ -5,6 +5,7 @@ import { join } from "node:path"; import { saveConfig } from "../src/config"; import { clearKeyCooldowns } from "../src/providers/key-failover"; import { deriveXaiConvId } from "../src/providers/xai-transport"; +import { clearReasoningReplayCacheForTests } from "../src/responses/reasoning-replay-cache"; import { startServer } from "../src/server"; import type { OcxConfig } from "../src/types"; import { installIsolatedCodexHome, type IsolatedCodexHome } from "./helpers/isolated-codex-home"; @@ -20,6 +21,7 @@ beforeEach(() => { testDir = mkdtempSync(join(tmpdir(), "ocx-keyfail-e2e-")); process.env.OPENCODEX_HOME = testDir; clearKeyCooldowns(); + clearReasoningReplayCacheForTests(); }); afterEach(() => { @@ -31,6 +33,7 @@ afterEach(() => { isolatedCodexHome = null; if (testDir) rmSync(testDir, { recursive: true, force: true }); clearKeyCooldowns(); + clearReasoningReplayCacheForTests(); }); describe("server 429 key failover (end-to-end)", () => { @@ -245,6 +248,124 @@ describe("server 429 key failover (end-to-end)", () => { } }); + test("reasoning replay misses after a 429 rotates to a different physical key", async () => { + const model = "reasoning-model"; + const callId = "call_key_rotation"; + const privateReasoning = "reasoning from physical key A"; + const seen: Array<{ auth: string; messages: Array> }> = []; + upstream = Bun.serve({ + hostname: "127.0.0.1", port: 0, + async fetch(req) { + const body = await req.json() as { messages?: Array> }; + seen.push({ + auth: req.headers.get("authorization") ?? "", + messages: body.messages ?? [], + }); + if (seen.length === 1) { + return Response.json({ + id: "chatcmpl-reasoning-seed", + object: "chat.completion", + choices: [{ + index: 0, + message: { + role: "assistant", + content: null, + reasoning_content: privateReasoning, + tool_calls: [{ + id: callId, + type: "function", + function: { name: "read_file", arguments: "{}" }, + }], + }, + finish_reason: "tool_calls", + }], + }); + } + if (seen.length === 2) { + return Response.json( + { error: { message: "rotate key" } }, + { status: 429, headers: { "retry-after": "30" } }, + ); + } + return Response.json({ + id: "chatcmpl-reasoning-rotated", + object: "chat.completion", + choices: [{ + index: 0, + message: { role: "assistant", content: "ok after isolated retry" }, + finish_reason: "stop", + }], + }); + }, + }); + const config: OcxConfig = { + port: 0, hostname: "127.0.0.1", defaultProvider: "reasoning-pool", + providers: { + "reasoning-pool": { + adapter: "openai-chat", + baseUrl: `http://127.0.0.1:${upstream.port}/v1`, + allowPrivateNetwork: true, + apiKey: "key-alpha-000111222333", + apiKeyPool: [ + { id: "k1", key: "key-alpha-000111222333", addedAt: 1 }, + { id: "k2", key: "key-beta-444555666777", addedAt: 2 }, + ], + preserveReasoningContentModels: [model], + requiresReasoningPlaceholderModels: [model], + }, + }, + } as OcxConfig; + saveConfig(config); + const server = startServer(0); + const headers = { + "content-type": "application/json", + "x-codex-parent-thread-id": "thread-key-rotation", + }; + try { + const first = await fetch(new URL("/v1/responses", server.url), { + method: "POST", + headers, + body: JSON.stringify({ + model: `reasoning-pool/${model}`, + input: "inspect the repo", + stream: false, + tools: [{ type: "function", name: "read_file", parameters: { type: "object" } }], + }), + }); + expect(first.status).toBe(200); + await first.json(); + + const second = await fetch(new URL("/v1/responses", server.url), { + method: "POST", + headers, + body: JSON.stringify({ + model: `reasoning-pool/${model}`, + stream: false, + input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "inspect the repo" }] }, + { type: "function_call", call_id: callId, name: "read_file", arguments: "{}" }, + { type: "function_call_output", call_id: callId, output: "contents" }, + ], + }), + }); + expect(second.status).toBe(200); + await second.json(); + + expect(seen.map(entry => entry.auth)).toEqual([ + "Bearer key-alpha-000111222333", + "Bearer key-alpha-000111222333", + "Bearer key-beta-444555666777", + ]); + const replayed = seen[1]!.messages.find(message => Array.isArray(message.tool_calls)); + const rotated = seen[2]!.messages.find(message => Array.isArray(message.tool_calls)); + expect(replayed?.reasoning_content).toBe(privateReasoning); + expect(rotated?.reasoning_content).toBe(" "); + expect(rotated?.reasoning_content).not.toBe(privateReasoning); + } finally { + await server.stop(true); + } + }); + test("network failure after a 429 key rotation surfaces the retry error", async () => { const originalFetch = globalThis.fetch; let upstreamAttempts = 0; diff --git a/tests/web-search.test.ts b/tests/web-search.test.ts index ae536e04ae..06970e6c70 100644 --- a/tests/web-search.test.ts +++ b/tests/web-search.test.ts @@ -1187,6 +1187,16 @@ describe("web-search sidecar native web_search_call emission", () => { const parsed = parseRequest({ model: "routed/model", input: "look up docs", stream: true, tools: [{ type: "web_search" }] }); parsed._clientThreadId = "web-search-raw-replay"; + parsed._reasoningReplayScope = { + clientThreadId: parsed._clientThreadId, + current: { + providerName: "routed", + providerDestinationIdentity: "destination:routed", + adapterName: adapter.name, + modelId: "model", + credentialIdentity: "key:test", + }, + }; const response = await runWithWebSearch({ parsed, adapter, @@ -1386,6 +1396,16 @@ describe("web-search sidecar native web_search_call emission", () => { const parsed = parseRequest({ model: "deepseek-v4-flash", input: "look up docs", stream: true, tools: [{ type: "web_search" }] }); parsed._clientThreadId = "web-search-deepseek-replay"; + parsed._reasoningReplayScope = { + clientThreadId: parsed._clientThreadId, + current: { + providerName: "routed", + providerDestinationIdentity: "destination:deepseek", + adapterName: "openai-chat", + modelId: "deepseek-v4-flash", + credentialIdentity: "key:test", + }, + }; const response = await runWithWebSearch({ parsed, adapter: createOpenAIChatAdapter(deepseekProvider),