diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index c5405b1c5a..68be694683 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -1696,7 +1696,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): if (!isCanonicalOpenAiForwardProvider(provider)) { outBody = promoteClientLoadedTools(outBody); } - if (provider.authMode !== "forward") { + if (!isCanonicalOpenAiForwardProvider(provider)) { const rewritten = rewriteRoutedCustomToolsForUpstream(outBody); outBody = rewritten.body; convertedRoutedCustomToolNames = rewritten.names; diff --git a/src/providers/openai-tiers.ts b/src/providers/openai-tiers.ts index 250cd3ca67..5e99c89963 100644 --- a/src/providers/openai-tiers.ts +++ b/src/providers/openai-tiers.ts @@ -1,5 +1,6 @@ import type { CodexAccountMode, OcxConfig, OcxProviderConfig, ProviderCostOverlay } from "../types"; import { OPENAI_PROVIDER_TIER_VERSION } from "../types"; +import { openaiResponsesUrl } from "../adapters/openai-responses-url"; import { MAX_COST4_RATE } from "../usage/expected-prices"; export const OPENAI_CODEX_PROVIDER_ID = "openai"; @@ -36,7 +37,35 @@ export function isCanonicalOpenAiForwardProvider(provider: OcxProviderConfig): b && normalizedBaseUrl(provider.baseUrl) === CODEX_FORWARD_BASE_URL; } -const OPENAI_API_BASE_URL = "https://api.openai.com/v1"; +const OPENAI_API_ORIGIN = "https://api.openai.com"; +const OPENAI_API_BASE_URL = `${OPENAI_API_ORIGIN}/v1`; +const OPENAI_API_RESPONSES_URL = `${OPENAI_API_BASE_URL}/responses`; + +/** + * The Responses endpoint the adapter would actually POST key-auth traffic to, normalized. + * + * Mirrors the adapter's own construction (`src/adapters/openai-responses.ts`): a configured + * `responsesPath` is appended to the base verbatim, and only the default branch runs the + * `/v1/responses` suffix normalization. Classifying on the base URL alone would call + * `baseUrl: "https://api.openai.com"` with `responsesPath: "/other"` official even though that + * request never reaches the official Responses endpoint. + */ +function resolvedResponsesEndpoint(provider: OcxProviderConfig): string | undefined { + try { + const raw = provider.responsesPath === undefined + ? openaiResponsesUrl(provider.baseUrl) + : `${provider.baseUrl.replace(/\/$/, "")}${provider.responsesPath}`; + return normalizedBaseUrl(raw); + } catch { + return undefined; + } +} + +function isOfficialOpenAiResponsesDestination(provider: OcxProviderConfig): boolean { + // Exact normalized URL keeps lookalike/suffix hosts out of this set: `api.openai.com.evil.test` + // resolves to its own origin, never to the official one. + return resolvedResponsesEndpoint(provider) === OPENAI_API_RESPONSES_URL; +} /** * Whether this provider can serve `POST /responses/compact`. The canonical ChatGPT @@ -65,7 +94,7 @@ export function supportsNativeResponsesCompactEndpoint( export function isOpenAiOperatedResponsesDestination(provider: OcxProviderConfig): boolean { if (isCanonicalOpenAiForwardProvider(provider)) return true; return provider.adapter === "openai-responses" - && normalizedBaseUrl(provider.baseUrl) === OPENAI_API_BASE_URL; + && isOfficialOpenAiResponsesDestination(provider); } /** diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts index c438f380c6..4b2d6d9166 100644 --- a/src/responses/reasoning-replay-cache.ts +++ b/src/responses/reasoning-replay-cache.ts @@ -133,32 +133,51 @@ function sweepExpiredServingIdentities(at: number): void { } } +function servingIdentityFor( + scope: OcxReasoningReplayScopeRef | undefined, +): { threadId: 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) }; +} + /** - * Compare this request's route with the last route recorded for its client thread, then - * record the current route. A live mismatch means replayed opaque reasoning was minted by - * another backend and must not be forwarded to this one. + * Compare this request's route with the last successfully serving route for its client thread. + * A live mismatch means replayed opaque reasoning was minted by another backend and must not be + * forwarded to this one. Comparison deliberately does not refresh or replace the recorded route: + * a failed candidate request did not serve the thread. * * Serving provenance uses restart-stable destination and credential dimensions so token * generations and other volatile credential material cannot create false route changes. Missing * durable identity, expired, or evicted state is deliberately unknown rather than a mismatch. * This store is process-local, so a backend switch spanning a proxy restart is not detected. */ -export function updateReasoningReplayServingIdentity( +export function reasoningReplayServingIdentityChanged( scope: OcxReasoningReplayScopeRef | undefined, ): boolean { - const threadId = scope?.clientThreadId; - const identityTuple = tupleForServingIdentity(scope?.current); - if (!nonEmpty(threadId) || !identityTuple) return false; - const identity = JSON.stringify(identityTuple); + const current = servingIdentityFor(scope); + if (!current) return false; + const at = now(); + sweepExpiredServingIdentities(at); + const previous = servingIdentities.get(current.threadId); + return previous !== undefined && previous.identity !== current.identity; +} +/** Record the route only after it has successfully served the client thread. */ +export function commitReasoningReplayServingIdentity( + scope: OcxReasoningReplayScopeRef | undefined, +): void { + const current = servingIdentityFor(scope); + if (!current) return; const at = now(); sweepExpiredServingIdentities(at); - const previous = servingIdentities.get(threadId); - const changed = previous !== undefined && previous.identity !== identity; + const previous = servingIdentities.get(current.threadId); + const { threadId, identity } = current; const bytes = Buffer.byteLength(JSON.stringify([threadId, identity]), "utf8"); if (bytes > MAX_TOTAL_BYTES) { deleteServingIdentity(threadId); - return false; + return; } if (previous) deleteServingIdentity(threadId); @@ -179,7 +198,6 @@ export function updateReasoningReplayServingIdentity( if (oldestThreadId === undefined) break; deleteServingIdentity(oldestThreadId); } - return changed; } function processLocalIdentity(domain: string, material: string): string { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 0199203755..efaa2fa2dc 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -16,13 +16,14 @@ import { import { parseRequest } from "../../responses/parser"; import { bindReasoningReplayScope, + commitReasoningReplayServingIdentity, reasoningReplayCodexCredentialIdentity, reasoningReplayDestinationIdentity, durableReplayDestinationIdentity, durableReplayCredentialIdentity, reasoningReplayKeyCredentialIdentity, reasoningReplayOAuthCredentialIdentity, - updateReasoningReplayServingIdentity, + reasoningReplayServingIdentityChanged, } from "../../responses/reasoning-replay-cache"; import { awaitThoughtSignatureDurability, thoughtSignatureReplaySalt } from "../../responses/thought-signature-replay"; import { buildCompactV1Output, COMPACT_PROMPT, decodeCompactionSummary, extractCompactUserMessages } from "../../responses/compaction"; @@ -510,12 +511,20 @@ function bindRouteReasoningReplayScope(args: { ); // Keep this sticky for the whole outbound request: a later auth/key rebind may compare equal // after the first mismatch, but it cannot make history minted by the prior route decodable. - if (updateReasoningReplayServingIdentity(parsed._reasoningReplayScope)) { + if (reasoningReplayServingIdentityChanged(parsed._reasoningReplayScope)) { parsed._stripReasoningEncryptedContent = true; } bindProviderContinuationForRoute(parsed, continuationOwner); } +function adapterResponseReachedServingTerminal( + events: readonly AdapterEvent[], + response: Readonly>, +): boolean { + return (response.status === "completed" || response.status === "incomplete") + && events.some(event => event.type === "done" || event.type === "incomplete"); +} + const OPAQUE_RESPONSES_INPUT_TYPES = new Set([ "reasoning", "compaction", @@ -2755,6 +2764,9 @@ async function handleResponsesInner( // message, and leave Codex fataling on a missing compaction item (#422). const routedCompaction = parsed._compactionRequest === true && !isCanonicalOpenAiForwardProvider(route.provider); + const commitReasoningReplayServingRoute = (): void => { + commitReasoningReplayServingIdentity(parsed._reasoningReplayScope); + }; if (routedCompaction) { delete parsed.context.tools; delete parsed._webSearch; @@ -2772,6 +2784,11 @@ async function handleResponsesInner( parsed.context.messages.push({ role: "user", content: COMPACT_PROMPT, timestamp: Date.now() }); } + let routedNamespaceToolAliases: RoutedNamespaceToolAliases = new Map(); + const refreshRoutedNamespaceToolAliases = (builtRequest: AdapterRequest): void => { + routedNamespaceToolAliases = builtRequest.convertedRoutedNamespaceToolAliases ?? new Map(); + }; + if ("passthrough" in adapter && adapter.passthrough && !routedCompaction) { let hostAdmissionLease = pendingHostAdmissionLease; pendingHostAdmissionLease = null; @@ -2781,7 +2798,6 @@ async function handleResponsesInner( : imageGenToolCallAliases(toolBridgeMaps.toolNsMap, parsed._rawBody, translatorBudget); const routedCustomToolNames = new Set(); const routedToolSearchNames = new Set(); - let routedNamespaceToolAliases: RoutedNamespaceToolAliases = new Map(); // Local continuation cache for the ChatGPT passthrough. Codex WS turns chain with // previous_response_id, ocx converts them to internal HTTP requests, and the ChatGPT Codex // REST backend rejects the parameter — the adapter strips it in forward mode, so the ONLY @@ -2817,7 +2833,7 @@ async function handleResponsesInner( } throw error; } - if (route.provider.authMode !== "forward") { + if (!isCanonicalOpenAiForwardProvider(route.provider)) { for (const name of request.convertedRoutedCustomToolNames ?? []) { if ( toolBridgeMaps.freeformToolNames.has(name) @@ -2831,7 +2847,7 @@ async function handleResponsesInner( // would incorrectly disable restoration for the exact ambiguous-name case the alias fixes. routedToolSearchNames.add(name); } - routedNamespaceToolAliases = request.convertedRoutedNamespaceToolAliases ?? routedNamespaceToolAliases; + refreshRoutedNamespaceToolAliases(request); // #1700: the bridged paths refuse a call to a tool the request never declared // (`declaredToolNames`, src/bridge.ts). The passthrough had no equivalent, so a routed // provider's top-level `apply_patch` — which under Codex code mode exists only as a nested @@ -3045,6 +3061,7 @@ async function handleResponsesInner( headers: selectedForwardHeaders, translatorBudget, }); + refreshRoutedNamespaceToolAliases(request); recordAdapterReasoning(logCtx, request); recordAdapterTier(logCtx, request); } catch (err) { @@ -3158,6 +3175,7 @@ async function handleResponsesInner( headers: selectedForwardHeaders, translatorBudget, }); + refreshRoutedNamespaceToolAliases(request); recordAdapterReasoning(logCtx, request); recordAdapterTier(logCtx, request); } catch (err) { @@ -3316,6 +3334,7 @@ async function handleResponsesInner( if (retry.kind === "retried") { authCtx = retry.authCtx; request = retry.request; + refreshRoutedNamespaceToolAliases(request); upstreamResponse = retry.upstreamResponse; selectedForwardHeaders = retry.selectedForwardHeaders; // Keep subagent quota-failure health keyed to the account that actually served. @@ -3342,11 +3361,6 @@ async function handleResponsesInner( } break; } - // Binding normally records before the first send. Repeat it only after a successful recovery - // so an eviction during the extra round trip cannot leave the next cross-route turn cold. - if (opaqueBlobRecoveryGuard.attempted && upstreamResponse.ok) { - updateReasoningReplayServingIdentity(parsed._reasoningReplayScope); - } const headers = sanitizePassthroughHeaders(upstreamResponse.headers); const resolvedModel = headers.get("openai-model")?.trim(); if (resolvedModel) logCtx.resolvedModel = resolvedModel; @@ -3470,6 +3484,10 @@ async function handleResponsesInner( // devlog/_fin/260731_macos_rss_retention/100_darwin_eager_optin.md). // The bundled known-bad runtime remains on tee by default on both platforms. if (isEventStream && upstreamResponse.body) { + // For streamed passthrough, a successful terminal response means non-error upstream status + // before relay starts. Waiting for SSE completion would retain request state across the whole + // stream; a later body failure does not undo that this destination accepted and served the turn. + commitReasoningReplayServingRoute(); const terminalRepairPolicy = providerModelResponsesTerminalRepair( route.providerName, route.provider, @@ -3755,6 +3773,7 @@ async function handleResponsesInner( return formatErrorResponse(502, "upstream_error", undeclaredToolCallMessage(undeclared)); } } + commitReasoningReplayServingRoute(); if (rememberPassthroughResponseChecked) { try { rememberPassthroughResponseChecked( @@ -3836,6 +3855,9 @@ async function handleResponsesInner( headers, }); } + // An unclassified passthrough body is relayed directly and has no bounded completion observer; + // use the same non-error-status success boundary as SSE instead of retaining per-stream state. + commitReasoningReplayServingRoute(); const body = relayWithAbort(upstreamResponse.body, upstream); const turnAc = new AbortController(); const tracked = body ? trackStreamLifetime(body, turnAc, undefined, options.turnAdmissionLease) : null; @@ -3981,13 +4003,15 @@ async function handleResponsesInner( retryOn429Policy: rateLimitRetryPolicyFor(route.provider), ...(options.onFirstOutput ? { onFirstOutput: options.onFirstOutput } : {}), ...(options.forceEmptyResponseId ? { forceEmptyResponseId: true } : {}), - onCompletedResponse: (response, providerState) => + onCompletedResponse: (response, providerState) => { + commitReasoningReplayServingRoute(); rememberResponseState( parsed._rawBody, response, continuationStateForResponse(providerState), responseStateOptions(adapterNeedsForcedContinuation(adapter.name)), - ), + ); + }, }); if (imgResponse.body) { const imgTurnAc = new AbortController(); @@ -4065,6 +4089,7 @@ async function handleResponsesInner( return rotatedAdapter; }, retryOn429Policy: rateLimitRetryPolicyFor(route.provider), + onCompletedResponse: commitReasoningReplayServingRoute, }); // Register the sidecar stream as an active turn so drainAndShutdown waits for (or aborts) // in-flight web-search turns instead of skipping them during graceful shutdown. @@ -4228,15 +4253,17 @@ async function handleResponsesInner( if (logCtx.activeAttempt) logCtx.activeAttempt.usage = usage; } }, - ...(routedCompaction ? {} : { - onCompletedResponse: (response: Record, providerState?: OcxProviderContinuationState) => + onCompletedResponse: (response: Record, providerState?: OcxProviderContinuationState) => { + commitReasoningReplayServingRoute(); + if (!routedCompaction) { rememberResponseState( parsed._rawBody, response, continuationStateForResponse(providerState), responseStateOptions(adapterNeedsForcedContinuation(adapter.name)), - ), - }), + ); + } + }, }, ); const bridgeTurnAc = new AbortController(); @@ -4299,6 +4326,9 @@ async function handleResponsesInner( // buildResponseJSON; bound the durability window before the JSON becomes // externally visible. await awaitThoughtSignatureDurability(); + if (adapterResponseReachedServingTerminal(events, json)) { + commitReasoningReplayServingRoute(); + } return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } }); } @@ -4321,6 +4351,7 @@ async function handleResponsesInner( let inputTokenEstimate: number | undefined; try { initialRequest = await activeAdapter.buildRequest(parsed, { headers: selectedForwardHeaders, translatorBudget }); + refreshRoutedNamespaceToolAliases(initialRequest); recordAdapterReasoning(logCtx, initialRequest); recordAdapterTier(logCtx, initialRequest); inputTokenEstimate = typeof initialRequest.usageLog?.inputTokens === "number" @@ -4453,6 +4484,7 @@ async function handleResponsesInner( sameTargetParsed = parsed; sameTargetToken = transportToken; } + refreshRoutedNamespaceToolAliases(retryRequest); const retryEstimate = typeof retryRequest.usageLog?.inputTokens === "number" ? retryRequest.usageLog.inputTokens : undefined; @@ -4690,11 +4722,6 @@ async function handleResponsesInner( } break; } - // Binding normally records before the first send. Repeat it only after a successful recovery - // so an eviction during the extra round trip cannot leave the next cross-route turn cold. - if (opaqueBlobRecoveryGuard.attempted && upstreamResponse.ok) { - updateReasoningReplayServingIdentity(parsed._reasoningReplayScope); - } if (!upstreamResponse.ok) { if (options.comboAttempt) { // No pre-read guard: `consumeComboFailure` -> `readBoundedResponseBody` reads @@ -5105,18 +5132,20 @@ async function handleResponsesInner( if (logCtx.activeAttempt) logCtx.activeAttempt.usage = usage; } }, - // Compaction turns must NOT enter the continuation cache: _rawBody still holds the full - // PRE-compaction history, and a later previous_response_id expansion would rehydrate the - // giant stale chain Codex just replaced. - ...(routedCompaction ? {} : { - onCompletedResponse: (response: Record, providerState?: OcxProviderContinuationState) => + onCompletedResponse: (response: Record, providerState?: OcxProviderContinuationState) => { + commitReasoningReplayServingRoute(); + // Compaction turns must NOT enter the continuation cache: _rawBody still holds the full + // PRE-compaction history, and a later previous_response_id expansion would rehydrate the + // giant stale chain Codex just replaced. + if (!routedCompaction) { rememberResponseState( parsed._rawBody, response, continuationStateForResponse(providerState), responseStateOptions(activeAdapter.name === "kiro"), - ), - }), + ); + } + }, }, ); const bridgeTurnAc = new AbortController(); @@ -5191,6 +5220,9 @@ async function handleResponsesInner( } // #1926 gap 2: same buffered-path durability bound as the primary branch. await awaitThoughtSignatureDurability(); + if (adapterResponseReachedServingTerminal(events, json)) { + commitReasoningReplayServingRoute(); + } return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } }); } diff --git a/src/types/request.ts b/src/types/request.ts index 7d7ce480dd..0dcfbbb8c4 100644 --- a/src/types/request.ts +++ b/src/types/request.ts @@ -65,7 +65,10 @@ export interface OcxParsedRequest { _clientThreadId?: string; /** Provider/account/model-bound namespace for process-local raw-reasoning replay. */ _reasoningReplayScope?: OcxReasoningReplayScopeRef; - /** A known in-process route switch requires opaque Responses reasoning blobs to be dropped. */ + /** + * Set by bindRouteReasoningReplayScope after a proven serving-identity change, or by + * prepareOpaqueBlobRecovery after an authoritative rejection; consumers strip replayed blobs. + */ _stripReasoningEncryptedContent?: boolean; /** * Optional authenticated tenant/operator namespace for Cursor thread→conversation derivation. diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index 19cb81563c..18800555b8 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -310,6 +310,8 @@ export interface WebSearchLoopDeps { on429?: (retryAfterHeader: string | null) => ProviderAdapter | null; /** Opt-in same-target 429 policy (key-auth providers). When present, 429 replays on the SAME key before on429 rotation. */ retryOn429Policy?: Required | null; + /** Called only when the final bridged Responses stream reaches completed or incomplete. */ + onCompletedResponse?: (response: Record) => void; } /** @@ -884,6 +886,7 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise { expect(isCanonicalOpenAiForwardProvider({ ...canonical, baseUrl: `${canonical.baseUrl}?x=1` })).toBe(false); }); + test("classifies only exact official OpenAI Responses destinations", () => { + const responsesProvider = { + adapter: "openai-responses", + authMode: "key" as const, + apiKey: "sk-test", + }; + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + baseUrl: "https://api.openai.com/v1", + })).toBe(true); + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + baseUrl: "https://api.openai.com", + responsesPath: "/v1/responses", + })).toBe(true); + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + baseUrl: "https://gateway.example.test/v1", + })).toBe(false); + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + baseUrl: "https://api.openai.com.evil.test/v1", + })).toBe(false); + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + adapter: "openai-chat", + baseUrl: "https://api.openai.com/v1", + })).toBe(false); + // The adapter sends key-auth traffic to `baseUrl + responsesPath`, so an official-looking base + // pointed at a non-Responses path is not an OpenAI-operated Responses destination. + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + baseUrl: "https://api.openai.com", + responsesPath: "/other", + })).toBe(false); + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + baseUrl: "https://api.openai.com/v1", + responsesPath: "/other", + })).toBe(false); + // The conventional defaults still classify: no `responsesPath` resolves to `/v1/responses`. + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + baseUrl: "https://api.openai.com", + })).toBe(true); + expect(isOpenAiOperatedResponsesDestination({ + ...responsesProvider, + baseUrl: "https://api.openai.com/v1", + responsesPath: "/responses", + })).toBe(true); + }); + test("publishes one Codex-login registry, preset, init, and default row", () => { for (const rows of [listRegistryEntries(), deriveProviderPresets(), deriveInitProviders()]) { expect(rows.some(entry => entry.id === LEGACY_OPENAI_MULTI_PROVIDER_ID)).toBe(false); diff --git a/tests/openai-responses-passthrough.test.ts b/tests/openai-responses-passthrough.test.ts index 61c40c66b5..c37269bcaa 100644 --- a/tests/openai-responses-passthrough.test.ts +++ b/tests/openai-responses-passthrough.test.ts @@ -2510,6 +2510,58 @@ describe("routed namespace and custom-tool identity", () => { globalThis.fetch = savedFetch; } }); + + test("noncanonical forward lowers a namespaced custom tool and restores its response identity", async () => { + const forwardConfig = { + port: 0, + defaultProvider: "fixture", + providers: { + fixture: { + adapter: "openai-responses", + baseUrl: "https://forward-gateway.example.test/v1", + authMode: "forward", + }, + }, + } as OcxConfig; + const savedFetch = globalThis.fetch; + let outbound: { tools?: Array> } | undefined; + globalThis.fetch = (async (_input, init) => { + outbound = JSON.parse(String(init?.body)) as { tools?: Array> }; + return Response.json({ + id: "resp_forward_custom", + status: "completed", + output: [customUpstreamItem], + }); + }) as typeof fetch; + + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "fixture/routed-model", + stream: false, + input: "read", + tools: [rawTools[0]], + }), + }), forwardConfig, { model: "", provider: "" }); + const body = await response.json() as { output: Array> }; + + expect(outbound?.tools).toEqual([ + expect.objectContaining({ type: "function", name: `${customNamespace}__read` }), + ]); + expect(outbound?.tools?.[0]).not.toHaveProperty("format"); + expect(body.output[0]).toMatchObject({ + type: "custom_tool_call", + namespace: customNamespace, + name: "read", + input: "freeform payload", + }); + expect(body.output[0]).not.toHaveProperty("arguments"); + } finally { + globalThis.fetch = savedFetch; + } + }); }); describe("OpenAI Responses forward-mode unsupported param stripping", () => { @@ -2757,6 +2809,7 @@ describe("reasoning input content channel", () => { for (const target of [ { adapter: "openai-responses", baseUrl: "https://chatgpt.com/backend-api/codex", authMode: "forward" as const }, { adapter: "openai-responses", baseUrl: "https://api.openai.com/v1", authMode: "key" as const, apiKey: "sk-t" }, + { adapter: "openai-responses", baseUrl: "https://api.openai.com", responsesPath: "/v1/responses", authMode: "key" as const, apiKey: "sk-t" }, ]) { const request = createResponsesPassthroughAdapter(target).buildRequest({ modelId: "gpt-5.6-sol", diff --git a/tests/reasoning-replay-identity.test.ts b/tests/reasoning-replay-identity.test.ts index 8f204b8141..834c23b5ea 100644 --- a/tests/reasoning-replay-identity.test.ts +++ b/tests/reasoning-replay-identity.test.ts @@ -3,14 +3,15 @@ import { bridgeToResponsesSSE } from "../src/bridge"; import { bindReasoningReplayScope, clearReasoningReplayCacheForTests, + commitReasoningReplayServingIdentity, peekReasoningForCall, reasoningReplayCodexCredentialIdentity, reasoningReplayCredentialIdentity, reasoningReplayDestinationIdentity, reasoningReplayKeyCredentialIdentity, reasoningReplayOAuthCredentialIdentity, + reasoningReplayServingIdentityChanged, rememberReasoningForCall, - updateReasoningReplayServingIdentity, } from "../src/responses/reasoning-replay-cache"; import type { AdapterEvent, OcxReasoningReplayScopeRef } from "../src/types"; @@ -73,10 +74,12 @@ describe("reasoning replay provider and credential identity", () => { }); test("serving identity ignores credential generation but reports durable route changes", () => { - expect(updateReasoningReplayServingIdentity(scope({ + const firstGeneration = scope({ credentialIdentity: "oauth:slot-a-generation-a", - }))).toBe(false); - expect(updateReasoningReplayServingIdentity(scope({ + }); + expect(reasoningReplayServingIdentityChanged(firstGeneration)).toBe(false); + commitReasoningReplayServingIdentity(firstGeneration); + expect(reasoningReplayServingIdentityChanged(scope({ credentialIdentity: "oauth:slot-a-generation-b", }))).toBe(false); @@ -84,16 +87,19 @@ describe("reasoning replay provider and credential identity", () => { modelId: "deepseek-v4", credentialIdentity: "oauth:slot-a-generation-b", }); - expect(updateReasoningReplayServingIdentity(changedModel)).toBe(true); - expect(updateReasoningReplayServingIdentity(changedModel)).toBe(false); + expect(reasoningReplayServingIdentityChanged(changedModel)).toBe(true); + expect(reasoningReplayServingIdentityChanged(changedModel)).toBe(true); + commitReasoningReplayServingIdentity(changedModel); + expect(reasoningReplayServingIdentityChanged(changedModel)).toBe(false); const changedCredential = scope({ modelId: "deepseek-v4", credentialIdentity: "oauth:slot-b-generation-a", credentialDurableIdentity: "credential:durable-slot-b", }); - expect(updateReasoningReplayServingIdentity(changedCredential)).toBe(true); - expect(updateReasoningReplayServingIdentity(changedCredential)).toBe(false); + expect(reasoningReplayServingIdentityChanged(changedCredential)).toBe(true); + commitReasoningReplayServingIdentity(changedCredential); + expect(reasoningReplayServingIdentityChanged(changedCredential)).toBe(false); const changedDestination = scope({ modelId: "deepseek-v4", @@ -102,30 +108,35 @@ describe("reasoning replay provider and credential identity", () => { credentialIdentity: "oauth:slot-b-generation-a", credentialDurableIdentity: "credential:durable-slot-b", }); - expect(updateReasoningReplayServingIdentity(changedDestination)).toBe(true); - expect(updateReasoningReplayServingIdentity(changedDestination)).toBe(false); + expect(reasoningReplayServingIdentityChanged(changedDestination)).toBe(true); + commitReasoningReplayServingIdentity(changedDestination); + expect(reasoningReplayServingIdentityChanged(changedDestination)).toBe(false); - expect(updateReasoningReplayServingIdentity(undefined)).toBe(false); - expect(updateReasoningReplayServingIdentity({ clientThreadId: "thread-unknown" })).toBe(false); + expect(reasoningReplayServingIdentityChanged(undefined)).toBe(false); + expect(reasoningReplayServingIdentityChanged({ clientThreadId: "thread-unknown" })).toBe(false); }); test("serving identity refuses to record when durable dimensions are unavailable", () => { const clientThreadId = "thread-without-durable-identity"; - expect(updateReasoningReplayServingIdentity({ + const withoutCredential = { ...scope({ credentialDurableIdentity: undefined }), clientThreadId, - })).toBe(false); - expect(updateReasoningReplayServingIdentity({ + }; + expect(reasoningReplayServingIdentityChanged(withoutCredential)).toBe(false); + commitReasoningReplayServingIdentity(withoutCredential); + expect(reasoningReplayServingIdentityChanged({ ...scope({ modelId: "different-model" }), clientThreadId, })).toBe(false); const destinationThreadId = "thread-without-durable-destination"; - expect(updateReasoningReplayServingIdentity({ + const withoutDestination = { ...scope({ providerDestinationDurableIdentity: undefined }), clientThreadId: destinationThreadId, - })).toBe(false); - expect(updateReasoningReplayServingIdentity({ + }; + expect(reasoningReplayServingIdentityChanged(withoutDestination)).toBe(false); + commitReasoningReplayServingIdentity(withoutDestination); + expect(reasoningReplayServingIdentityChanged({ ...scope({ modelId: "different-model" }), clientThreadId: destinationThreadId, })).toBe(false); @@ -134,10 +145,11 @@ describe("reasoning replay provider and credential identity", () => { test("expired serving identity is unknown rather than a backend change", () => { let clock = 1_000; clearReasoningReplayCacheForTests(() => clock); - expect(updateReasoningReplayServingIdentity(scope())).toBe(false); + expect(reasoningReplayServingIdentityChanged(scope())).toBe(false); + commitReasoningReplayServingIdentity(scope()); clock += 60 * 60 * 1000 + 1; - expect(updateReasoningReplayServingIdentity(scope({ modelId: "deepseek-v4" }))).toBe(false); + expect(reasoningReplayServingIdentityChanged(scope({ modelId: "deepseek-v4" }))).toBe(false); }); test("repeated identity changes do not grow the thread store beyond 64 entries", () => { @@ -150,15 +162,21 @@ describe("reasoning replay provider and credential identity", () => { }); for (let i = 0; i < 64; i++) { - expect(updateReasoningReplayServingIdentity(servingScope(`thread-${i}`, "model-a"))).toBe(false); + const candidate = servingScope(`thread-${i}`, "model-a"); + expect(reasoningReplayServingIdentityChanged(candidate)).toBe(false); + commitReasoningReplayServingIdentity(candidate); } for (let i = 0; i < 70; i++) { - expect(updateReasoningReplayServingIdentity(servingScope("thread-63", `model-change-${i}`))).toBe(true); + const candidate = servingScope("thread-63", `model-change-${i}`); + expect(reasoningReplayServingIdentityChanged(candidate)).toBe(true); + commitReasoningReplayServingIdentity(candidate); } - expect(updateReasoningReplayServingIdentity(servingScope("thread-64", "model-a"))).toBe(false); - expect(updateReasoningReplayServingIdentity(servingScope("thread-1", "model-b"))).toBe(true); - expect(updateReasoningReplayServingIdentity(servingScope("thread-0", "model-b"))).toBe(false); + const added = servingScope("thread-64", "model-a"); + expect(reasoningReplayServingIdentityChanged(added)).toBe(false); + commitReasoningReplayServingIdentity(added); + expect(reasoningReplayServingIdentityChanged(servingScope("thread-1", "model-b"))).toBe(true); + expect(reasoningReplayServingIdentityChanged(servingScope("thread-0", "model-b"))).toBe(false); }); test("incomplete, unscoped, and legacy thread-only namespaces fail closed", () => { diff --git a/tests/responses-opaque-blob-recovery.test.ts b/tests/responses-opaque-blob-recovery.test.ts index b547d322a3..4502c0b340 100644 --- a/tests/responses-opaque-blob-recovery.test.ts +++ b/tests/responses-opaque-blob-recovery.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { ADAPTER_REGISTRY } from "../src/adapters/registry"; import { clearReasoningReplayCacheForTests } from "../src/responses/reasoning-replay-cache"; import { OPAQUE_COMPACTION_NOTE } from "../src/responses/compaction"; import { resetThoughtSignatureReplayForTests } from "../src/responses/thought-signature-replay"; @@ -210,6 +211,80 @@ describe("opaque blob recovery through /v1/responses", () => { expect(logCtx.activeAttempt?.recoveryKinds).toEqual(["opaque-blob-rejection"]); }); + test("restores namespace names from the rebuilt request alias set", async () => { + const definition = ADAPTER_REGISTRY["openai-responses"] as unknown as { + create: typeof ADAPTER_REGISTRY["openai-responses"]["create"]; + }; + const originalCreate = definition.create; + let buildCount = 0; + definition.create = (provider, context) => { + const adapter = originalCreate(provider, context); + const buildRequest = adapter.buildRequest.bind(adapter); + adapter.buildRequest = async (parsed, incoming) => { + const built = await buildRequest(parsed, incoming); + buildCount += 1; + built.convertedRoutedNamespaceToolAliases = buildCount === 1 + ? new Map([["stale_catalog__read", { namespace: "stale_catalog", name: "read" }]]) + : new Map([["fresh_catalog__read", { namespace: "fresh_catalog", name: "read" }]]); + return built; + }; + return adapter; + }; + + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + if (outbound.length === 1) return rejection(); + return Response.json({ + id: "resp-rebuilt-aliases", + status: "completed", + output: [{ + type: "function_call", + id: "fc_fresh_read", + call_id: "call_fresh_read", + name: "fresh_catalog__read", + arguments: "{}", + status: "completed", + }], + }); + }) as typeof fetch; + + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { + "content-type": "application/json", + "x-codex-parent-thread-id": "thread-rebuilt-namespace-aliases", + }, + body: JSON.stringify({ + model: "first/model-a", + stream: false, + store: false, + input: reasoningReplayInput(), + tools: [{ + type: "namespace", + name: "fresh_catalog", + tools: [{ type: "function", name: "read", parameters: { type: "object" } }], + }], + }), + }), config(), { model: "", provider: "" }); + const body = await response.json() as { output: Array> }; + + expect(response.status).toBe(200); + expect(buildCount).toBe(2); + expect(outbound).toHaveLength(2); + expect(body.output[0]).toMatchObject({ + type: "function_call", + namespace: "fresh_catalog", + name: "read", + arguments: "{}", + }); + expect(JSON.stringify(body)).not.toContain("stale_catalog"); + } finally { + definition.create = originalCreate; + } + }); + test("degrades a compaction blob through the generic routed-compaction recovery resend", async () => { const outbound: Array> = []; globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { @@ -291,3 +366,70 @@ describe("opaque blob recovery through /v1/responses", () => { expect(hasBlob(outbound.get("second")![0]!)).toBe(false); }); }); + +describe("reasoning replay serving identity commit through /v1/responses", () => { + test("a failed A-to-B turn does not commit B, so the next B retry still strips A-minted blobs", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + if (outbound.length === 2) { + return new Response(JSON.stringify({ error: { message: "rate limited" } }), { + status: 429, + headers: { "content-type": "application/json" }, + }); + } + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + const first = await handleResponses(request("first", "thread-failed-switch"), config(), { model: "", provider: "" }); + expect(first.status).toBe(200); + await first.text(); + + const failedSwitch = await handleResponses(request("second", "thread-failed-switch"), config(), { model: "", provider: "" }); + expect(failedSwitch.status).toBe(429); + await failedSwitch.text(); + + const retry = await handleResponses(request("second", "thread-failed-switch"), config(), { model: "", provider: "" }); + expect(retry.status).toBe(200); + await retry.text(); + + expect(outbound).toHaveLength(3); + expect(hasBlob(outbound[0]!)).toBe(true); + expect(hasBlob(outbound[1]!)).toBe(false); + expect(hasBlob(outbound[2]!)).toBe(false); + }); + + test("a successful A-to-B turn commits B, so the following B turn keeps B-minted blobs", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success(`resp-${outbound.length}`); + }) as typeof fetch; + + for (const provider of ["first", "second", "second"]) { + const response = await handleResponses(request(provider, "thread-successful-switch"), config(), { model: "", provider: "" }); + expect(response.status).toBe(200); + await response.text(); + } + + expect(outbound).toHaveLength(3); + expect(hasBlob(outbound[0]!)).toBe(true); + expect(hasBlob(outbound[1]!)).toBe(false); + expect(hasBlob(outbound[2]!)).toBe(true); + }); + + test("a thread without a serving record keeps opaque blobs", async () => { + const outbound: Array> = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + outbound.push(JSON.parse(String(init?.body)) as Record); + return success("resp-cold-thread"); + }) as typeof fetch; + + const response = await handleResponses(request("first", "thread-without-record"), config(), { model: "", provider: "" }); + expect(response.status).toBe(200); + await response.text(); + + expect(outbound).toHaveLength(1); + expect(hasBlob(outbound[0]!)).toBe(true); + }); +});