diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 5f226f55b41..56b9fafe25f 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -384,7 +384,7 @@ "claude-gateway-cache.test.ts": "claude-integration", "claude-inbound-cache-stabilize.test.ts": "claude-integration", "claude-inbound-token-footer.test.ts": "claude-integration", "claude-inbound-debug.test.ts": "claude-integration", - "claude-inbound.test.ts": "claude-integration", "claude-inbound-tool-reference.test.ts": "claude-integration", + "claude-inbound.test.ts": "claude-integration", "claude-inbound-tool-reference.test.ts": "claude-integration", "claude-native-reasoning-continuity.test.ts": "claude-integration", "claude-native-reasoning-provenance.test.ts": "claude-integration", "claude-output-schema-strict.test.ts": "claude-integration", "claude-intercept-integration.test.ts": "server", "claude-intercept-local-ca.test.ts": "claude-integration", diff --git a/src/claude/inbound.ts b/src/claude/inbound.ts index 53047a35941..7378c6498ac 100644 --- a/src/claude/inbound.ts +++ b/src/claude/inbound.ts @@ -281,7 +281,7 @@ function userMessageToItems(content: unknown, input: Rec[], elide: SkillElisionC pushUserMessage(input, pending); } -function assistantMessageToItems(content: unknown, input: Rec[], budget: TranslatorBudget): void { +function assistantMessageToItems(content: unknown, input: Rec[], budget: TranslatorBudget, requestedModel: string, nativeReasoningReplay: Map): void { if (typeof content === "string") { if (content.length > 0) input.push({ type: "message", role: "assistant", content: [{ type: "output_text", text: content }] }); return; @@ -314,6 +314,24 @@ function assistantMessageToItems(content: unknown, input: Rec[], budget: Transla const owned = decodeReasoningEnvelope(signature, budget); if (!owned) throw new AnthropicRequestError("malformed ocxr1 reasoning signature"); if (Object.hasOwn(owned, "sig")) throw new AnthropicRequestError("OpenCodex reasoning continuity cannot be replayed as an Anthropic signature"); + if (owned.nat) { + // The provider's own blob goes back only to the model that minted it; any other + // model gets the visible text alone, as before this envelope field existed. + const nat = owned.nat.model === requestedModel ? owned.nat : undefined; + if (nat) { + const previous = nativeReasoningReplay.get(nat.enc); + nativeReasoningReplay.set(nat.enc, previous === undefined || previous === nat.tag ? nat.tag : ""); + } + if (nat) budget.chargeRetained(2 * nat.enc.length, { kind: "reasoning" }); + if (thinking.length === 0 && !nat) break; + input.push({ + type: "reasoning", + id: nat?.id ?? `rs_${crypto.randomUUID().replace(/-/g, "")}`, + summary: thinking.length > 0 ? [{ type: "summary_text", text: thinking }] : [], + ...(nat ? { encrypted_content: nat.enc } : {}), + }); + break; + } } const encrypted = signature.length === 0 ? undefined : signature.startsWith(OCX_REASONING_PREFIX) ? signature : encodeReasoningEnvelope({ sig: signature }, budget); if (encrypted) budget.chargeRetained(2 * encrypted.length, { kind: "reasoning" }); @@ -355,6 +373,7 @@ export type ClaudeCacheKeySource = "metadata" | "system" | null; export interface ClaudeInboundTranslation { body: Rec; cacheKeySource: ClaudeCacheKeySource; + nativeReasoningReplay: ReadonlyMap; } /** @@ -400,6 +419,7 @@ function translateAnthropicRequest( } const input: Rec[] = []; + const nativeReasoningReplay = new Map(); const systemParts: string[] = []; const topLevelSystem = systemToInstructions(raw.system); if (topLevelSystem !== undefined) systemParts.push(topLevelSystem); @@ -411,7 +431,7 @@ function translateAnthropicRequest( for (const msg of raw.messages) { if (!isRec(msg)) throw new AnthropicRequestError("each message must be an object"); if (msg.role === "user") userMessageToItems(msg.content, input, elide); - else if (msg.role === "assistant") assistantMessageToItems(msg.content, input, budget); + else if (msg.role === "assistant") assistantMessageToItems(msg.content, input, budget, raw.model, nativeReasoningReplay); else if (msg.role === "system") { const text = systemMessageText(msg.content); // Keep it where the client put it. `developer` is first-class in the Responses @@ -531,7 +551,61 @@ function translateAnthropicRequest( reasoning.effort = effortForThinkingBudget(thinking.budget_tokens); } body.reasoning = reasoning; + // store:false returns a provider's encrypted reasoning only on request. The outbound + // envelope (`nat`) carries it to the client so the next turn can hand it back. + body.include = ["reasoning.encrypted_content"]; } - return { body, cacheKeySource }; + return { body, cacheKeySource, nativeReasoningReplay }; +} + +/** + * Remove proxy-owned native blobs before sending a Messages body to native Anthropic. + * + * Every decode, and every re-encode on top of it, is reserved against the request's translator + * budget first (eight bytes per code unit, as `decodeReasoningEnvelope` does), and every new + * signature is charged as a retained copy, so a body of near-limit envelopes cannot allocate past the per-turn bound. + * Throws TranslatorBudgetExceededError when it would; callers answer 413. + */ +export function nativeAnthropicProjection(body: Rec, budget: TranslatorBudget): Rec { + if (!Array.isArray(body.messages)) return body; + let changed = false; + const messages = body.messages.flatMap(message => { + if (!isRec(message) || message.role !== "assistant" || !Array.isArray(message.content)) return [message]; + let localChanged = false; + const content = message.content.flatMap(block => { + if (!isRec(block) || block.type !== "thinking" || typeof block.signature !== "string" + || !block.signature.startsWith(OCX_REASONING_PREFIX)) return [block]; + // The decoded object stays alive while it is re-encoded, so its reservation is held until + // the new signature exists; the encode copies get their own reservation on top of it. + const decodeReservation = budget.reserveTransient(8 * block.signature.length, { kind: "reasoning" }); + try { + let decoded: unknown; + try { decoded = JSON.parse(Buffer.from(block.signature.slice(OCX_REASONING_PREFIX.length), "base64").toString("utf8")); } + catch { decoded = undefined; } + if (!isRec(decoded)) { localChanged = true; return []; } + if (!Object.hasOwn(decoded, "nat")) return [block]; + localChanged = true; + const { nat: _nat, ...rest } = decoded; + if (!rest.sig && !rest.red && !rest.krc && !(typeof rest.txt === "string" && rest.txt.length > 0)) return []; + // The re-encoded envelope is never larger than the decoded one (a field was removed), so a + // decode-sized reservation bounds the stringify, buffer and base64 copies before they exist. + const encodeReservation = budget.reserveTransient(8 * block.signature.length, { kind: "reasoning" }); + let signature: string; + try { + signature = OCX_REASONING_PREFIX + Buffer.from(JSON.stringify(rest), "utf8").toString("base64"); + } finally { + encodeReservation.release(); + } + budget.chargeRetained(2 * signature.length, { kind: "reasoning" }); + return [{ ...block, signature }]; + } finally { + decodeReservation.release(); + } + }); + if (!localChanged) return [message]; + changed = true; + return content.length ? [{ ...message, content }] : []; + }); + return changed ? { ...body, messages } : body; } diff --git a/src/claude/outbound.ts b/src/claude/outbound.ts index 9fc981ee72c..60924c92def 100644 --- a/src/claude/outbound.ts +++ b/src/claude/outbound.ts @@ -20,7 +20,7 @@ import { type TranslatorBudget, } from "../lib/translator-budget"; import { sseFieldOffset, sseFieldValue } from "../lib/sse-decoder"; -import { decodeReasoningEnvelope, encodeReasoningEnvelope } from "../responses/reasoning-envelope"; +import { decodeReasoningEnvelope, encodeReasoningEnvelope, OCX_REASONING_PREFIX } from "../responses/reasoning-envelope"; type Rec = Record; @@ -339,6 +339,13 @@ interface OpenBlock { thinkingBuf?: string; thinkingBufBytes?: number; reasoningSig?: string; + /** The provider's own encrypted reasoning for this item; signed into the block on close. */ + reasoningNative?: { enc: string; id?: string }; +} + +/** A provider-minted (not ocxr1) `encrypted_content`, or `""`. */ +function nativeEncryptedReasoning(encrypted: string): string { + return encrypted.length > 0 && !encrypted.startsWith(OCX_REASONING_PREFIX) ? encrypted : ""; } /** Streaming: Responses SSE bytes -> Anthropic Messages SSE bytes. */ @@ -353,6 +360,7 @@ export function responsesSseToAnthropicSse( * upstream sent no confirmed usage before the first frame. See `messageSnapshot` (#4857). */ inputTokenFloor?: number; + nativeReasoningTagFor?: (blob: string) => string | undefined; }, ): ReadableStream { const translatorBudget = opts.translatorBudget; @@ -441,6 +449,13 @@ export function responsesSseToAnthropicSse( open.webSearchArgsEmitted = true; } if (open.kind === "thinking") { + const native = open.reasoningNative; + const tag = native && opts.nativeReasoningTagFor?.(native.enc); + if (!open.thinkingBuf && !open.reasoningSig && !tag) { + releaseThinkingBuffer(open); + open = null; + return; + } // Delay the index and all thinking frames until closure so a matching // done envelope can put its redacted blocks first. The existing buffer // remains charged through signature emission, including queued frames. @@ -455,7 +470,10 @@ export function responsesSseToAnthropicSse( delta: { type: "thinking_delta", thinking: open.thinkingBuf }, }); } - const signature = open.reasoningSig ?? encodeReasoningEnvelope({ txt: open.thinkingBuf ?? "" }, translatorBudget); + const signature = open.reasoningSig ?? encodeReasoningEnvelope({ + txt: open.thinkingBuf ?? "", + ...(tag ? { nat: { ...native, model, tag } } : {}), + }, translatorBudget); emit("content_block_delta", { type: "content_block_delta", index: open.index, delta: { type: "signature_delta", signature }, @@ -743,11 +761,15 @@ export function responsesSseToAnthropicSse( emit("content_block_start", { type: "content_block_start", index: idx, content_block: { type: "redacted_thinking", data } }); emit("content_block_stop", { type: "content_block_stop", index: idx }); } - if (env?.sig && open?.kind !== "thinking") { + // A provider's own blob is the only copy of its reasoning the client can hold, so + // it gets a thinking block even when no summary text arrived. + const native = nativeEncryptedReasoning(encrypted); + if ((env?.sig || native) && open?.kind !== "thinking") { ensureBlock("thinking"); } if (open?.kind === "thinking") { if (env?.sig) open.reasoningSig = env.sig; + if (native) open.reasoningNative = typeof item.id === "string" && item.id.length > 0 ? { enc: native, id: item.id } : { enc: native }; closeOpenBlock(); } } @@ -940,7 +962,7 @@ export function responsesSseToAnthropicSse( } /** Non-streaming: /v1/responses JSON -> Anthropic message JSON. */ -export function responsesJsonToAnthropicMessage(json: unknown, model: string, translatorBudget?: TranslatorBudget): Rec { +export function responsesJsonToAnthropicMessage(json: unknown, model: string, translatorBudget?: TranslatorBudget, nativeReasoningTagFor?: (blob: string) => string | undefined): Rec { const body = isRec(json) ? json : {}; const output = Array.isArray(body.output) ? body.output : []; const content: Rec[] = []; @@ -976,10 +998,14 @@ export function responsesJsonToAnthropicMessage(json: unknown, model: string, tr // Legacy combined envelopes place redacted blocks before the signed block, // matching the Anthropic adapter. New bridge output uses separate items. for (const data of env?.red ?? []) content.push({ type: "redacted_thinking", data }); - // env.txt may be locally hidden text. Do not expose it here or manufacture - // a new signed continuity carrier; hidden-summary replay remains limited. - if (parts.length > 0 || env?.sig) { - content.push({ type: "thinking", thinking: parts.join("\n\n"), signature: env?.sig ?? encodeReasoningEnvelope({ txt: parts.join("\n\n") }, translatorBudget) }); + // env.txt may be locally hidden text. Do not expose it here. A provider's own blob is + // carried in the envelope so the next request can hand it back (see inbound.ts). + const native = nativeEncryptedReasoning(encrypted); + const tag = native && nativeReasoningTagFor?.(native); + if (parts.length > 0 || env?.sig || tag) { + const text = parts.join("\n\n"); + const nat = tag ? { enc: native, model, tag, ...(typeof raw.id === "string" && raw.id.length > 0 ? { id: raw.id } : {}) } : undefined; + content.push({ type: "thinking", thinking: text, signature: env?.sig ?? encodeReasoningEnvelope({ txt: text, ...(nat ? { nat } : {}) }, translatorBudget) }); } break; } diff --git a/src/responses/reasoning-envelope.ts b/src/responses/reasoning-envelope.ts index 60612864fdd..2e6b9d229e9 100644 --- a/src/responses/reasoning-envelope.ts +++ b/src/responses/reasoning-envelope.ts @@ -36,6 +36,32 @@ export interface ReasoningEnvelope { * same way a signature does. */ krc?: string; + /** + * A Responses provider's own `encrypted_content` for one reasoning item, minted while serving + * `model` (the client-facing model the Claude route was asked for). The Claude route has no + * other channel to hand a provider its reasoning back: the body is `store: false` and Claude + * Code replays only the thinking block. `src/claude/inbound.ts` restores it only for a request + * naming the same model, so a model switch cannot carry it to a provider that cannot decrypt it. + */ + nat?: NativeReasoning; +} + +export interface NativeReasoning { + /** The provider's opaque `encrypted_content`, verbatim. */ + enc: string; + /** Client-facing model the item was produced for. */ + model: string; + /** Route-bound HMAC for this exact ciphertext. */ + tag: string; + /** The provider's reasoning item id, when it sent one. */ + id?: string; +} + +function decodeNativeReasoning(value: unknown): NativeReasoning | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + const { enc, model, id, tag } = value as { enc?: unknown; model?: unknown; id?: unknown; tag?: unknown }; + if (typeof enc !== "string" || enc.length === 0 || typeof model !== "string" || model.length === 0 || typeof tag !== "string" || tag.length === 0) return undefined; + return typeof id === "string" && id.length > 0 ? { enc, model, tag, id } : { enc, model, tag }; } export function encodeReasoningEnvelope(envelope: ReasoningEnvelope, budget?: TranslatorBudget): string { @@ -116,7 +142,9 @@ export function decodeReasoningEnvelope(encryptedContent: string, budget?: Trans if (hasTxt) envelope.txt = txt; const krc = (parsed as { krc?: unknown }).krc; if (typeof krc === "string" && krc.length > 0) envelope.krc = krc; - return envelope.sig || envelope.red || hasTxt || envelope.krc ? envelope : null; + const nat = decodeNativeReasoning((parsed as { nat?: unknown }).nat); + if (nat) envelope.nat = nat; + return envelope.sig || envelope.red || hasTxt || envelope.krc || envelope.nat ? envelope : null; } catch { return null; } finally { diff --git a/src/responses/reasoning-replay-cache.ts b/src/responses/reasoning-replay-cache.ts index 1084a47f745..a481ffc35f0 100644 --- a/src/responses/reasoning-replay-cache.ts +++ b/src/responses/reasoning-replay-cache.ts @@ -19,7 +19,8 @@ * long-lived proxy cannot grow without limit. */ -import { createHash, createHmac, randomBytes } from "node:crypto"; +import { createHash, createHmac, randomBytes, timingSafeEqual } from "node:crypto"; +import { thoughtSignatureReplaySalt } from "./thought-signature-replay"; import type { OcxProviderConfig, OcxReasoningReplayIdentity, @@ -27,6 +28,21 @@ import type { } from "../types"; const MAX_ENTRIES = 64; +export interface NativeReasoningOwner { destination: string; credential: string } + +/** Bind exact ciphertext to the durable route that accepted it. */ +export function nativeReasoningTag(owner: NativeReasoningOwner | undefined, blob: string): string | undefined { + const salt = thoughtSignatureReplaySalt(); + if (!owner || !salt || salt.length < 16 || blob.length === 0) return undefined; + return createHmac("sha256", salt).update("native-reasoning\0") + .update(JSON.stringify([owner.destination, owner.credential, createHash("sha256").update(blob).digest("hex")])) + .digest("hex"); +} + +export function nativeReasoningTagMatches(expected: string | undefined, presented: string | undefined): boolean { + if (!expected || !presented || !/^[a-f0-9]{64}$/.test(expected) || !/^[a-f0-9]{64}$/.test(presented)) return false; + return timingSafeEqual(Buffer.from(expected, "hex"), Buffer.from(presented, "hex")); +} const MAX_TOTAL_BYTES = 256 * 1024; const TTL_MS = 60 * 60 * 1000; const OPAQUE_BLOB_REJECTION_TTL_MS = 5 * 60 * 1000; diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index 7b66ac91a2e..3bd6ce3a34e 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -21,7 +21,8 @@ import { normalizeAnthropicImages } from "../adapters/anthropic-image-normalize" import { createToolCallIdAllocator } from "../adapters/tool-call-id"; import { openAIChatSerializesThinking } from "../adapters/openai-chat/messages"; import { messagesToResponsesTranslation } from "../protocols/codecs/messages"; -import { AnthropicRequestError, DesktopModelMappingUnavailableError, extractOcxEffortDirective, extractOcxRouteDirective, resolveInboundModel, type ClaudeCacheKeySource } from "../claude/inbound"; +import { AnthropicRequestError, DesktopModelMappingUnavailableError, extractOcxEffortDirective, extractOcxRouteDirective, nativeAnthropicProjection, resolveInboundModel, type ClaudeCacheKeySource } from "../claude/inbound"; +import { nativeReasoningTag, type NativeReasoningOwner } from "../responses/reasoning-replay-cache"; import { isKnownDesktop3pModelId, resolveDesktop3pAlias } from "../claude/desktop-3p"; import { resolveAlias, claudeCodeNativeAlias, legacyAliasForNative } from "../claude/alias"; import { recordDesktopRequest } from "../claude/desktop-health"; @@ -872,6 +873,9 @@ async function handleClaudeMessagesWithBudget( let anthropicBody: unknown; let internalBody: Rec; let cacheKeySource: ClaudeCacheKeySource = null; + let nativeReasoningReplay: ReadonlyMap | undefined; + const nativeReasoningMint: { owner?: NativeReasoningOwner } = {}; + let nativeProjectionOnlyError: AnthropicRequestError | undefined; let effortOverride: string | null = null; let effortRow: ParsedEffortRowId | null = null; let fastRow: ParsedFastRowId | null = null; @@ -962,7 +966,15 @@ async function handleClaudeMessagesWithBudget( if (!effortRow && !fastRow && isRec(anthropicBody) && wantsNativePassthrough(req, config, requestPolicy, anthropicBody.model, cc)) { markProtocolEntry(logCtx, { inbound: "messages", lane: "native", features: messagesFeatures }); recordProtocolShadowPlan(logCtx, config, { inbound: "messages", model: requestedModel }); - return await anthropicNativePassthrough(req, config, logCtx, logIds, anthropicBody, "/v1/messages"); + let projectedBody: Rec; + try { + projectedBody = nativeAnthropicProjection(anthropicBody, translatorBudget); + } catch (err) { + if (!isTranslatorBudgetExceededError(err)) throw err; + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" }); + return anthropicErrorResponse(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); + } + return await anthropicNativePassthrough(req, config, logCtx, logIds, projectedBody, "/v1/messages"); } if (isRec(anthropicBody) && typeof anthropicBody.model === "string" && messagesSecondaryInstanceUnavailable(config, anthropicBody.model, cc)) { @@ -1017,7 +1029,19 @@ async function handleClaudeMessagesWithBudget( }; delete anthropicBody.thinking; } - const translation = messagesToResponsesTranslation(anthropicBody, cc, translatorBudget); + let translation; + try { + translation = messagesToResponsesTranslation(anthropicBody, cc, translatorBudget); + } catch (err) { + // A native Messages route can discard an undecodable proxy envelope. Keep the + // original error for any route that ultimately translates to Responses. + if (!(err instanceof AnthropicRequestError) || err.message !== "malformed ocxr1 reasoning signature" + || !isRec(anthropicBody)) throw err; + const projected = nativeAnthropicProjection(anthropicBody, translatorBudget); + if (projected === anthropicBody) throw err; + translation = messagesToResponsesTranslation(projected, cc, translatorBudget); + nativeProjectionOnlyError = err; + } internalBody = translation.body; // The Anthropic translator builds its body from model/input/store/stream plus sampling // fields only, so the caller intent is applied to the TRANSLATED body rather than the @@ -1025,6 +1049,7 @@ async function handleClaudeMessagesWithBudget( if (fastRow) internalBody.service_tier = "priority"; translatorBudget.chargeRetained(jsonUtf8Bytes(internalBody), { kind: "request_copies" }); cacheKeySource = translation.cacheKeySource; + nativeReasoningReplay = translation.nativeReasoningReplay; } catch (err) { const overflow = isTranslatorBudgetExceededError(err); const unavailable = err instanceof DesktopModelMappingUnavailableError; @@ -1171,6 +1196,10 @@ async function handleClaudeMessagesWithBudget( ? nativeMessagesDeclineReason(settledRoute, anthropicBody, config, nativeSelector) : "rollout-disabled"; const nativeMessagesRoute = settledRoute && nativeDecline === undefined ? settledRoute : undefined; + if (nativeProjectionOnlyError && !nativeMessagesRoute) { + if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 400, { closeReason: "non_stream" }); + return anthropicErrorResponse(400, nativeProjectionOnlyError.message, "invalid_request_error"); + } // With the switch on, a declined route says why on its bridge mark, as native Chat does. if (nativeDecline && nativeDecline !== "rollout-disabled") { markProtocolEntry(logCtx, { @@ -1209,7 +1238,7 @@ async function handleClaudeMessagesWithBudget( let nativeBody: Rec; try { // Built from the source envelope when there is one, after the managed-client steps above. - nativeBody = envelope ? envelope.freshBody() : anthropicBody as Rec; + nativeBody = nativeAnthropicProjection(envelope ? envelope.freshBody() : anthropicBody as Rec, translatorBudget); } catch (err) { if (!isTranslatorBudgetExceededError(err)) throw err; if (logIds) addFinalRequestLog(logIds.requestId, logIds.start, logCtx, 413, { closeReason: "non_stream" }); @@ -1314,6 +1343,8 @@ async function handleClaudeMessagesWithBudget( // Without this the replay would look native and a Responses-scoped wire default // would fire, disagreeing with the pre-flight decision above. inboundWire: "anthropic", + nativeReasoningReplay, + nativeReasoningMint, claudeGoAffinity: { sessionLane: claudeGoSessionLane }, claudeNativeSessionId, stripClaudeMainAuthForNoncanonicalForward: true, @@ -1395,6 +1426,7 @@ async function handleClaudeMessagesWithBudget( if (contentType.includes("text/event-stream") && response.body) { const anthropicSse = responsesSseToAnthropicSse(response.body, requestedModel, { translatorBudget, + nativeReasoningTagFor: blob => nativeReasoningTag(nativeReasoningMint.owner, blob), // Only a floor, and only for the first frame: an upstream that reports usage early wins // over it inside the translator, and the terminal `message_delta` carries the // authoritative count either way (#4857). @@ -1464,7 +1496,7 @@ async function handleClaudeMessagesWithBudget( } let message: Rec; try { - message = responsesJsonToAnthropicMessage(json, requestedModel, translatorBudget); + message = responsesJsonToAnthropicMessage(json, requestedModel, translatorBudget, blob => nativeReasoningTag(nativeReasoningMint.owner, blob)); } catch (err) { if (!isTranslatorBudgetExceededError(err)) throw err; return anthropicErrorResponse(413, "upstream translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); diff --git a/src/server/responses/core-options.ts b/src/server/responses/core-options.ts index 51c43ec7a0c..3dec0714a3d 100644 --- a/src/server/responses/core-options.ts +++ b/src/server/responses/core-options.ts @@ -16,6 +16,7 @@ import type { ExplicitOpenAiCallerAuth } from "../../providers/openai-sidecar"; import type { CallerDirectAuth } from "../../providers/caller-authorization"; import type { CompactionRoutingOverride } from "./compaction-routing"; import type { TranslatorBudget } from "../../lib/translator-budget"; +import type { NativeReasoningOwner } from "../../responses/reasoning-replay-cache"; import type { TransientSendBudget } from "../../lib/upstream-retry"; import type { RequestLogContext } from "../request-log"; import type { UpstreamHostAdmissionLease } from "../../codex/upstream-host-health"; @@ -134,6 +135,8 @@ export interface HandleResponsesOptions { * it. Omitted means a genuine Responses inbound. */ inboundWire?: InboundWire; + nativeReasoningReplay?: ReadonlyMap; + nativeReasoningMint?: { owner?: NativeReasoningOwner }; /** Droid's per-request effort default; each concrete combo or policy target applies it only if its ladder allows it. */ droidDefaultEffort?: string; /** PF-07: the Chat source a combo child may send natively; set only by the Chat ingress. */ diff --git a/src/server/responses/core-replay.ts b/src/server/responses/core-replay.ts index efb5cd41587..d842deefb51 100644 --- a/src/server/responses/core-replay.ts +++ b/src/server/responses/core-replay.ts @@ -19,6 +19,9 @@ import { reasoningReplayCodexCredentialIdentity, reasoningReplayKeyCredentialIdentity, durableReplayDestinationIdentity, + nativeReasoningTag, + nativeReasoningTagMatches, + type NativeReasoningOwner, bindReasoningReplayScope, reasoningReplayServingIdentityChanged, reasoningReplayItemStoreChanged, @@ -118,6 +121,16 @@ export function bindRouteReasoningReplayScope(args: { ...args, modelId: parsed.modelId, }); + const owner = nativeReasoningOwnerForRoute(args); + if (owner) parsed._nativeReasoningBoundOwner = owner; + else delete parsed._nativeReasoningBoundOwner; + if (parsed._nativeReasoningReplay?.size) { + for (const [blob, tag] of parsed._nativeReasoningReplay) { + if (!nativeReasoningTagMatches(nativeReasoningTag(owner, blob), tag)) { + stripNativeReasoningBlob(parsed, blob); + } + } + } const continuationDestinationIdentity = providerContinuationDestinationIdentity(parsed, provider); const continuationOwner = providerContinuationOwnerFromReplayIdentity( replayIdentity && continuationDestinationIdentity @@ -156,6 +169,66 @@ export function bindRouteReasoningReplayScope(args: { bindProviderContinuationForRoute(parsed, continuationOwner); } +function stripNativeReasoningBlob(parsed: OcxParsedRequest, blob: string): void { + const input = (parsed._rawBody as { input?: unknown } | undefined)?.input; + if (Array.isArray(input)) { + for (const item of input) { + if (item && typeof item === "object" && item.type === "reasoning" && item.encrypted_content === blob) { + delete item.encrypted_content; + delete item.id; + } + } + } + for (const message of parsed.context.messages) { + if (message.role !== "assistant" || !Array.isArray(message.content)) continue; + for (const part of message.content) { + if (part.type !== "thinking" || typeof part.signature !== "string") continue; + try { + const item: unknown = JSON.parse(part.signature); + if (!item || typeof item !== "object" || Array.isArray(item)) continue; + const reasoning = item as Record; + if (reasoning.type !== "reasoning" || reasoning.encrypted_content !== blob) continue; + delete reasoning.encrypted_content; + delete reasoning.id; + part.signature = JSON.stringify(reasoning); + delete part.itemId; + } catch { /* A provider signature is not the parser's JSON carrier. */ } + } + } +} + +export function nativeReasoningOwnerForRoute(args: { + provider: OcxProviderConfig; + oauthCredentialSnapshot?: Pick; + codexAuthContext?: CodexAuthContext; +}): NativeReasoningOwner | undefined { + const { provider } = args; + const salt = thoughtSignatureReplaySalt(); + if (!salt || salt.length < 16) return undefined; + const destination = durableReplayDestinationIdentity(JSON.stringify([ + provider.baseUrl.trim().replace(/\/+$/, ""), provider.responsesPath ?? "", + ])); + let credential: string | undefined; + if (provider.authMode === "oauth") { + const snapshot = args.oauthCredentialSnapshot; + if (snapshot?.accountId && snapshot.generation !== undefined) credential = durableReplayCredentialIdentity( + "oauth", `${snapshot.accountId}\0${snapshot.generation}`, provider.headers, salt, + ); + } else if (provider.authMode === "forward") { + const context = args.codexAuthContext; + if (context && (context.kind === "pool" || context.kind === "main-pool") + && context.accountId && context.writerGeneration !== undefined + && (context.kind !== "pool" || context.generation !== undefined)) { + credential = durableReplayCredentialIdentity("codex", JSON.stringify([ + context.accountId, context.kind === "pool" ? context.generation : null, context.writerGeneration, + ]), provider.headers, salt); + } + } else if (provider.authMode !== "local") { + credential = durableReplayCredentialIdentity("key", nonEmptyProviderApiKey(provider), provider.headers, salt); + } + return destination && credential ? { destination, credential } : undefined; +} + function routeReasoningReplayIdentity(args: { providerName: string; diff --git a/src/server/responses/request-prepare.ts b/src/server/responses/request-prepare.ts index e70095e6b78..94995a88f26 100644 --- a/src/server/responses/request-prepare.ts +++ b/src/server/responses/request-prepare.ts @@ -381,6 +381,10 @@ export async function prepareResponsesRequest( let toolBridgeMaps: ReturnType; try { parsed = parseRequest(body); + if (options.inboundWire === "anthropic") { + parsed._nativeReasoningReplay = options.nativeReasoningReplay; + parsed._nativeReasoningMint = options.nativeReasoningMint; + } parsed._promptCacheKeyIsSharedCohort = options.promptCacheKeyIsSharedCohort; // The body may have been rebuilt since the inbound observation (previous-response // expansion); alias the parsed raw body to the same draft so the outbound diff --git a/src/server/responses/response-effects.ts b/src/server/responses/response-effects.ts index 30606eced40..97ea0b28652 100644 --- a/src/server/responses/response-effects.ts +++ b/src/server/responses/response-effects.ts @@ -75,6 +75,7 @@ export function createResponsesEffects( // message, and leave Codex fataling on a missing compaction item (#422). const commitReasoningReplayServingRoute = (outboundHeaders?: HeadersInit): void => { commitReasoningReplayServingIdentity(parsed._reasoningReplayScope); + if (parsed._nativeReasoningMint) parsed._nativeReasoningMint.owner = parsed._nativeReasoningBoundOwner; rememberServingConversationStateIssuer(admissionState.authCtx, poolAffinityKey); // History has no model namespace. Record the account that actually accepted this // final attempt, after refresh/failover, rather than guessing from mutable affinity. diff --git a/src/types/request.ts b/src/types/request.ts index 6e3edc3a8a7..eca27f0d7db 100644 --- a/src/types/request.ts +++ b/src/types/request.ts @@ -1,4 +1,5 @@ import type { KiroOAuthMetadata } from "../oauth/types"; +import type { NativeReasoningOwner } from "../responses/reasoning-replay-cache"; import type { OcxTool, OcxToolChoice } from "./tools"; import type { TierDecision, TierObservationContext } from "./provider"; @@ -57,6 +58,10 @@ export interface OcxParsedRequest { stream: boolean; options: OcxRequestOptions; _rawBody?: unknown; + /** Claude replay tags, retained only until each route binding verifies them. */ + _nativeReasoningReplay?: ReadonlyMap; + _nativeReasoningBoundOwner?: NativeReasoningOwner; + _nativeReasoningMint?: { owner?: NativeReasoningOwner }; /** * Boundary between replayed history and this turn's newly appended input. Usually the * items the proxy restored from local previous_response_id state; also set when the diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index aa94003a72c..c04b5a86c17 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -544,6 +544,8 @@ Translated tool results in `src/claude/inbound.ts` retain nonempty string `tool_ `src/claude/inbound-content-options.ts` strips Claude Code's leading `x-anthropic-billing-header:` line from a string system prompt or from the first text block of a system array before it becomes Responses `instructions`, dropping a block left empty. The line's `cch` value rotates per request, so keeping it made the translated prefix and the system-derived fallback `prompt_cache_key` change every turn (#6627). The match is anchored at the prompt start, like the Antigravity strip in `src/adapters/google.ts`; native Anthropic passthrough does not use this translation and keeps the client preamble. `tests/claude-integration/claude-inbound.test.ts` covers string and array systems, header-only blocks, later mentions and key stability. +`src/claude/outbound.ts` carries a Responses provider's own `encrypted_content` in an `ocxr1:` `nat` envelope (`src/responses/reasoning-envelope.ts`) only when the serving attempt has committed and `src/server/responses/core-replay.ts` supplied a durable destination and credential owner. The `nat.tag` HMAC binds the exact blob to that owner and the installation's persisted salt. `src/claude/inbound.ts` restores the blob and item id only for the same client-facing model; each route binding verifies the tag against its effective endpoint, including `responsesPath`, and generation-bearing credential, then removes a mismatch from both the raw reasoning item and the parser's thinking-signature carrier before send. Missing owner or tag fails closed, including direct-forward credentials and OAuth generation changes. Native Anthropic ingress projects away every `nat` on both native branches, dropping blob-only blocks and leaving signed or visible-text fields intact. A thinking block that closes before serving commit has no `nat`. Translated bodies that reason request `include: ["reasoning.encrypted_content"]`; a body without reasoning does not. Anthropic `sig` and `red` envelopes keep their existing behavior. `tests/claude-integration/claude-native-reasoning-provenance.test.ts` covers route ownership and native projection. + ## Claude Code picker descriptions `src/claude/model-info.ts` gives every readable (`idStyle: "readable"`, Claude Code CLI) `/v1/models` row a `description` that Claude Code 2.1.257 and later shows under the picker entry instead of the generic "From gateway": `Routed by OpenCodex to native ` for native rows and `Routed by OpenCodex to /` for routed rows. The 1M copy keeps the base description and a Fast sibling appends ` ยท Fast`. Desktop 3P rows keep the ModelInfo shape without a description. `src/claude/gateway-cache.ts` preserves a string `description` when it refreshes and rewrites the gateway-model cache and drops any other type. `tests/claude-integration/claude-model-info.test.ts` and `tests/claude-integration/claude-gateway-cache.test.ts` cover both. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 9520b144d5d..c4ab8415105 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -340,8 +340,8 @@ ChatGPT forward surface, the official OpenAI API, or a provider with the explici than one backend, including OpenAI and xAI, mints native blobs: a destination can decode its own blob without being able to decode the previous backend's. The same serving-identity mismatch signal therefore strips reasoning `encrypted_content` and degrades native compaction blobs through the -existing opaque-note path. When the thread has no recorded identity, the destination-only behavior -is deliberately unchanged. Forward auth alone is not evidence: noncanonical forward providers +existing opaque-note path. Claude-native reasoning adds a per-blob tag check at every route binding: +only the matching durable endpoint and generation-bearing credential retain raw and parsed carriers. Without recorded thread identity, the destination gate remains. Noncanonical forward providers receive no caller credentials and may point at any backend. On any other routed destination the blob also degrades to the same opaque note the bridged parser uses, because forwarding it there fails the turn and the item outlives the failure in the client transcript, repeating on every later turn diff --git a/tests/claude-integration/claude-native-reasoning-continuity.test.ts b/tests/claude-integration/claude-native-reasoning-continuity.test.ts new file mode 100644 index 00000000000..70d38f8dff0 --- /dev/null +++ b/tests/claude-integration/claude-native-reasoning-continuity.test.ts @@ -0,0 +1,186 @@ +import { describe, expect, test } from "bun:test"; +import { sanitizeReasoningInputContent } from "../../src/adapters/openai-responses/reasoning"; +import { anthropicToResponsesBody } from "../../src/claude/inbound"; +import { responsesJsonToAnthropicMessage, responsesSseToAnthropicSse } from "../../src/claude/outbound"; +import { decodeReasoningEnvelope, encodeReasoningEnvelope, OCX_REASONING_PREFIX } from "../../src/responses/reasoning-envelope"; +import { responsesRequestSchema } from "../../src/responses/schema"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; + +const MODEL = "ocx-claude-meta-muse--muse-spark-1.3-contributor"; +const OTHER_MODEL = "ocx-claude-native--gpt-6.1-sol"; +const BLOB = "Q-PaDg-provider-minted-reasoning-blob=="; +const TAG = "a".repeat(64); + +function sse(name: string, data: Record): string { + return `event: ${name}\ndata: ${JSON.stringify(data)}\n\n`; +} + +function streamFrom(text: string): ReadableStream { + const encoder = new TextEncoder(); + return new ReadableStream({ + start(controller) { + for (let i = 0; i < text.length; i += 7) controller.enqueue(encoder.encode(text.slice(i, i + 7))); + controller.close(); + }, + }); +} + +async function collectEvents(stream: ReadableStream): Promise<{ name: string; data: Record }[]> { + const text = await new Response(stream).text(); + const events: { name: string; data: Record }[] = []; + for (const frame of text.split("\n\n")) { + if (!frame.trim()) continue; + let name = ""; + let data = ""; + for (const line of frame.split("\n")) { + if (line.startsWith("event: ")) name = line.slice(7); + else if (line.startsWith("data: ")) data += line.slice(6); + } + events.push({ name, data: JSON.parse(data) }); + } + return events; +} + +async function streamedThinking(frames: string[]): Promise<{ thinking: string; signature: string }[]> { + const events = await collectEvents(responsesSseToAnthropicSse(streamFrom([ + ...frames, + sse("response.completed", { response: { status: "completed", usage: {} } }), + ].join("")), MODEL, { translatorBudget: createTestTranslatorBudget(), nativeReasoningTagFor: () => TAG })); + const blocks: { thinking: string; signature: string }[] = []; + for (const event of events) { + if (event.name === "content_block_start" && event.data.content_block?.type === "thinking") blocks.push({ thinking: "", signature: "" }); + const delta = event.data.delta; + if (event.name !== "content_block_delta" || !delta) continue; + if (delta.type === "thinking_delta") blocks[blocks.length - 1]!.thinking += delta.thinking; + if (delta.type === "signature_delta") blocks[blocks.length - 1]!.signature += delta.signature; + } + return blocks; +} + +function replay(model: string, thinking: string, signature: string, thinkingConfig?: Record) { + return { + model, max_tokens: 1024, + ...(thinkingConfig ? { thinking: thinkingConfig } : {}), + messages: [ + { role: "user", content: "Start the task." }, + { role: "assistant", content: [{ type: "thinking", thinking, signature }, { type: "text", text: "Reading first." }] }, + { role: "user", content: "Continue." }, + ], + }; +} + +describe("Claude route carries a provider's own encrypted reasoning", () => { + test("a streamed blob with no summary still yields a signed thinking block that carries it", async () => { + const blocks = await streamedThinking([ + sse("response.output_item.done", { output_index: 0, item: { type: "reasoning", id: "rs_native", summary: [], encrypted_content: BLOB } }), + ]); + expect(blocks).toHaveLength(1); + expect(blocks[0]!.thinking).toBe(""); + expect(decodeReasoningEnvelope(blocks[0]!.signature)?.nat).toEqual({ enc: BLOB, model: MODEL, id: "rs_native", tag: TAG }); + }); + + test("streamed summary text and the blob share one thinking block", async () => { + const blocks = await streamedThinking([ + sse("response.output_item.added", { output_index: 0, item: { type: "reasoning", id: "rs_native" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_native", output_index: 0, summary_index: 0, delta: "Plan: read, then edit." }), + sse("response.output_item.done", { output_index: 0, item: { type: "reasoning", id: "rs_native", encrypted_content: BLOB } }), + ]); + expect(blocks).toHaveLength(1); + expect(blocks[0]!.thinking).toBe("Plan: read, then edit."); + const envelope = decodeReasoningEnvelope(blocks[0]!.signature); + expect(envelope?.txt).toBe("Plan: read, then edit."); + expect(envelope?.nat).toEqual({ enc: BLOB, model: MODEL, id: "rs_native", tag: TAG }); + }); + + test("reasoning without a native blob keeps the summary-only signature", async () => { + const blocks = await streamedThinking([ + sse("response.output_item.added", { output_index: 0, item: { type: "reasoning", id: "rs_plain" } }), + sse("response.reasoning_summary_text.delta", { item_id: "rs_plain", output_index: 0, summary_index: 0, delta: "hmm" }), + sse("response.output_item.done", { output_index: 0, item: { type: "reasoning", id: "rs_plain" } }), + ]); + expect(decodeReasoningEnvelope(blocks[0]!.signature)).toEqual({ txt: "hmm" }); + }); + + test("the JSON path signs the blob into its thinking block, with or without summary text", () => { + const message = responsesJsonToAnthropicMessage({ + status: "completed", + output: [ + { type: "reasoning", id: "rs_a", summary: [], encrypted_content: BLOB }, + { type: "reasoning", id: "rs_b", summary: [{ type: "summary_text", text: "Next: edit." }], encrypted_content: `${BLOB}b` }, + { type: "message", content: [{ type: "output_text", text: "Done." }] }, + ], + }, MODEL, createTestTranslatorBudget(), () => TAG); + const thinking = (message.content as Record[]).filter(block => block.type === "thinking"); + expect(thinking.map(block => block.thinking)).toEqual(["", "Next: edit."]); + expect(thinking.map(block => decodeReasoningEnvelope(block.signature)?.nat)).toEqual([ + { enc: BLOB, model: MODEL, id: "rs_a", tag: TAG }, + { enc: `${BLOB}b`, model: MODEL, id: "rs_b", tag: TAG }, + ]); + }); + + test("replay to the same model restores the blob and item id, and the sanitizer forwards it", () => { + const signature = encodeReasoningEnvelope({ txt: "Plan.", nat: { enc: BLOB, model: MODEL, id: "rs_native", tag: TAG } }); + const body = anthropicToResponsesBody(replay(MODEL, "Plan.", signature, { type: "adaptive" })); + expect(() => responsesRequestSchema.parse(body)).not.toThrow(); + const reasoning = (body.input as Record[]).filter(item => item.type === "reasoning"); + expect(reasoning).toEqual([{ + type: "reasoning", id: "rs_native", + summary: [{ type: "summary_text", text: "Plan." }], + encrypted_content: BLOB, + }]); + const sanitized = sanitizeReasoningInputContent(body) as { input: Record[] }; + expect(sanitized.input.find(item => item.type === "reasoning")?.encrypted_content).toBe(BLOB); + }); + + test("replay to a different model drops the blob and keeps the visible text", () => { + const signature = encodeReasoningEnvelope({ txt: "Plan.", nat: { enc: BLOB, model: MODEL, id: "rs_native", tag: TAG } }); + const body = anthropicToResponsesBody(replay(OTHER_MODEL, "Plan.", signature)); + const reasoning = (body.input as Record[]).filter(item => item.type === "reasoning"); + expect(reasoning).toHaveLength(1); + expect(reasoning[0]).not.toHaveProperty("encrypted_content"); + expect(reasoning[0]!.summary).toEqual([{ type: "summary_text", text: "Plan." }]); + expect(JSON.stringify(body)).not.toContain(BLOB); + }); + + test("an empty thinking block for a different model leaves no reasoning item", () => { + const signature = encodeReasoningEnvelope({ txt: "", nat: { enc: BLOB, model: MODEL, tag: TAG } }); + const body = anthropicToResponsesBody(replay(OTHER_MODEL, "", signature)); + expect((body.input as Record[]).some(item => item.type === "reasoning")).toBe(false); + }); + + test("a full turn round-trips the blob back to the provider", () => { + const message = responsesJsonToAnthropicMessage({ + status: "completed", + output: [ + { type: "reasoning", id: "rs_turn", summary: [], encrypted_content: BLOB }, + { type: "message", content: [{ type: "output_text", text: "OK" }] }, + ], + }, MODEL, createTestTranslatorBudget(), () => TAG); + const body = anthropicToResponsesBody({ + model: MODEL, max_tokens: 1024, thinking: { type: "adaptive" }, + messages: [ + { role: "user", content: "Remember a number privately." }, + { role: "assistant", content: message.content }, + { role: "user", content: "Which number?" }, + ], + }); + const reasoning = (body.input as Record[]).find(item => item.type === "reasoning"); + expect(reasoning).toMatchObject({ id: "rs_turn", encrypted_content: BLOB }); + }); + + test("requests that reason ask for encrypted reasoning; disabled thinking does not", () => { + const enabled = anthropicToResponsesBody({ model: MODEL, max_tokens: 64, thinking: { type: "adaptive" }, messages: [{ role: "user", content: "hi" }] }); + expect(enabled.include).toEqual(["reasoning.encrypted_content"]); + const effort = anthropicToResponsesBody({ model: MODEL, max_tokens: 64, output_config: { effort: "high" }, messages: [{ role: "user", content: "hi" }] }); + expect(effort.include).toEqual(["reasoning.encrypted_content"]); + const disabled = anthropicToResponsesBody({ model: MODEL, max_tokens: 64, thinking: { type: "disabled" }, messages: [{ role: "user", content: "hi" }] }); + expect(disabled).not.toHaveProperty("include"); + }); + + test("a malformed native field is ignored rather than trusted", () => { + const malformed = OCX_REASONING_PREFIX + Buffer.from(JSON.stringify({ txt: "t", nat: { enc: 7, model: MODEL } })).toString("base64"); + expect(decodeReasoningEnvelope(malformed)).toEqual({ txt: "t" }); + const noModel = OCX_REASONING_PREFIX + Buffer.from(JSON.stringify({ nat: { enc: BLOB } })).toString("base64"); + expect(decodeReasoningEnvelope(noModel)).toBeNull(); + }); +}); diff --git a/tests/claude-integration/claude-native-reasoning-provenance.test.ts b/tests/claude-integration/claude-native-reasoning-provenance.test.ts new file mode 100644 index 00000000000..659624ed2f2 --- /dev/null +++ b/tests/claude-integration/claude-native-reasoning-provenance.test.ts @@ -0,0 +1,306 @@ +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { handleClaudeMessages } from "../../src/server/claude-messages"; +import { nativeAnthropicProjection } from "../../src/claude/inbound"; +import { anthropicToResponsesTranslation } from "../../src/claude/inbound"; +import { responsesSseToAnthropicSse } from "../../src/claude/outbound"; +import { decodeReasoningEnvelope, encodeReasoningEnvelope, OCX_REASONING_PREFIX } from "../../src/responses/reasoning-envelope"; +import { nativeReasoningTag } from "../../src/responses/reasoning-replay-cache"; +import { bindRouteReasoningReplayScope, nativeReasoningOwnerForRoute } from "../../src/server/responses/core-replay"; +import { parseRequest } from "../../src/responses/parser"; +import { clearReasoningReplayCacheForTests } from "../../src/responses/reasoning-replay-cache"; +import type { OcxConfig, OcxProviderConfig } from "../../src/types"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { TranslatorBudgetExceededError } from "../../src/lib/translator-budget"; + +const BLOB = "fixture-native-ciphertext"; +const MODEL = "m"; +const previousFetch = globalThis.fetch; +let previousHome: string | undefined; +let home = ""; +let releaseSpend: (() => void) | undefined; + +function provider(baseUrl = "https://reasoning.example/v1", apiKey = "key-a", responsesPath?: string): OcxProviderConfig { + return { adapter: "openai-responses", authMode: "key", baseUrl, apiKey, models: [MODEL], ...(responsesPath ? { responsesPath } : {}) } as OcxProviderConfig; +} + +function config(selected: OcxProviderConfig = provider()): OcxConfig { + return { defaultProvider: "a", providers: { a: selected } } as OcxConfig; +} + +function turn(signature?: string, model = MODEL, stream = false) { + return { + model, max_tokens: 128, stream, thinking: { type: "adaptive" }, + messages: signature ? [ + { role: "user", content: "First." }, + { role: "assistant", content: [{ type: "thinking", thinking: "Visible plan.", signature }, { type: "text", text: "Done." }] }, + { role: "user", content: "Continue." }, + ] : [{ role: "user", content: "First." }], + }; +} + +function upstreamResponse(blob = BLOB) { + return Response.json({ id: "resp_fixture", object: "response", status: "completed", output: [ + { type: "reasoning", id: "rs_fixture", summary: [{ type: "summary_text", text: "Visible plan." }], encrypted_content: blob }, + { type: "message", role: "assistant", content: [{ type: "output_text", text: "Done." }] }, + ], usage: { input_tokens: 2, output_tokens: 2, total_tokens: 4 } }); +} + +async function send(cfg: OcxConfig, payload: ReturnType, observe: (wire: Record, url: string) => Response = () => upstreamResponse()) { + const wires: Record[] = []; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const wire = JSON.parse(String(init?.body)) as Record; + wires.push(wire); + return observe(wire, String(input)); + }) as typeof fetch; + try { + const response = await handleClaudeMessages(new Request("http://localhost/v1/messages", { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(payload), + }), cfg, { model: "", provider: "" }); + const body = await response.json() as { content?: Array<{ type: string; signature?: string }> }; + expect(response.status).toBe(200); + return { body, wires }; + } finally { globalThis.fetch = previousFetch; } +} + +function signature(body: { content?: Array<{ type: string; signature?: string }> }): string { + return body.content?.find(part => part.type === "thinking")?.signature ?? ""; +} + +beforeAll(() => { + previousHome = process.env.OPENCODEX_HOME; + home = mkdtempSync(join(tmpdir(), "ocx-native-provenance-")); + process.env.OPENCODEX_HOME = home; + releaseSpend = acquireOwnedSpendHome(); +}); +afterAll(() => { + releaseSpend?.(); + globalThis.fetch = previousFetch; + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + removeTreeWithRetry(home); +}); + +describe("Claude native reasoning provenance", () => { + test("same route replays across a simulated restart; destination, credential and responsesPath switches strip before send", async () => { + const first = await send(config(), turn()); + const signed = signature(first.body); + expect(decodeReasoningEnvelope(signed)?.nat?.enc).toBe(BLOB); + clearReasoningReplayCacheForTests(); + const same = await send(config(), turn(signed)); + expect(JSON.stringify(same.wires[0])).toContain(BLOB); + for (const other of [provider("https://other.example/v1"), provider(undefined, "key-b"), provider(undefined, "key-a", "/custom/responses")]) { + const switched = await send(config(other), turn(signed)); + expect(JSON.stringify(switched.wires[0])).not.toContain(BLOB); + expect(JSON.stringify(switched.wires[0])).not.toContain("rs_fixture"); + } + }); + + test("model switch and transplanted, altered or conflicting tags never replay ciphertext", async () => { + const signed = signature((await send(config(), turn())).body); + const nat = decodeReasoningEnvelope(signed)!.nat!; + const switched = await send(config(), turn(signed, "other")); + expect(JSON.stringify(switched.wires[0])).not.toContain(BLOB); + expect(JSON.stringify(switched.wires[0])).toContain("Visible plan."); + for (const value of [ + { ...nat, enc: "other-ciphertext" }, + { ...nat, enc: BLOB + "altered" }, + { ...nat, tag: "0".repeat(64) }, + ]) { + const forged = encodeReasoningEnvelope({ txt: "Visible plan.", nat: value }); + const replay = await send(config(), turn(forged)); + expect(JSON.stringify(replay.wires[0])).not.toContain(value.enc); + } + const duplicate = turn(signed); + duplicate.messages.splice(2, 0, { role: "assistant", content: [{ type: "thinking", thinking: "Visible plan.", signature: encodeReasoningEnvelope({ nat: { ...nat, tag: "0".repeat(64) } }) }, { type: "text", text: "Done." }] } as typeof duplicate.messages[number]); + const conflict = await send(config(), duplicate); + expect(JSON.stringify(conflict.wires[0])).not.toContain(BLOB); + }); + + test("owner requires durable credential and generation and distinguishes route generations", () => { + const key = provider(); + const owner = nativeReasoningOwnerForRoute({ provider: key }); + expect(owner).toBeDefined(); + expect(nativeReasoningTag(owner, BLOB)).toMatch(/^[a-f0-9]{64}$/); + expect(nativeReasoningOwnerForRoute({ provider: { ...key, authMode: "forward" } })).toBeUndefined(); + expect(nativeReasoningOwnerForRoute({ provider: { ...key, apiKey: undefined } })).toBeUndefined(); + const oauth = { ...key, authMode: "oauth" } as OcxProviderConfig; + const one = nativeReasoningOwnerForRoute({ provider: oauth, oauthCredentialSnapshot: { accountId: "slot", generation: 1 } }); + const same = nativeReasoningOwnerForRoute({ provider: oauth, oauthCredentialSnapshot: { accountId: "slot", generation: 1 } }); + const refreshed = nativeReasoningOwnerForRoute({ provider: oauth, oauthCredentialSnapshot: { accountId: "slot", generation: 2 } }); + expect(one).toEqual(same); + expect(one).not.toEqual(refreshed); + expect(nativeReasoningOwnerForRoute({ provider: oauth, oauthCredentialSnapshot: { accountId: "slot" } })).toBeUndefined(); + }); + + test("OAuth generation changes strip both replay carriers while the same generation keeps them", () => { + const oauth = { ...provider(), authMode: "oauth" } as OcxProviderConfig; + const minted = nativeReasoningOwnerForRoute({ provider: oauth, oauthCredentialSnapshot: { accountId: "slot", generation: 1 } })!; + const signed = encodeReasoningEnvelope({ txt: "Visible plan.", nat: { enc: BLOB, model: MODEL, id: "rs_fixture", tag: nativeReasoningTag(minted, BLOB)! } }); + const bind = (generation: number) => { + const translated = anthropicToResponsesTranslation(turn(signed)); + const parsed = parseRequest(translated.body); + parsed._nativeReasoningReplay = translated.nativeReasoningReplay; + bindRouteReasoningReplayScope({ parsed, providerName: "a", provider: oauth, adapterName: "openai-responses", + oauthCredentialSnapshot: { accountId: "slot", generation } }); + return parsed; + }; + const same = bind(1); + expect(JSON.stringify(same._rawBody)).toContain(BLOB); + const refreshed = bind(2); + expect(JSON.stringify(refreshed._rawBody)).not.toContain(BLOB); + expect(JSON.stringify(refreshed.context)).not.toContain(BLOB); + }); + + test("a thinking block closing before commit has no nat; one closing after commit is tagged", async () => { + const owner = nativeReasoningOwnerForRoute({ provider: provider() }); + const frames = `event: response.output_item.done\ndata: ${JSON.stringify({ output_index: 0, item: { type: "reasoning", id: "rs_fixture", summary: [], encrypted_content: BLOB } })}\n\n` + + `event: response.completed\ndata: ${JSON.stringify({ response: { status: "completed", usage: {} } })}\n\n`; + const translate = async (committed: boolean) => { + const stream = new ReadableStream({ start(controller) { controller.enqueue(new TextEncoder().encode(frames)); controller.close(); } }); + const text = await new Response(responsesSseToAnthropicSse(stream, MODEL, { + translatorBudget: createTestTranslatorBudget(), nativeReasoningTagFor: blob => committed ? nativeReasoningTag(owner, blob) : undefined, + })).text(); + const delta = text.split("\n\n").filter(frame => frame.includes('"type":"signature_delta"')).at(0); + const data = delta?.split("data: ")[1]; + return data ? decodeReasoningEnvelope((JSON.parse(data) as { delta: { signature: string } }).delta.signature)?.nat : undefined; + }; + expect(await translate(false)).toBeUndefined(); + expect((await translate(true))?.tag).toBe(nativeReasoningTag(owner, BLOB)); + }); + + test("native Anthropic projection drops blob-only and undecodable blocks, preserves visible fields and input", () => { + const nat = { enc: BLOB, model: MODEL, tag: "a".repeat(64) }; + const visible = encodeReasoningEnvelope({ txt: "Visible plan.", nat }); + const blobOnly = encodeReasoningEnvelope({ nat }); + const escaped = OCX_REASONING_PREFIX + Buffer.from('{"\\u006e\\u0061\\u0074":{"enc":"' + BLOB + '"}}').toString("base64"); + const raw = { messages: [{ role: "assistant", content: [ + { type: "thinking", thinking: "", signature: blobOnly }, + { type: "thinking", thinking: "Visible plan.", signature: visible }, + { type: "thinking", thinking: "", signature: escaped }, + { type: "thinking", thinking: "", signature: OCX_REASONING_PREFIX + "garbage" }, + ] }] }; + const projected = nativeAnthropicProjection(raw, createTestTranslatorBudget()); + expect(JSON.stringify(projected)).not.toContain(BLOB); + expect((projected.messages as typeof raw.messages).at(0)?.content).toHaveLength(1); + expect((projected.messages as typeof raw.messages)[0]!.content[0]!.signature).toBe(encodeReasoningEnvelope({ txt: "Visible plan." })); + expect(raw.messages[0]!.content[0]!.signature).toBe(blobOnly); + }); + + test("native projection reserves decode and re-encode copies against the translator budget", () => { + const nat = { enc: BLOB, model: MODEL, tag: "a".repeat(64) }; + const big = encodeReasoningEnvelope({ txt: "x".repeat(200_000), nat }); + const raw = { messages: [{ role: "assistant", content: [{ type: "thinking", thinking: "x", signature: big }] }] }; + // Eight bytes per code unit of a ~270 KB signature cannot fit a 1 MiB turn budget. + expect(() => nativeAnthropicProjection(raw, createTestTranslatorBudget({ maxTurnBytes: 1024 * 1024 }))).toThrow(TranslatorBudgetExceededError); + const roomy = createTestTranslatorBudget({ maxTurnBytes: 32 * 1024 * 1024 }); + expect(JSON.stringify(nativeAnthropicProjection(raw, roomy))).not.toContain(BLOB); + expect(roomy.snapshot().currentBytes).toBeGreaterThan(0); + // Room for the decode copies but not for decode, the retained signature and the encode copies: + // the encode reservation must reject before stringify/base64 run. + const sigBytes = 8 * big.length; + expect(() => nativeAnthropicProjection(raw, createTestTranslatorBudget({ maxTurnBytes: sigBytes + 1024 }))).toThrow(TranslatorBudgetExceededError); + }); + + test("a direct-forward route cannot mint a native reasoning tag", async () => { + const forward = { ...provider(), authMode: "forward" } as OcxProviderConfig; + const result = await send(config(forward), turn(), () => upstreamResponse()); + expect(decodeReasoningEnvelope(signature(result.body))?.nat).toBeUndefined(); + }); + + test("a request without reasoning does not ask upstream for encrypted content", async () => { + const payload = { model: MODEL, max_tokens: 64, stream: false, messages: [{ role: "user", content: "Hello." }] }; + const sent = await send(config(), payload as ReturnType); + expect(sent.wires[0]).not.toHaveProperty("include"); + }); + + test("a switched openai-chat destination receives no blob through its context signature carrier", async () => { + const signed = signature((await send(config(), turn())).body); + const chat = { ...provider("https://chat.example/v1"), adapter: "openai-chat" } as OcxProviderConfig; + const sent = await send(config(chat), turn(signed), () => new Response( + 'data: {"choices":[{"index":0,"delta":{"role":"assistant","content":"OK"}}]}\n\n' + + 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n' + + 'data: [DONE]\n\n', { headers: { "content-type": "text/event-stream" } }, + )); + expect(JSON.stringify(sent.wires)).not.toContain(BLOB); + }); + + test("streamed passthrough failover strips the failed route's blob and tags the serving route", async () => { + const a = provider("https://route-a.example/v1", "key-a"); + const b = provider("https://route-b.example/v1", "key-b"); + const cfg = { defaultProvider: "a", providers: { a, b }, combos: { pair: { + strategy: "failover", targets: [{ provider: "a", model: MODEL }, { provider: "b", model: MODEL }], + } } } as OcxConfig; + const first = await send(cfg, turn(undefined, "combo/pair")); + const signed = signature(first.body); + expect(decodeReasoningEnvelope(signed)?.nat).toBeDefined(); + const servedBlob = "fixture-serving-route-ciphertext"; + const second = await send(cfg, turn(signed, "combo/pair"), (_wire, url) => { + if (url.includes("route-a.example")) return Response.json({ error: { message: "busy" } }, { status: 429 }); + return new Response( + `event: response.output_item.done\ndata: ${JSON.stringify({ output_index: 0, item: { type: "reasoning", id: "rs_second", summary: [], encrypted_content: servedBlob } })}\n\n` + + `event: response.completed\ndata: ${JSON.stringify({ response: { status: "completed", usage: { input_tokens: 2, output_tokens: 1, total_tokens: 3 } } })}\n\n`, + { headers: { "content-type": "text/event-stream" } }, + ); + }); + expect(second.wires).toHaveLength(2); + expect(JSON.stringify(second.wires[0])).toContain(BLOB); + expect(JSON.stringify(second.wires[1])).not.toContain(BLOB); + const tag = decodeReasoningEnvelope(signature(second.body))?.nat?.tag; + expect(tag).toBe(nativeReasoningTag(nativeReasoningOwnerForRoute({ provider: b }), servedBlob)); + expect(tag).not.toBe(nativeReasoningTag(nativeReasoningOwnerForRoute({ provider: a }), servedBlob)); + }); + + test("both native Anthropic sends strip translated blobs, including escaped and undecodable envelopes", async () => { + const signed = signature((await send(config(), turn())).body); + const nat = decodeReasoningEnvelope(signed)!.nat!; + const blobOnly = encodeReasoningEnvelope({ nat }); + const escaped = OCX_REASONING_PREFIX + Buffer.from('{"\\u006e\\u0061\\u0074":{"enc":"' + BLOB + '"}}').toString("base64"); + const malformed = OCX_REASONING_PREFIX + Buffer.from(JSON.stringify({ nat: { enc: BLOB, model: MODEL } })).toString("base64"); + for (const branch of ["caller", "managed"] as const) { + const seen: Record[] = []; + const transport = (async (_input: RequestInfo | URL, init?: RequestInit) => { + seen.push(JSON.parse(String(init?.body)) as Record); + return Response.json({ id: "msg_fixture", type: "message", role: "assistant", model: "claude-x", + content: [{ type: "text", text: "OK" }], stop_reason: "end_turn", usage: { input_tokens: 2, output_tokens: 1 } }); + }) as typeof fetch; + const nativeProvider = { adapter: "anthropic", authMode: "key", baseUrl: "https://api.anthropic.com", apiKey: "fixture-key", + models: ["claude-x"], fetch: transport } as OcxProviderConfig & { fetch: typeof fetch }; + const cfg = branch === "managed" + ? { defaultProvider: "native", providers: { native: nativeProvider }, protocols: { rollout: { managedMessagesNative: true } } } as OcxConfig + : { defaultProvider: "native", providers: { native: nativeProvider }, claudeCode: { anthropicBaseUrl: "https://api.anthropic.com" } } as OcxConfig; + const raw = { + model: branch === "managed" ? "native/claude-x" : "claude-x", max_tokens: 128, + messages: [ + { role: "user", content: "First." }, + { role: "assistant", content: [ + { type: "thinking", thinking: "", signature: blobOnly }, + { type: "thinking", thinking: "Visible plan.", signature: signed }, + { type: "thinking", thinking: "", signature: escaped }, + { type: "thinking", thinking: "", signature: malformed }, + { type: "thinking", thinking: "", signature: OCX_REASONING_PREFIX + "garbage" }, + ] }, + { role: "user", content: "Continue." }, + ], + }; + globalThis.fetch = transport; + try { + const response = await handleClaudeMessages(new Request("http://localhost/v1/messages", { + method: "POST", headers: { "content-type": "application/json", "x-api-key": "sk-ant-fixture" }, body: JSON.stringify(raw), + }), cfg, { model: "", provider: "" }); + await response.text(); + expect(response.status).toBe(200); + expect(seen).toHaveLength(1); + const wire = seen[0]!; + const assistant = (wire.messages as Array<{ role: string; content: Array<{ signature?: string }> }>).find(message => message.role === "assistant"); + expect(assistant?.content).toHaveLength(1); + expect(assistant?.content[0]?.signature).toBe(encodeReasoningEnvelope({ txt: "Visible plan." })); + expect(JSON.stringify(wire)).not.toContain(blobOnly); + } finally { globalThis.fetch = previousFetch; } + } + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index e7953e2947a..385cfde6b8c 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -271,7 +271,7 @@ "claude-first-party-union.test.ts": "claude-integration", "claude-desktop-system-proxy.test.ts": "claude-integration", "claude-gateway-cache.test.ts": "claude-integration", "claude-inbound-cache-stabilize.test.ts": "claude-integration", "claude-inbound-token-footer.test.ts": "claude-integration", "claude-inbound-debug.test.ts": "claude-integration", "claude-inbound.test.ts": "claude-integration", - "claude-inbound-tool-reference.test.ts": "claude-integration", + "claude-inbound-tool-reference.test.ts": "claude-integration", "claude-native-reasoning-continuity.test.ts": "claude-integration", "claude-native-reasoning-provenance.test.ts": "claude-integration", "claude-output-schema-strict.test.ts": "claude-integration", "claude-intercept-integration.test.ts": "server", "claude-intercept-local-ca.test.ts": "claude-integration", "claude-intercept-local-ca-files.test.ts": "claude-integration",