From e7409f64f22def7e67c5f19b39bf12a1a0288858 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:54:26 +0900 Subject: [PATCH 01/20] refactor(anthropic): share the Messages URL, version and key-auth headers The adapter's endpoint, pinned anthropic-version and key placement move into exported helpers so a second Messages sender can reuse them instead of restating them. --- src/adapters/anthropic.ts | 49 ++++++++++++++++++++++++++++----------- 1 file changed, 36 insertions(+), 13 deletions(-) diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index 8f17a2710b1..95839988c75 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -523,6 +523,39 @@ function anthropicKeyUsesBearer(provider: OcxProviderConfig): boolean { return provider.apiKeyTransport === "bearer"; } +/** The `anthropic-version` every Messages request from this proxy pins. */ +export const ANTHROPIC_API_VERSION = "2023-06-01"; + +/** + * The fixed headers of every Messages request this proxy builds, before credentials. Shared by + * the adapter and the managed native lane so both pin the same version and client identity. + */ +export function anthropicBaseRequestHeaders(stream: boolean | undefined): Record { + return { + "Content-Type": "application/json", + "anthropic-version": ANTHROPIC_API_VERSION, + "Accept": stream ? "text/event-stream" : "application/json", + "User-Agent": "@anthropic-ai/sdk/0.74.0", + }; +} + +/** Key-auth credential placement: `x-api-key`, or a bearer when the provider asks for one. */ +export function applyAnthropicKeyAuth(headers: Record, provider: OcxProviderConfig): void { + if (typeof provider.apiKey !== "string") return; + if (anthropicKeyUsesBearer(provider)) headers["Authorization"] = `Bearer ${provider.apiKey}`; + else headers["x-api-key"] = provider.apiKey; +} + +/** The provider's Messages endpoint, refusing a base URL with an unresolved `{placeholder}`. */ +export function resolveAnthropicMessagesUrl(provider: Pick): string { + const url = anthropicMessagesUrl(provider.baseUrl); + const unresolvedPlaceholder = url.match(/\{[^}]*\}/)?.[0]; + if (unresolvedPlaceholder) { + throw new Error(`anthropic baseUrl contains unresolved ${unresolvedPlaceholder}`); + } + return url; +} + /** Map a Responses reasoning effort to an Anthropic extended-thinking budget (tokens, >= 1024). */ function reasoningBudget(effort: string): number { switch (effort) { @@ -1129,22 +1162,13 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti body.tool_choice = { ...settledToolChoice, disable_parallel_tool_use: true }; } - const url = anthropicMessagesUrl(provider.baseUrl); - const unresolvedPlaceholder = url.match(/\{[^}]*\}/)?.[0]; - if (unresolvedPlaceholder) { - throw new Error(`anthropic baseUrl contains unresolved ${unresolvedPlaceholder}`); - } + const url = resolveAnthropicMessagesUrl(provider); // Anthropic fast mode: `speed` is only accepted beside its beta; without it the API // answers 400 "speed: Extra inputs are not permitted". The beta is merged below, after // any header override, so a request never carries one without the other. const fastSpeed = anthropicFastSpeed(parsed, provider); if (fastSpeed) body.speed = fastSpeed.value; - const headers: Record = { - "Content-Type": "application/json", - "anthropic-version": "2023-06-01", - "Accept": parsed.stream ? "text/event-stream" : "application/json", - "User-Agent": "@anthropic-ai/sdk/0.74.0", - }; + const headers = anthropicBaseRequestHeaders(parsed.stream); if (isOAuth) { headers["Authorization"] = `Bearer ${provider.apiKey}`; headers["anthropic-beta"] = ANTHROPIC_OAUTH_BETA; @@ -1155,8 +1179,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti headers["X-Claude-Code-Session-Id"] = claudeCodeSessionId(provider.apiKey); headers["x-client-request-id"] = crypto.randomUUID(); } else { - if (anthropicKeyUsesBearer(provider)) headers["Authorization"] = `Bearer ${provider.apiKey}`; - else headers["x-api-key"] = provider.apiKey; + applyAnthropicKeyAuth(headers, provider); } if (provider.headers) Object.assign(headers, provider.headers); mergeAnthropicBetaHeader(headers, fastSpeed?.betas ?? []); From c321dfe078d6111945bf8727ab89af0b5aad37d5 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:54:26 +0900 Subject: [PATCH 02/20] feat(anthropic): build a managed native Messages request from the source body Allowlisted source fields, the wire model and the provider's own key; no caller header is read, so the managed lane cannot inherit caller-forward authority. --- src/adapters/anthropic/passthrough.ts | 87 +++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) create mode 100644 src/adapters/anthropic/passthrough.ts diff --git a/src/adapters/anthropic/passthrough.ts b/src/adapters/anthropic/passthrough.ts new file mode 100644 index 00000000000..9c10de68001 --- /dev/null +++ b/src/adapters/anthropic/passthrough.ts @@ -0,0 +1,87 @@ +/** + * Managed native Messages request builder (PF-08). + * + * The request a proxy-managed Anthropic key sends when the client already spoke Messages: the + * caller's own body, cut to a fixed field allowlist, with the wire model and the provider's + * credential. URL, `anthropic-version`, client identity and credential placement come from the + * same helpers the Anthropic adapter uses, so the two lanes cannot drift apart. + * + * Authority: only the provider's configured key is ever placed on the request. No caller header + * is read here at all — the caller-forward passthrough in `src/server/claude-messages.ts` is the + * only place a caller's Anthropic credential may travel, and it does not come through here. + */ +import type { OcxConfig, OcxProviderConfig } from "../../types"; +import { anthropicBaseRequestHeaders, applyAnthropicKeyAuth, resolveAnthropicMessagesUrl } from "../anthropic"; + +/** + * Top-level Messages fields the native lane forwards. Everything else is dropped: an unknown or + * beta-gated field would otherwise reach the provider unchecked. None of the dropped fields has + * a name in the protocol feature vocabulary, so no feature effect is recorded for them. + */ +export const ANTHROPIC_MESSAGES_PASSTHROUGH_FIELDS = [ + "model", + "messages", + "system", + "max_tokens", + "metadata", + "stop_sequences", + "stream", + "temperature", + "top_p", + "top_k", + "tools", + "tool_choice", + "thinking", + "output_config", + "service_tier", +] as const; + +const PASSTHROUGH_FIELD_SET: ReadonlySet = new Set(ANTHROPIC_MESSAGES_PASSTHROUGH_FIELDS); + +export interface AnthropicMessagesPassthroughRequest { + url: string; + headers: Record; + /** The serialized wire body. */ + body: string; + /** The same body before serialization, for callers that count or inspect what is sent. */ + wireBody: Record; +} + +/** The allowlisted copy of `body` with `model` set to the wire model. Shallow: nothing is cloned. */ +export function anthropicMessagesPassthroughBody( + body: Readonly>, + modelId: string, +): Record { + const out: Record = {}; + for (const [key, value] of Object.entries(body)) { + if (PASSTHROUGH_FIELD_SET.has(key) && value !== undefined) out[key] = value; + } + out.model = modelId; + return out; +} + +/** + * Build the upstream request. Throws the adapter's own errors for a missing key or a malformed + * or unresolved base URL. `config` is accepted for parity with the other passthrough builders; + * no config key changes the wire today. + */ +export function buildAnthropicMessagesPassthroughRequest( + provider: OcxProviderConfig, + modelId: string, + body: Readonly>, + _config?: OcxConfig, +): AnthropicMessagesPassthroughRequest { + if (provider.authMode !== undefined && provider.authMode !== "key") { + throw new Error("managed native Messages requires a key-auth anthropic provider"); + } + if (typeof provider.apiKey !== "string" || provider.apiKey.trim() === "") { + throw new Error("anthropic provider requires a non-empty apiKey (authMode: key)"); + } + const url = resolveAnthropicMessagesUrl(provider); + const wireBody = anthropicMessagesPassthroughBody(body, modelId); + const headers = anthropicBaseRequestHeaders(wireBody.stream === true); + applyAnthropicKeyAuth(headers, provider); + // Operator-configured provider headers apply exactly as the adapter applies them. + if (provider.headers) Object.assign(headers, provider.headers); + return { url, headers, body: JSON.stringify(wireBody), wireBody }; +} From 88f21f3a45b11a7d1b58c54bae7aeec60ab56fb0 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:54:32 +0900 Subject: [PATCH 03/20] feat(claude): name why a Messages route declines the managed native lane One pure rule for the ingress, count_tokens and the planner: switch, anthropic adapter, key auth, no combo or policy, no synthetic effort or fast row, no vision preprocessing. --- src/server/messages-native-eligibility.ts | 71 +++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 src/server/messages-native-eligibility.ts diff --git a/src/server/messages-native-eligibility.ts b/src/server/messages-native-eligibility.ts new file mode 100644 index 00000000000..1b1742b30a4 --- /dev/null +++ b/src/server/messages-native-eligibility.ts @@ -0,0 +1,71 @@ +/** + * Whether a settled Messages route may take the managed native Messages lane (PF-08). + * + * PURE: reads the route, the body's structure and config, and nothing else. The Messages ingress, + * `count_tokens` and the protocol planner (`src/protocols/plan-snapshot.ts`) all ask here, so a + * preview, a count and the real send cannot disagree about the lane. + */ +import type { ProtocolReasonCode } from "../protocols/contract"; +import { featuresFromMessagesBody } from "../protocols/features"; +import { resolveProtocolSettings } from "../protocols/settings"; +import type { RouteResult } from "../router"; +import type { OcxConfig } from "../types"; +import { requiresVisionPreprocessing } from "../vision"; + +/** Why the managed native Messages lane declines a route, as a protocol reason code. */ +export type NativeMessagesDeclineReason = Extract< + ProtocolReasonCode, + | "rollout-disabled" + | "cross-wire-ir" + | "auth-mode-not-native" + | "combo-or-policy-route" + | "effort-row" + | "fast-row" + | "vision-preprocessing" +>; + +/** Model-selector facts the body cannot carry: a synthetic effort or fast row was requested. */ +export interface NativeMessagesSelector { + effortRow?: boolean; + fastRow?: boolean; +} + +/** + * The first rule that keeps a Messages request off the managed native lane, or `undefined` when + * the route is eligible. + * + * - the `protocols.rollout.managedMessagesNative` switch is off; + * - the final adapter is not `anthropic`; + * - the credential is not a proxy-managed key (OAuth is PF-10; `forward` belongs to the caller); + * - a combo or policy route owns multi-candidate execution in the Responses pipeline; + * - a synthetic effort or fast row needs the adapter that owns its wire rewrite; + * - an image would reach a model the operator declared unable to read it. + */ +export function nativeMessagesDeclineReason( + route: RouteResult, + body: Readonly>, + config: OcxConfig, + selector: NativeMessagesSelector = {}, +): NativeMessagesDeclineReason | undefined { + if (!resolveProtocolSettings(config).rollout.managedMessagesNative) return "rollout-disabled"; + const provider = route.provider; + if (provider.adapter !== "anthropic") return "cross-wire-ir"; + if (provider.authMode !== undefined && provider.authMode !== "key") return "auth-mode-not-native"; + if (route.combo || route.routeKind === "combo" || route.routeKind === "policy") return "combo-or-policy-route"; + if (selector.effortRow) return "effort-row"; + if (selector.fastRow) return "fast-row"; + if (featuresFromMessagesBody(body).has("request.images") + && requiresVisionPreprocessing(config, provider, route.modelId, route.providerName)) { + return "vision-preprocessing"; + } + return undefined; +} + +export function isNativeMessagesRouteEligible( + route: RouteResult, + body: Readonly>, + config: OcxConfig, + selector?: NativeMessagesSelector, +): boolean { + return nativeMessagesDeclineReason(route, body, config, selector) === undefined; +} From 20cddfe122daf8ee68dc5ecf9c4420f815685c74 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:54:32 +0900 Subject: [PATCH 04/20] feat(claude): send eligible managed-key Messages natively Modelled on native Chat: shared attempt, final log, spend tracker, proactive key pick, 401/429 key failover and replay; SSE relays through the existing Anthropic log tap. --- src/server/claude-messages.ts | 4 +- src/server/messages-native.ts | 618 ++++++++++++++++++++++++++++++++++ 2 files changed, 620 insertions(+), 2 deletions(-) create mode 100644 src/server/messages-native.ts diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index 996f67845ed..64924b7ffbb 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -254,7 +254,7 @@ function uuidFromHex(hex32: string): string { return `${h.slice(0, 8)}-${h.slice(8, 12)}-4${h.slice(13, 16)}-8${h.slice(17, 20)}-${h.slice(20, 32)}`; } -function anthropicUsageToOcx(usage: Rec | undefined): { inputTokens: number; outputTokens: number; cachedInputTokens?: number; cacheReadInputTokens?: number; cacheCreationInputTokens?: number } | undefined { +export function anthropicUsageToOcx(usage: Rec | undefined): { inputTokens: number; outputTokens: number; cachedInputTokens?: number; cacheReadInputTokens?: number; cacheCreationInputTokens?: number } | undefined { if (!usage) return undefined; const num = (v: unknown) => typeof v === "number" ? v : 0; const hasCache = usage.cache_read_input_tokens !== undefined || usage.cache_creation_input_tokens !== undefined; @@ -448,7 +448,7 @@ export function tapAnthropicSseForLog( * An empty id has no representable wire form, and forwarding `""` is what Anthropic * rejects (#1767), so the request fails locally with a 400 before any upstream fetch. */ -function sanitizePassthroughToolCallIds(messages: unknown[]): void { +export function sanitizePassthroughToolCallIds(messages: unknown[]): void { const blocks: Rec[] = []; for (const message of messages) { if (!isRec(message) || !Array.isArray(message.content)) continue; diff --git a/src/server/messages-native.ts b/src/server/messages-native.ts new file mode 100644 index 00000000000..3fb7a499b3f --- /dev/null +++ b/src/server/messages-native.ts @@ -0,0 +1,618 @@ +/** + * Managed native Messages lane (PF-08, behind `protocols.rollout.managedMessagesNative`). + * + * A Messages request whose settled route is a proxy-managed key on the `anthropic` adapter is + * sent as Messages: the source body (after the ingress's managed-client steps) cut to a field + * allowlist, the wire model, and the provider's own key. Modelled on the native Chat lane and + * built on the same shared pieces — attempt row, finish-once final log, spend tracker, proactive + * key selection, key failover and 429 replay, connection policy — so nothing the Responses + * pipeline enforces is bypassed. + * + * Authority. This lane never reads a caller header. The caller-forward passthrough in + * `claude-messages.ts` (the caller's own Anthropic credential) is a different branch decided + * before this one, and nothing here can reach it or be reached from it. + * + * Loaded lazily by `claude-messages.ts` only when the switch is on and a route is eligible. + */ +import { enforceAnthropicImageLimits } from "../adapters/anthropic-image-guard"; +import { normalizeAnthropicImages } from "../adapters/anthropic-image-normalize"; +import { formatAnthropicErrorBody } from "../adapters/anthropic"; +import { + buildAnthropicMessagesPassthroughRequest, + type AnthropicMessagesPassthroughRequest, +} from "../adapters/anthropic/passthrough"; +import { resolveInboundModel } from "../claude/inbound"; +import { anthropicErrorBody, anthropicErrorResponse, collectAnthropicMessage } from "../claude/outbound"; +import type { AdmissionLease } from "../lib/admission"; +import { readBoundedResponseBody } from "../lib/bounded-body"; +import { classifyError } from "../lib/errors"; +import { redactSecretString } from "../lib/redact"; +import { resolveClientRetryAfter } from "../lib/retry-after"; +import { isTranslatorBudgetExceededError, type TranslatorBudget } from "../lib/translator-budget"; +import { + applyReplayRefusalClientHeaders, + applyUpstreamRecoveryInit, + fetchWithResetRetry, + fetchWithTransientRetry, + isNonReplayableResponse, + isReplayRefusalResponse, + isTransientUpstreamStatus, + prepareSameTarget429Wait, + REPLAY_REFUSED_STATUS, + retainReplayRefusal, + UpstreamRetryEvidenceError, + type UpstreamSendRecovery, + UPSTREAM_RESET_REPLAY_REFUSED_CODE, +} from "../lib/upstream-retry"; +import { providerApiKeySelectionIsCurrent, resolveCurrentProviderApiKeyTransport } from "../providers/api-key-selection"; +import { + hasKeyPoolFailover, + rateLimitRetryDelayMs, + rateLimitRetryPolicyFor, + rotateProviderTransportOn401, + rotateProviderTransportOn429, + selectProactiveApiKeyTransport, + transientRetryPolicyFor, +} from "../providers/key-failover"; +import { stampApiKeyAccountLabel } from "../providers/label"; +import type { OcxProviderTransport } from "../providers/xai-transport"; +import { preservesPhysicalComboProvider, resolveComboId } from "../combos"; +import { captureRouteStaticPolicy, routeModel, type RouteResult } from "../router"; +import { POLICY_NAMESPACE, resolvePolicyProfileId } from "../routing/profile"; +import type { OcxConfig, OcxProviderConfig, OcxUsage } from "../types"; +import { resolveWireProtocolOverride } from "./adapter-resolve"; +import { + anthropicUsageToOcx, + estimateClaudeRequestTokens, + resolvePassthroughBodyGuard, + sanitizePassthroughToolCallIds, + tapAnthropicSseForLog, +} from "./claude-messages"; +import { beginInferenceAttempt } from "./inference/attempt"; +import { createFinalRequestLog, type FinalRequestLogMeta } from "./inference/final-log"; +import { registerTurn, unregisterTurn } from "./lifecycle"; +import { nativeMessagesDeclineReason, type NativeMessagesSelector } from "./messages-native-eligibility"; +import { + noteProviderAttemptSend, + recordAttemptCredentialSource, + recordFirstOutput, + recordKeyAttemptFailure, + recordKeyWireAttemptUsage, + type RequestLogContext, +} from "./request-log"; +import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, sendWithConnectionPolicy } from "./responses/fetch-helpers"; +import { linkAbortSignal } from "./responses/core-lifetime"; +import { attachRequestSpendTracker } from "./responses/request-spend"; +import { workflowRefusalResponse } from "./workflow-refusal"; + +export { + isNativeMessagesRouteEligible, + nativeMessagesDeclineReason, + type NativeMessagesDeclineReason, +} from "./messages-native-eligibility"; + +type Rec = Record; + +const MAX_NATIVE_MESSAGES_JSON_BYTES = 32 * 1024 * 1024; +const MAX_NATIVE_MESSAGES_ERROR_BYTES = 64 * 1024; + +class NativeMessagesSpendRefusal extends Error {} + +function isRec(value: unknown): value is Rec { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +export interface HandleNativeMessagesOptions { + req: Request; + config: OcxConfig; + logCtx: RequestLogContext; + logIds?: { requestId: string; start: number; turnAdmissionLease?: AdmissionLease }; + /** The settled, eligible route (see `isNativeMessagesRouteEligible`). */ + route: RouteResult; + /** + * The Messages body after the ingress's managed-client steps. Owned by this lane from here + * on: a fresh envelope copy when an envelope exists, otherwise the ingress's own body. + */ + body: Rec; + /** The selector the client sent, echoed on the request log. */ + requestedModel: string; + translatorBudget: TranslatorBudget; +} + +type FinishLog = (status: number, message?: string, closeReason?: FinalRequestLogMeta["closeReason"]) => void; + +/** Relay an upstream body unchanged, recording first output on its first non-empty chunk. */ +function observeFirstChunk(body: ReadableStream, onFirst: () => void): ReadableStream { + const reader = body.getReader(); + let seen = false; + return new ReadableStream({ + async pull(controller) { + const { done, value } = await reader.read(); + if (done) { + controller.close(); + return; + } + if (!seen && value.byteLength > 0) { + seen = true; + onFirst(); + } + controller.enqueue(value); + }, + cancel(reason) { + return reader.cancel(reason); + }, + }); +} + +/** A minimal valid Messages stream for a streaming caller whose upstream answered JSON. */ +function messageAsSse(message: Rec): string { + const frames: string[] = []; + const emit = (name: string, data: Rec) => frames.push(`event: ${name}\ndata: ${JSON.stringify(data)}\n\n`); + emit("message_start", { type: "message_start", message: { ...message, content: [], stop_reason: null } }); + const blocks = Array.isArray(message.content) ? message.content.filter(isRec) : []; + blocks.forEach((block, index) => { + emit("content_block_start", { type: "content_block_start", index, content_block: block }); + emit("content_block_stop", { type: "content_block_stop", index }); + }); + emit("message_delta", { + type: "message_delta", + delta: { stop_reason: message.stop_reason ?? "end_turn", stop_sequence: message.stop_sequence ?? null }, + usage: message.usage ?? {}, + }); + emit("message_stop", { type: "message_stop" }); + return frames.join(""); +} + +/** Reject a body the Messages API cannot carry before anything is sent. */ +async function prepareNativeBody(body: Rec, signal: AbortSignal): Promise { + if (!Array.isArray(body.messages)) return; + // The adapter normally owns these; this lane bypasses it, so it runs the same steps as the + // caller-forward passthrough: tier-normalize and guard images, then repair tool-call ids. + await normalizeAnthropicImages(body.messages, { abortSignal: signal }); + enforceAnthropicImageLimits(body.messages); + sanitizePassthroughToolCallIds(body.messages); +} + +/** + * Handle one eligible Messages request natively. Opens the attempt, owns the request's final + * log row, and answers in the Messages wire (SSE or JSON per the caller's own `stream`). + */ +export async function handleNativeMessages(options: HandleNativeMessagesOptions): Promise { + const { req, config, logCtx, logIds, route, body, requestedModel, translatorBudget } = options; + const requestedStream = body.stream === true; + logCtx.inboundProtocol = "messages"; + logCtx.model = route.modelId; + logCtx.provider = route.providerName; + logCtx.providerAdapter = route.provider.adapter; + logCtx.requestedModel = requestedModel; + if (route.routeReason === "model-alias" || route.modelId !== requestedModel) logCtx.requestedAlias = requestedModel; + logCtx.requestedServiceTier = typeof body.service_tier === "string" ? body.service_tier : undefined; + // Reserve spend the way native Chat does: an input estimate that never enters usage, and the + // caller's own output ceiling. + if (logCtx.usageLogInputTokens === undefined) { + logCtx.spendInputEstimateTokens = estimateClaudeRequestTokens(body, requestedModel); + } + if (typeof body.max_tokens === "number" && body.max_tokens > 0) { + logCtx.spendOutputCeilingTokens = Math.trunc(body.max_tokens); + } + + const attemptHandle = beginInferenceAttempt(logCtx, { + provider: route.providerName, + model: route.modelId, + adapter: "anthropic", + }); + attemptHandle.seal(logCtx.accountLogLabel); + const { attempt } = attemptHandle; + const finalLog = createFinalRequestLog(logIds, logCtx); + const finishLog: FinishLog = (status, message, closeReason = "non_stream") => { + if (finalLog.finished()) return; + if (message) logCtx.upstreamError = redactSecretString(message).slice(0, 500); + finalLog.finish(status, { closeReason }); + }; + const bindUsage = (usage: OcxUsage | undefined) => { + if (!usage) return; + if (!recordKeyWireAttemptUsage(logCtx, usage)) { + logCtx.usage = usage; + attempt.usage = usage; + } + }; + const fail = (status: number, message: string, type?: string, code?: string): Response => { + const safeMessage = redactSecretString(message); + finishLog(status, safeMessage); + return anthropicErrorResponse(status, safeMessage, type, code); + }; + + try { + await prepareNativeBody(body, req.signal); + } catch (error) { + if (req.signal.aborted) return fail(499, "Client cancelled request", "api_error"); + if (isTranslatorBudgetExceededError(error)) { + return fail(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); + } + // An AnthropicRequestError (empty tool id) or an image the guard cannot carry. + return fail(400, error instanceof Error ? error.message : String(error), "invalid_request_error"); + } + + const upstream = new AbortController(); + const cleanupAbort = linkAbortSignal(upstream, req.signal); + let streamTurnRegistered = false; + const transferTurnToStream = () => { + const lease = logIds?.turnAdmissionLease; + if (!lease || typeof (lease as { bindAbortController?: unknown }).bindAbortController !== "function") return; + registerTurn(upstream, lease); + streamTurnRegistered = true; + }; + const releaseStreamTurn = () => { + if (!streamTurnRegistered) return; + streamTurnRegistered = false; + unregisterTurn(upstream); + }; + const connectMs = config.connectTimeoutMs ?? 200_000; + // Same pre-dispatch key preference every direct send path applies (see chat-native.ts). + const proactiveKeyProvider = selectProactiveApiKeyTransport(config, route.providerName, route.provider); + if (proactiveKeyProvider) route.provider = proactiveKeyProvider; + let activeProvider: OcxProviderConfig = route.provider; + stampApiKeyAccountLabel(logCtx, route.providerName, activeProvider); + const spendTracker = attachRequestSpendTracker(req, logCtx); + let activeRequest: AnthropicMessagesPassthroughRequest; + let retainedRequestBytes = 0; + const releaseRetainedRequest = () => { + if (retainedRequestBytes === 0) return; + translatorBudget.releaseRetained(retainedRequestBytes, { kind: "request_copies" }); + retainedRequestBytes = 0; + }; + const retainRequest = (request: AnthropicMessagesPassthroughRequest) => { + const bytes = Buffer.byteLength(request.body); + translatorBudget.chargeRetained(bytes, { kind: "request_copies" }); + retainedRequestBytes = bytes; + }; + // Every rebuild reads the same `body`; the builder copies, so no credential or header from an + // earlier key survives into the next request. + const buildActiveRequest = () => { + recordAttemptCredentialSource(attempt, route.providerName, activeProvider, "anthropic"); + return buildAnthropicMessagesPassthroughRequest(activeProvider, route.modelId, body, config); + }; + const rebuildFor = (provider: OcxProviderConfig) => { + activeProvider = provider; + stampApiKeyAccountLabel(logCtx, route.providerName, activeProvider); + releaseRetainedRequest(); + activeRequest = buildActiveRequest(); + retainRequest(activeRequest); + }; + try { + activeRequest = buildActiveRequest(); + retainRequest(activeRequest); + } catch (error) { + releaseRetainedRequest(); + cleanupAbort(); + upstream.abort(); + if (isTranslatorBudgetExceededError(error)) { + return fail(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); + } + return fail(400, error instanceof Error ? error.message : String(error), "invalid_request_error"); + } + + // One inbound request owns one transient send allowance, captured before any key rotation. + const requestTransientPolicy = transientRetryPolicyFor(activeProvider); + let transientSendsUsed = 0; + const remainingTransientSends = (): number => requestTransientPolicy + ? Math.max(0, requestTransientPolicy.attempts - transientSendsUsed) + : Number.POSITIVE_INFINITY; + const transientSendAvailable = (): boolean => remainingTransientSends() > 0; + const selector: NativeMessagesSelector = {}; + + const send = async (recovery?: "rate-limit-429" | "key-429" | "key-401"): Promise => { + const remaining = remainingTransientSends(); + if (requestTransientPolicy && remaining <= 0) { + throw new Error("native Messages transient send budget exhausted before recovery dispatch"); + } + const fetchWithPolicy = requestTransientPolicy ? fetchWithTransientRetry : fetchWithResetRetry; + const request = activeRequest; + return await fetchWithPolicy( + (transportRecovery?: UpstreamSendRecovery) => fetchWithHeaderTimeout( + request.url, + applyUpstreamRecoveryInit({ method: "POST", headers: request.headers, body: request.body }, transportRecovery), + upstream.signal, + connectMs, + requestedStream, + providerFetch(activeProvider, undefined, { + providerName: route.providerName, + modelId: route.modelId, + dispatchOverride: async (_input, init, execute) => { + if (!providerApiKeySelectionIsCurrent(config, route.providerName, activeProvider)) { + const current = resolveCurrentProviderApiKeyTransport(config, route.providerName, activeProvider); + if (!current || nativeMessagesDeclineReason({ ...route, provider: current }, body, config, selector) !== undefined) { + throw new Error("Provider key selection is no longer available for native Messages"); + } + rebuildFor(current); + } + // The retry closure may hold a pre-reselection request: send the current one whole. + const wire = activeRequest; + const headers = new Headers(wire.headers); + const encoding = new Headers(init.headers).get("accept-encoding"); + if (!headers.has("accept-encoding") && encoding) headers.set("accept-encoding", encoding); + if (init.signal?.aborted) throw init.signal.reason; + if (!spendTracker.charge()) throw new NativeMessagesSpendRefusal(); + noteProviderAttemptSend(logCtx, route.providerName, activeProvider, logCtx.usageLogInputTokens, transportRecovery ?? recovery); + const dispatched = await sendWithConnectionPolicy( + (activeProvider as OcxProviderTransport).fetch ?? execute, + wire.url, + applyUpstreamRecoveryInit({ ...init, method: "POST", headers, body: wire.body }, transportRecovery), + { providerName: route.providerName, provider: activeProvider }, + ); + if (!dispatched.ok) await recordKeyAttemptFailure(logCtx, dispatched, init.signal ?? upstream.signal); + return dispatched; + }, + }), + ), + { + abortSignal: upstream.signal, + label: safeHostLabel(request.url), + ...(requestTransientPolicy + ? { + attempts: remaining, + onSendsConsumed: (sends: number) => { transientSendsUsed += Math.max(0, sends); }, + } + : {}), + }, + ); + }; + const discard = (response: Response) => { + try { void response.body?.cancel().catch(() => {}); } catch { /* already closed */ } + }; + + let response: Response; + try { + response = await send(); + // A credential-scoped 401 on a static key pool says nothing about the sibling keys. + while (response.status === 401 && hasKeyPoolFailover(activeProvider) && transientSendAvailable()) { + const rotated = rotateProviderTransportOn401(config, route.providerName, activeProvider, { + now: Date.now(), + attemptedKey: activeProvider.apiKey, + }); + if (!rotated) break; + discard(response); + rebuildFor(rotated); + response = await send("key-401"); + } + const retryPolicy = rateLimitRetryPolicyFor(activeProvider); + let retries = 0; + // A refusal this proxy synthesized for a reset replay is not a provider rate limit. + while ( + response.status === 429 + && !isNonReplayableResponse(response) + && retryPolicy + && retries < retryPolicy.attempts + && transientSendAvailable() + ) { + retries += 1; + for await (const _ of prepareSameTarget429Wait({ + body: response.body, + signal: upstream.signal, + delayMs: rateLimitRetryDelayMs(retryPolicy, response.headers.get("retry-after"), Date.now()), + })) { /* pre-stream wait */ } + if (upstream.signal.aborted) throw upstream.signal.reason; + response = await send("rate-limit-429"); + } + while (response.status === 429 && !isNonReplayableResponse(response) && hasKeyPoolFailover(activeProvider)) { + const rotated = rotateProviderTransportOn429(config, route.providerName, activeProvider, { + retryAfter: response.headers.get("retry-after"), + now: Date.now(), + attemptedKey: activeProvider.apiKey, + attemptedSelection: activeProvider._apiKeyAttempt, + }); + if (!rotated) break; + // Rotation already recorded the cooldown; keep the terminal 429 when no send remains. + if (!transientSendAvailable()) break; + discard(response); + rebuildFor(rotated); + response = await send("key-429"); + } + } catch (error) { + releaseRetainedRequest(); + cleanupAbort(); + upstream.abort(); + if (req.signal.aborted) return fail(499, "Client cancelled request", "api_error"); + const sendError = error instanceof UpstreamRetryEvidenceError ? error.cause : error; + if (sendError instanceof NativeMessagesSpendRefusal) { + const refusal = workflowRefusalResponse("workflow-spend-exhausted", logCtx); + finishLog(429); + return refusal; + } + if (isTranslatorBudgetExceededError(error)) { + return fail(413, "request translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); + } + return fail(502, error instanceof Error ? error.message : String(error), "api_error"); + } + releaseRetainedRequest(); + + if (!response.ok) { + let bodyText = ""; + try { + const read = await readBoundedResponseBody(response, { signal: upstream.signal, maxBytes: MAX_NATIVE_MESSAGES_ERROR_BYTES }); + if (read.displaySafe) bodyText = read.text; + } catch { /* status-only fallback */ } + cleanupAbort(); + if (req.signal.aborted) { + upstream.abort(); + return fail(499, "Client cancelled request", "api_error"); + } + return nativeMessagesErrorResponse(response, bodyText, finishLog); + } + + const contentType = response.headers.get("content-type")?.toLowerCase() ?? ""; + if (contentType.includes("text/event-stream") && response.body) { + const bodyGuard = resolvePassthroughBodyGuard(config, req.signal); + const source = logIds ? observeFirstChunk(response.body, () => recordFirstOutput(logCtx, logIds.start)) : response.body; + if (requestedStream) { + transferTurnToStream(); + const relayed = tapAnthropicSseForLog(source, logCtx, (status, meta) => { + try { + cleanupAbort(); + bindUsage(logCtx.usage); + finishLog(status, undefined, meta.closeReason); + if (meta.closeReason !== "terminal") upstream.abort(); + } finally { + releaseStreamTurn(); + } + }, bodyGuard); + return new Response(relayed, { + status: 200, + headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache", Connection: "keep-alive" }, + }); + } + // A non-streaming caller whose upstream streamed anyway: fold the stream into one message. + const tapState: { closeReason?: FinalRequestLogMeta["closeReason"] } = {}; + const tapped = tapAnthropicSseForLog(source, logCtx, (_status, meta) => { + tapState.closeReason = meta.closeReason; + }, bodyGuard); + try { + const message = await collectAnthropicMessage(tapped, requestedModel, translatorBudget); + cleanupAbort(); + if (tapState.closeReason === "client_cancel" || req.signal.aborted) { + return fail(499, "Client cancelled request", "api_error"); + } + bindUsage(logCtx.usage); + if (message.type === "error") { + const error = isRec(message.error) ? message.error : {}; + return fail(502, typeof error.message === "string" ? error.message : "upstream stream failed", "api_error"); + } + finishLog(200); + return Response.json(message); + } catch (error) { + cleanupAbort(); + upstream.abort(); + if (req.signal.aborted) return fail(499, "Client cancelled request", "api_error"); + if (isTranslatorBudgetExceededError(error)) { + return fail(413, "upstream translation buffer exceeded the safe limit", "request_too_large", "translation_buffer_limit"); + } + return fail(502, error instanceof Error ? error.message : String(error), "api_error"); + } + } + + let read; + try { + read = await readBoundedResponseBody(response, { + signal: upstream.signal, + maxBytes: MAX_NATIVE_MESSAGES_JSON_BYTES, + totalTimeoutMs: Math.max(connectMs, 5_000), + inactivityTimeoutMs: Math.max(connectMs, 5_000), + }); + } catch (error) { + cleanupAbort(); + upstream.abort(); + if (req.signal.aborted) return fail(499, "Client cancelled request", "api_error"); + return fail(502, error instanceof Error ? error.message : String(error), "api_error"); + } + cleanupAbort(); + if (read.oversized) { + upstream.abort(); + return fail(502, "upstream response exceeded the safe limit", "api_error", "translation_buffer_limit"); + } + let message: unknown; + try { + message = JSON.parse(read.text); + } catch { + return fail(502, "upstream returned malformed Messages JSON", "api_error"); + } + if (!isRec(message) || message.type !== "message") { + return fail(502, "upstream response was not a Messages result", "api_error"); + } + bindUsage(anthropicUsageToOcx(isRec(message.usage) ? message.usage : undefined)); + if (logIds) recordFirstOutput(logCtx, logIds.start); + try { + translatorBudget.chargeRetained(Buffer.byteLength(read.text) * 2, { kind: "live_transient" }); + } catch (error) { + if (isTranslatorBudgetExceededError(error)) { + return fail(502, "upstream translation buffer exceeded the safe limit", "api_error", "translation_buffer_limit"); + } + throw error; + } + finishLog(200); + if (requestedStream) { + return new Response(messageAsSse(message), { + status: 200, + headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache" }, + }); + } + return new Response(read.text, { status: 200, headers: { "Content-Type": "application/json" } }); +} + +/** + * A non-OK upstream answer in Anthropic shape. Status and retry policy follow the translated + * Messages path, so a client sees the same contract on either lane: transient 5xx become 529 + * with a retry hint, and a proxy-side replay refusal keeps its no-retry marking. + */ +function nativeMessagesErrorResponse(response: Response, bodyText: string, finishLog: FinishLog): Response { + let upstreamType: string | undefined; + let upstreamMessage: string | undefined; + try { + const parsed = JSON.parse(bodyText) as Rec; + const details = isRec(parsed.error) ? parsed.error : parsed; + if (typeof details.type === "string") upstreamType = details.type; + if (typeof details.message === "string" && details.message.trim()) { + upstreamMessage = redactSecretString(details.message.trim()); + } + } catch { /* keep generic classification */ } + const detail = formatAnthropicErrorBody(response.status, response.headers, bodyText); + const message = upstreamMessage + ?? (detail ? `Provider error ${response.status}: ${detail}` : `Provider error ${response.status}`); + const classified = classifyError( + response.status, + upstreamType ?? (response.status === 401 ? "authentication_error" + : response.status === 429 ? "rate_limit_error" + : response.status >= 500 ? "api_error" : "invalid_request_error"), + message, + ); + const safeMessage = redactSecretString(classified.message); + const replayRefusal = isReplayRefusalResponse(response); + const transient = !replayRefusal && isTransientUpstreamStatus(response.status); + const status = replayRefusal ? REPLAY_REFUSED_STATUS : transient ? 529 : response.status; + const upstreamRetryAfter = response.headers.get("retry-after"); + const retryAfter = replayRefusal + ? undefined + : resolveClientRetryAfter({ status: response.status, message: safeMessage, upstreamRetryAfter }) + ?? (upstreamRetryAfter?.trim() === "0" ? "0" : undefined); + finishLog(response.status, safeMessage); + const headers = new Headers({ "Content-Type": "application/json" }); + if (retryAfter) headers.set("Retry-After", retryAfter); + else if (transient) headers.set("Retry-After", "2"); + if (replayRefusal) applyReplayRefusalClientHeaders(headers); + const out = new Response(JSON.stringify(anthropicErrorBody( + status, + safeMessage, + transient ? "overloaded_error" : upstreamType, + replayRefusal ? UPSTREAM_RESET_REPLAY_REFUSED_CODE : undefined, + )), { status, headers }); + return replayRefusal ? retainReplayRefusal(out) : out; +} + +/** + * The body `count_tokens` should count for this selector: what the native lane would send when + * the route is eligible, else `undefined` so the caller keeps its existing estimate. Reads config + * and the router's deterministic branches only; sends nothing. + */ +export function nativeMessagesCountBody( + config: OcxConfig, + cc: OcxConfig["claudeCode"], + body: Rec, + selector: NativeMessagesSelector, +): Rec | undefined { + if (typeof body.model !== "string") return undefined; + try { + const selectorId = resolveInboundModel(body.model, cc); + // Combo and policy selectors are never eligible, and routing them would advance round-robin + // state or run the policy evaluator for a request that sends nothing. + if (resolvePolicyProfileId(config, selectorId) !== null || selectorId.startsWith(`${POLICY_NAMESPACE}/`)) return undefined; + if (!preservesPhysicalComboProvider(config) && resolveComboId(config, selectorId) !== null) return undefined; + const route = routeModel(config, selectorId); + route.staticPolicy = captureRouteStaticPolicy( + route.providerName, route.modelId, route.provider, route.staticPolicy.effectiveAlias, "anthropic", + ); + route.provider = resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, "anthropic", route.staticPolicy); + if (nativeMessagesDeclineReason(route, body, config, selector) !== undefined) return undefined; + return buildAnthropicMessagesPassthroughRequest(route.provider, route.modelId, body, config).wireBody; + } catch { + return undefined; + } +} From d042f7b0517d46fa4d4aa3da685f1fc627974f4d Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:55:17 +0900 Subject: [PATCH 05/20] feat(claude): route eligible managed-key Messages to the native lane Decided after route settlement and the managed-client steps; the caller-forward branch keeps its own earlier decision. The reject guard judges the native path when it applies. --- src/server/claude-messages.ts | 31 ++++++++++++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index 64924b7ffbb..788b314b265 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -66,11 +66,12 @@ import type { ClientEncoderOption } from "./responses/core-options"; import { handleResponses } from "./responses"; import { upstreamWireForAdapter } from "../protocols/contract"; import { createProtocolEnvelope, type ProtocolEnvelope } from "../protocols/envelope"; -import { featuresFromMessagesBody } from "../protocols/features"; +import { featuresFromMessagesBody, type ProtocolFeature } from "../protocols/features"; import { checkRepresentable, unrepresentableMessage } from "../protocols/guard"; import { requestPathForLane } from "../protocols/path"; import { resolveApiSurfaceSettings, resolveProtocolSettings } from "../protocols/settings"; import { markProtocolBlocked, markProtocolEntry } from "../protocols/trace"; +import { isNativeMessagesRouteEligible } from "./messages-native-eligibility"; import { isApiAuthRequired, isDataPlaneAdmissionSecret, @@ -749,6 +750,7 @@ async function handleClaudeMessagesWithBudget( let requestedModel = ""; // Built only under the reject policy; the legacy default leaves this request untouched. let envelope: ProtocolEnvelope | undefined; + let messagesFeatures: () => Iterable = () => []; try { anthropicBody = await readAnthropicBody(req, translatorBudget, resolveInboundBodyLimitBytes(config.maxInboundBodyBytes)); // Defensive [1m] strip (devlog 138): clients normally remove the context-variant @@ -822,7 +824,7 @@ async function handleClaudeMessagesWithBudget( const sourceEnvelope = envelope; // The bridge entry mark below reads these before an effort override rewrites `thinking`, // which also fixes the envelope's cached features on the caller's own settings. - const messagesFeatures = sourceEnvelope + messagesFeatures = sourceEnvelope ? () => sourceEnvelope.features() : () => featuresFromMessagesBody(messagesBody); if (!effortRow && !fastRow && isRec(anthropicBody) && wantsNativePassthrough(req, config, requestPolicy, anthropicBody.model, cc)) { @@ -982,11 +984,17 @@ async function handleClaudeMessagesWithBudget( /* unknown model: let handleResponses shape the 404 */ } + // PF-08: a managed-key Anthropic route sends its Messages body natively. The caller-forward + // passthrough above was decided on the caller's own credential and never reaches this point. + const nativeMessagesRoute = settledRoute && isRec(anthropicBody) + && isNativeMessagesRouteEligible(settledRoute, anthropicBody, config, { effortRow: !!effortRow, fastRow: !!fastRow }) + ? settledRoute + : undefined; // Combo and policy children are judged per candidate (PF-07); an unknown model has no route. if (envelope && settledRoute && !settledRoute.combo && settledRoute.routeKind !== "policy") { const verdict = checkRepresentable({ inbound: "messages", - requestPath: requestPathForLane("messages", "bridge", upstreamWireForAdapter(settledRoute.provider.adapter)), + requestPath: requestPathForLane("messages", nativeMessagesRoute ? "native" : "bridge", upstreamWireForAdapter(settledRoute.provider.adapter)), features: envelope.features(), policy: "reject", }); @@ -997,6 +1005,23 @@ async function handleClaudeMessagesWithBudget( return anthropicErrorResponse(400, unrepresentableMessage(verdict.features), "invalid_request_error"); } } + if (nativeMessagesRoute) { + markProtocolEntry(logCtx, { inbound: "messages", lane: "native", features: messagesFeatures }); + 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; + } 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"); + } + const { handleNativeMessages } = await import("./messages-native"); + return await handleNativeMessages({ + req, config, logCtx, ...(logIds ? { logIds } : {}), + route: nativeMessagesRoute, body: nativeBody, requestedModel, translatorBudget, + }); + } const headers = new Headers({ "content-type": "application/json" }); let trustedClaudeMainAuth: { authorization: string; chatgptAccountId?: string } | undefined; From 0dca6744cc2a9cfb45df0b02511a8fcdb6a5b909 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:55:17 +0900 Subject: [PATCH 06/20] feat(claude): count the native Messages body in count_tokens With the switch on and an eligible route, count_tokens estimates the allowlisted body the native lane would send; otherwise the existing estimate is unchanged. --- src/server/claude-messages.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index 788b314b265..5760b730192 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -1399,7 +1399,11 @@ export async function handleClaudeCountTokens( if (wantsNativePassthrough(req, config, requestPolicy, model, cc)) { return await anthropicNativePassthrough(req, config, { model, provider: "anthropic-native", surface: "claude" }, undefined, raw, "/v1/messages/count_tokens"); } - const inputTokens = estimateClaudeRequestTokens(raw, model); + // PF-08: an eligible managed-key route counts the body the native lane would send. + const nativeCountBody = resolveProtocolSettings(config).rollout.managedMessagesNative + ? (await import("./messages-native")).nativeMessagesCountBody(config, cc, raw, { fastRow: countFastRow !== null }) + : undefined; + const inputTokens = estimateClaudeRequestTokens(nativeCountBody ?? raw, model); return new Response(JSON.stringify({ input_tokens: inputTokens }), { status: 200, headers: { "Content-Type": "application/json" }, From beda74b10a86bee47552c6a68656936710e06dbd Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:55:39 +0900 Subject: [PATCH 07/20] feat(protocols): preview managed native Messages candidates With the switch on, a Messages candidate is judged by the same decline rule the ingress uses; with it off the preview is unchanged. --- src/protocols/plan-snapshot.ts | 47 +++++++++++++++++++++++++--------- 1 file changed, 35 insertions(+), 12 deletions(-) diff --git a/src/protocols/plan-snapshot.ts b/src/protocols/plan-snapshot.ts index a45f6c5d66c..e7a9684648f 100644 --- a/src/protocols/plan-snapshot.ts +++ b/src/protocols/plan-snapshot.ts @@ -22,6 +22,7 @@ import { captureRouteStaticPolicy, routeConcreteModel, routeModel, type RouteRes import { getRoutingProfile, POLICY_NAMESPACE, resolvePolicyProfileId } from "../routing/profile"; import { resolveWireProtocolOverride } from "../server/adapter-resolve"; import { nativeChatDeclineReason } from "../server/chat-native-eligibility"; +import { nativeMessagesDeclineReason } from "../server/messages-native-eligibility"; import { parseSyntheticRowId } from "../server/fast-row"; import type { OcxConfig } from "../types"; import { inboundWireForProtocol, type Protocol, type ProtocolReasonCode } from "./contract"; @@ -90,13 +91,26 @@ function chatBodyForFeatures(features: ReadonlySet): Record): Record { + const content = features.has("request.images") + ? [{ type: "image", source: { type: "base64", media_type: "image/png", data: "AA==" } }] + : ""; + return { messages: [{ role: "user", content }] }; +} + +interface SyntheticRows { + effortRow: boolean; + fastRow: boolean; +} + function candidateFor( config: OcxConfig, inbound: Protocol, route: RouteResult, routeKind: SettledRouteKind, features: ReadonlySet, - effortRow: boolean, + rows: SyntheticRows, ): ProtocolPlanCandidateInput { const wire = inboundWireForProtocol(inbound); // The same two steps every ingress runs: recapture static policy for the original inbound, @@ -108,17 +122,24 @@ function candidateFor( const adapter = provider.adapter ?? "openai-responses"; let declineReasons: ProtocolReasonCode[] = []; let nativeEligible = false; + // With `nativeChatCombos` on, the combo loop judges each Chat candidate as the concrete route it + // is (PF-07), so the preview must too; a policy still resolves one candidate on the bridge. + const comboChildNative = inbound === "chat" && routeKind === "combo" + && resolveProtocolSettings(config).rollout.nativeChatCombos; + const settled: RouteResult = { + ...route, + provider, + staticPolicy, + ...(routeKind === "direct" || comboChildNative ? {} : { routeKind }), + }; if (inbound === "chat") { - // With `nativeChatCombos` on, the combo loop judges each candidate as the concrete route it - // is (PF-07), so the preview must too; a policy still resolves one candidate on the bridge. - const comboChildNative = routeKind === "combo" && resolveProtocolSettings(config).rollout.nativeChatCombos; - const settled: RouteResult = { - ...route, - provider, - staticPolicy, - ...(routeKind === "direct" || comboChildNative ? {} : { routeKind }), - }; - const reason = effortRow ? "effort-row" : nativeChatDeclineReason(settled, chatBodyForFeatures(features), config); + const reason = rows.effortRow ? "effort-row" : nativeChatDeclineReason(settled, chatBodyForFeatures(features), config); + nativeEligible = reason === undefined; + if (reason) declineReasons = [reason]; + } else if (inbound === "messages" && resolveProtocolSettings(config).rollout.managedMessagesNative) { + // The runtime rule itself. With the switch off nothing is judged, so the default preview + // is exactly what it was before the managed native lane existed. + const reason = nativeMessagesDeclineReason(settled, messagesBodyForFeatures(features), config, rows); nativeEligible = reason === undefined; if (reason) declineReasons = [reason]; } @@ -153,6 +174,7 @@ export function buildProtocolPlanSnapshot( const reasonCodes: ProtocolReasonCode[] = []; let routeKey = request.model; let effortRow = false; + let fastRow = false; let syntheticRow = false; try { const parsed = parseSyntheticRowId(request.model, config); @@ -162,6 +184,7 @@ export function buildProtocolPlanSnapshot( syntheticRow = true; } else if (parsed.fastRow) { routeKey = parsed.fastRow.baseId; + fastRow = true; syntheticRow = true; } } catch { @@ -182,7 +205,7 @@ export function buildProtocolPlanSnapshot( return { ...base, routeKind: settled.routeKind, - candidates: settled.routes.map(route => candidateFor(config, request.inbound, route, settled.routeKind, features, effortRow)), + candidates: settled.routes.map(route => candidateFor(config, request.inbound, route, settled.routeKind, features, { effortRow, fastRow })), reasonCodes, }; } From 3cfded6fffd907b0ec03e38830b6b0c997727651 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:59:12 +0900 Subject: [PATCH 08/20] test(claude): pin the native Messages builder, decline rule and preview The builder keeps only allowlisted fields and the adapter's headers; each decline reason is named; the planner reports native only with the switch on and a managed key. --- scripts/test-layout/layout.json | 2 + .../anthropic-messages-passthrough.test.ts | 105 ++++++++++++++ tests/fixtures/test-layout-expected.json | 2 + .../messages-native-eligibility.test.ts | 128 ++++++++++++++++++ 4 files changed, 237 insertions(+) create mode 100644 tests/adapters/anthropic/anthropic-messages-passthrough.test.ts create mode 100644 tests/responses/messages-native-eligibility.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 0298c49c554..7fe58871bd6 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -238,6 +238,7 @@ "anthropic-tail-guard.test.ts": "adapters/anthropic", "anthropic-thinking-signature.test.ts": "adapters/anthropic", "anthropic-tool-call-id.test.ts": "adapters/anthropic", + "anthropic-messages-passthrough.test.ts": "adapters/anthropic", "anthropic-tool-declaration-constraints.test.ts": "adapters/anthropic", "anthropic-tool-schema.test.ts": "adapters/anthropic", "antigravity-baseurl-override.test.ts": "adapters/google", @@ -343,6 +344,7 @@ "protocol-direct-encoders-chat.test.ts": "responses", "protocol-direct-encoders-messages.test.ts": "responses", "chat-native-combo.test.ts": "responses", + "messages-native-eligibility.test.ts": "responses", "chat-inbound-reasoning-none.test.ts": "responses", "chat-native-decline-reason.test.ts": "responses", "chat-inbound-reasoning-replay.test.ts": "responses", diff --git a/tests/adapters/anthropic/anthropic-messages-passthrough.test.ts b/tests/adapters/anthropic/anthropic-messages-passthrough.test.ts new file mode 100644 index 00000000000..e2184b2b170 --- /dev/null +++ b/tests/adapters/anthropic/anthropic-messages-passthrough.test.ts @@ -0,0 +1,105 @@ +/** + * The managed native Messages builder (src/adapters/anthropic/passthrough.ts, PF-08): the + * allowlisted source body with the wire model, the adapter's own URL, pinned version and key + * placement, and nothing taken from the caller. + */ +import { describe, expect, test } from "bun:test"; +import { ANTHROPIC_API_VERSION, createAnthropicAdapter } from "../../../src/adapters/anthropic"; +import { + ANTHROPIC_MESSAGES_PASSTHROUGH_FIELDS, + buildAnthropicMessagesPassthroughRequest, +} from "../../../src/adapters/anthropic/passthrough"; +import { createTranslatorBudget } from "../../../src/lib/translator-budget"; +import type { OcxParsedRequest, OcxProviderConfig } from "../../../src/types"; + +function provider(overrides: Partial = {}): OcxProviderConfig { + return { + adapter: "anthropic", + baseUrl: "https://anthropic.example/v1", + authMode: "key", + apiKey: "fixture-managed-key", + ...overrides, + } as OcxProviderConfig; +} + +const SOURCE = { + model: "client-selector", + max_tokens: 64, + top_k: 7, + temperature: 0.2, + stream: true, + thinking: { type: "enabled", budget_tokens: 2048 }, + system: [{ type: "text", text: "fixture system", cache_control: { type: "ephemeral" } }], + messages: [{ role: "user", content: [{ type: "text", text: "fixture", cache_control: { type: "ephemeral", ttl: "1h" } }] }], + metadata: { user_id: "fixture-user" }, + // Not on the allowlist: dropped rather than forwarded unchecked. + context_management: { edits: [] }, + mcp_servers: [{ type: "url", url: "https://mcp.example" }], + container: "fixture-container", +}; + +describe("buildAnthropicMessagesPassthroughRequest", () => { + test("keeps exactly the allowlisted source fields and swaps in the wire model", () => { + const built = buildAnthropicMessagesPassthroughRequest(provider(), "claude-wire", SOURCE); + const expected: Record = {}; + for (const [key, value] of Object.entries(SOURCE)) { + if ((ANTHROPIC_MESSAGES_PASSTHROUGH_FIELDS as readonly string[]).includes(key)) expected[key] = value; + } + expected.model = "claude-wire"; + expect(JSON.parse(built.body)).toEqual(expected); + expect(built.wireBody).toEqual(expected); + expect(built.wireBody).toMatchObject({ top_k: 7, thinking: SOURCE.thinking, system: SOURCE.system }); + expect(built.wireBody).not.toHaveProperty("context_management"); + expect(built.wireBody).not.toHaveProperty("mcp_servers"); + expect(built.wireBody).not.toHaveProperty("container"); + // The source is not mutated. + expect(SOURCE.model).toBe("client-selector"); + }); + + test("uses the adapter's endpoint, version and key placement", async () => { + const built = buildAnthropicMessagesPassthroughRequest(provider(), "claude-wire", SOURCE); + expect(built.url).toBe("https://anthropic.example/v1/messages"); + expect(built.headers).toMatchObject({ + "anthropic-version": ANTHROPIC_API_VERSION, + "x-api-key": "fixture-managed-key", + Accept: "text/event-stream", + "Content-Type": "application/json", + }); + expect(built.headers).not.toHaveProperty("Authorization"); + + const parsed = { + modelId: "claude-wire", + stream: true, + context: { messages: [{ role: "user", content: "fixture", timestamp: 0 }] }, + options: {}, + } as unknown as OcxParsedRequest; + const adapterRequest = await createAnthropicAdapter(provider()) + .buildRequest(parsed, { headers: new Headers(), translatorBudget: createTranslatorBudget() }); + expect(adapterRequest.url).toBe(built.url); + const adapterHeaders = adapterRequest.headers as Record; + for (const name of ["anthropic-version", "x-api-key", "Accept", "User-Agent"]) { + expect(built.headers[name]).toBe(adapterHeaders[name]); + } + }); + + test("a bearer key transport and operator headers apply as in the adapter", () => { + const built = buildAnthropicMessagesPassthroughRequest( + provider({ apiKeyTransport: "bearer", headers: { "x-operator": "fixture" } } as Partial), + "claude-wire", + { ...SOURCE, stream: false }, + ); + expect(built.headers.Authorization).toBe("Bearer fixture-managed-key"); + expect(built.headers).not.toHaveProperty("x-api-key"); + expect(built.headers["x-operator"]).toBe("fixture"); + expect(built.headers.Accept).toBe("application/json"); + }); + + test("refuses what the adapter refuses and never carries an OAuth or forwarded credential", () => { + expect(() => buildAnthropicMessagesPassthroughRequest(provider({ apiKey: "" }), "m", SOURCE)) + .toThrow("non-empty apiKey"); + expect(() => buildAnthropicMessagesPassthroughRequest(provider({ baseUrl: "https://{region}.example/v1" }), "m", SOURCE)) + .toThrow("unresolved {region}"); + expect(() => buildAnthropicMessagesPassthroughRequest(provider({ authMode: "oauth" }), "m", SOURCE)).toThrow(); + expect(() => buildAnthropicMessagesPassthroughRequest(provider({ authMode: "forward" }), "m", SOURCE)).toThrow(); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 5fd7bc92bae..54be7f84079 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -64,6 +64,7 @@ "anthropic-tail-guard.test.ts": "adapters/anthropic", "anthropic-thinking-signature.test.ts": "adapters/anthropic", "anthropic-tool-call-id.test.ts": "adapters/anthropic", + "anthropic-messages-passthrough.test.ts": "adapters/anthropic", "anthropic-tool-declaration-constraints.test.ts": "adapters/anthropic", "anthropic-tool-schema.test.ts": "adapters/anthropic", "antigravity-baseurl-override.test.ts": "adapters/google", @@ -169,6 +170,7 @@ "protocol-direct-encoders-chat.test.ts": "responses", "protocol-direct-encoders-messages.test.ts": "responses", "chat-native-combo.test.ts": "responses", + "messages-native-eligibility.test.ts": "responses", "chat-inbound-reasoning-none.test.ts": "responses", "chat-native-decline-reason.test.ts": "responses", "chat-inbound-reasoning-replay.test.ts": "responses", diff --git a/tests/responses/messages-native-eligibility.test.ts b/tests/responses/messages-native-eligibility.test.ts new file mode 100644 index 00000000000..2bd3a0d6049 --- /dev/null +++ b/tests/responses/messages-native-eligibility.test.ts @@ -0,0 +1,128 @@ +/** + * Which Messages routes take the managed native lane (PF-08), and that the planner reports the + * same rule: `nativeMessagesDeclineReason` (src/server/messages-native-eligibility.ts) and the + * Messages candidates of `buildProtocolPlanSnapshot` (src/protocols/plan-snapshot.ts). + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { buildProtocolPlanSnapshot, previewProtocolPlan } from "../../src/protocols/plan-snapshot"; +import type { RouteResult } from "../../src/router"; +import { + isNativeMessagesRouteEligible, + nativeMessagesDeclineReason, +} from "../../src/server/messages-native-eligibility"; +import type { OcxConfig } from "../../src/types"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +let testDir = ""; +let previousHome: string | undefined; + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + testDir = mkdtempSync(join(tmpdir(), "ocx-messages-native-eligibility-")); + process.env.OPENCODEX_HOME = testDir; +}); + +afterEach(() => { + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (testDir) removeTreeWithRetry(testDir); +}); + +const ON = { protocols: { rollout: { managedMessagesNative: true } } } as Partial; + +function config(overrides: Partial = {}): OcxConfig { + return { + port: 10100, + defaultProvider: "anth", + providers: { + anth: { adapter: "anthropic", baseUrl: "https://anth.example/v1", authMode: "key", apiKey: "ka", models: ["claude-x"] }, + anthoauth: { adapter: "anthropic", baseUrl: "https://anth.example/v1", authMode: "oauth", apiKey: "ko", models: ["claude-o"] }, + chat: { adapter: "openai-chat", baseUrl: "https://chat.example/v1", apiKey: "kc", models: ["m1"] }, + }, + combos: { + pair: { strategy: "failover", targets: [{ provider: "anth", model: "claude-x" }, { provider: "chat", model: "m1" }] }, + }, + ...overrides, + } as OcxConfig; +} + +type RouteOverrides = Omit, "provider"> & { adapter?: string; authMode?: string; provider?: Record }; + +function route(overrides: RouteOverrides = {}): RouteResult { + const { adapter = "anthropic", authMode, provider, ...rest } = overrides; + return { + providerName: "anth", + modelId: "claude-x", + routeKind: "direct", + routeReason: "test", + provider: { adapter, baseUrl: "https://anth.example/v1", apiKey: "ka", ...(authMode ? { authMode } : {}), ...provider }, + ...rest, + } as unknown as RouteResult; +} + +const TEXT_BODY = { messages: [{ role: "user", content: "fixture" }] }; +const IMAGE_BODY = { + messages: [{ role: "user", content: [{ type: "image", source: { type: "base64", media_type: "image/png", data: "AA==" } }] }], +}; + +describe("native Messages decline reasons", () => { + test("the switch is checked first and defaults off", () => { + expect(nativeMessagesDeclineReason(route(), TEXT_BODY, config())).toBe("rollout-disabled"); + expect(isNativeMessagesRouteEligible(route(), TEXT_BODY, config())).toBe(false); + }); + + test("an eligible managed-key Anthropic route has no reason", () => { + expect(nativeMessagesDeclineReason(route(), TEXT_BODY, config(ON))).toBeUndefined(); + expect(nativeMessagesDeclineReason(route({ authMode: "key" }), TEXT_BODY, config(ON))).toBeUndefined(); + expect(isNativeMessagesRouteEligible(route(), TEXT_BODY, config(ON))).toBe(true); + }); + + test("each rule names its reason", () => { + const on = config(ON); + expect(nativeMessagesDeclineReason(route({ adapter: "openai-chat" }), TEXT_BODY, on)).toBe("cross-wire-ir"); + expect(nativeMessagesDeclineReason(route({ authMode: "oauth" }), TEXT_BODY, on)).toBe("auth-mode-not-native"); + expect(nativeMessagesDeclineReason(route({ authMode: "forward" }), TEXT_BODY, on)).toBe("auth-mode-not-native"); + expect(nativeMessagesDeclineReason(route({ routeKind: "policy" } as RouteOverrides), TEXT_BODY, on)).toBe("combo-or-policy-route"); + expect(nativeMessagesDeclineReason(route({ routeKind: "combo" } as RouteOverrides), TEXT_BODY, on)).toBe("combo-or-policy-route"); + expect(nativeMessagesDeclineReason(route(), TEXT_BODY, on, { effortRow: true })).toBe("effort-row"); + expect(nativeMessagesDeclineReason(route(), TEXT_BODY, on, { fastRow: true })).toBe("fast-row"); + }); + + test("an image for a model declared text-only needs vision preprocessing", () => { + const blind = route({ provider: { modelCapabilities: { "claude-x": { inputModalities: ["text"] } } } }); + expect(nativeMessagesDeclineReason(blind, IMAGE_BODY, config(ON))).toBe("vision-preprocessing"); + expect(nativeMessagesDeclineReason(blind, TEXT_BODY, config(ON))).toBeUndefined(); + expect(nativeMessagesDeclineReason(route(), IMAGE_BODY, config(ON))).toBeUndefined(); + }); +}); + +describe("planner mirrors the native Messages rule", () => { + test("switch off: the preview is unchanged (bridge, no decline reason)", () => { + const snapshot = buildProtocolPlanSnapshot(config(), { model: "anth/claude-x", inbound: "messages", features: [] }); + expect(snapshot.candidates).toEqual([ + { provider: "anth", model: "claude-x", adapter: "anthropic", nativeEligible: false, declineReasons: [] }, + ]); + expect(previewProtocolPlan(config(), { model: "anth/claude-x", inbound: "messages", features: [] }).mode).toBe("legacy-bridge"); + }); + + test("switch on: a managed-key Anthropic candidate is native", () => { + const snapshot = buildProtocolPlanSnapshot(config(ON), { model: "anth/claude-x", inbound: "messages", features: ["request.top_k"] }); + expect(snapshot.candidates).toEqual([ + { provider: "anth", model: "claude-x", adapter: "anthropic", nativeEligible: true, declineReasons: [] }, + ]); + const plan = previewProtocolPlan(config(ON), { model: "anth/claude-x", inbound: "messages", features: ["request.top_k"] }); + expect(plan.mode).toBe("native"); + expect(plan.candidates[0]).toMatchObject({ requestPath: ["messages", "messages"], eligible: true }); + expect(plan.guaranteedFeatures).toEqual(["request.top_k"]); + }); + + test("switch on: OAuth and combo candidates report the rule that declined them", () => { + const oauth = buildProtocolPlanSnapshot(config(ON), { model: "anthoauth/claude-o", inbound: "messages", features: [] }); + expect(oauth.candidates[0]).toMatchObject({ nativeEligible: false, declineReasons: ["auth-mode-not-native"] }); + const combo = buildProtocolPlanSnapshot(config(ON), { model: "combo/pair", inbound: "messages", features: [] }); + expect(combo.candidates[0]).toMatchObject({ provider: "anth", nativeEligible: false, declineReasons: ["combo-or-policy-route"] }); + }); +}); From 258012019560218d7ef88f1d752ecbb99bed1d8e Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:59:12 +0900 Subject: [PATCH 09/20] test(claude): exercise managed native Messages against a fake upstream Allowlisted body and provider key on the wire, 401/429 key failover, stream relay and usage, JSON callers, switch-off bridge, caller-forward precedence and count_tokens. --- scripts/test-layout/layout.json | 1 + .../messages-native.test.ts | 265 ++++++++++++++++++ tests/fixtures/test-layout-expected.json | 1 + 3 files changed, 267 insertions(+) create mode 100644 tests/claude-integration/messages-native.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 7fe58871bd6..d246747b47d 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -421,6 +421,7 @@ "claude-management-api.test.ts": "claude-integration", "claude-manual-env.test.ts": "gui", "claude-messages-endpoint.test.ts": "claude-integration", + "messages-native.test.ts": "claude-integration", "messages-surface-matrix.test.ts": "claude-integration", "claude-model-info.test.ts": "claude-integration", "claude-models-discovery.test.ts": "claude-integration", diff --git a/tests/claude-integration/messages-native.test.ts b/tests/claude-integration/messages-native.test.ts new file mode 100644 index 00000000000..24eb3551070 --- /dev/null +++ b/tests/claude-integration/messages-native.test.ts @@ -0,0 +1,265 @@ +/** + * Managed native Messages lane (PF-08) against fake upstreams. With + * `protocols.rollout.managedMessagesNative` on, a direct route to a key-auth `anthropic` + * provider receives the caller's Messages body itself (allowlisted, wire model, the provider's + * key) instead of the Responses replay. Everything here is a local fixture: no real credential + * or service is reached. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { saveConfig } from "../../src/config"; +import { clearKeyCooldowns } from "../../src/providers/key-failover"; +import { estimateClaudeRequestTokens, handleClaudeCountTokens, handleClaudeMessages } from "../../src/server/claude-messages"; +import { getRequestLogEntries } from "../../src/server/request-log"; +import type { OcxConfig } from "../../src/types"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +interface Seen { + path: string; + headers: Headers; + body: Record; +} + +type Reply = (seen: Seen, index: number) => Response; + +let upstream: ReturnType | undefined; +let callerForward: ReturnType | undefined; +let seen: Seen[] = []; +let callerForwardSeen: Seen[] = []; +let releaseSpendHome: (() => void) | undefined; +let testDir = ""; +let previousHome: string | undefined; + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + testDir = mkdtempSync(join(tmpdir(), "ocx-messages-native-")); + process.env.OPENCODEX_HOME = testDir; + clearKeyCooldowns(); + seen = []; + callerForwardSeen = []; +}); + +afterEach(async () => { + releaseSpendHome?.(); + releaseSpendHome = undefined; + await upstream?.stop(true); + upstream = undefined; + await callerForward?.stop(true); + callerForward = undefined; + clearKeyCooldowns(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (testDir) removeTreeWithRetry(testDir); +}); + +const MESSAGE_JSON = { + id: "msg_fixture", + type: "message", + role: "assistant", + model: "claude-x", + content: [{ type: "text", text: "fixture reply" }], + stop_reason: "end_turn", + stop_sequence: null, + usage: { input_tokens: 11, output_tokens: 3 }, +}; + +const SSE_FRAMES = [ + { event: "message_start", data: { type: "message_start", message: { ...MESSAGE_JSON, content: [], stop_reason: null, usage: { input_tokens: 5, output_tokens: 0, cache_read_input_tokens: 2 } } } }, + { event: "content_block_start", data: { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } } }, + { event: "content_block_delta", data: { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "streamed" } } }, + { event: "content_block_stop", data: { type: "content_block_stop", index: 0 } }, + { event: "message_delta", data: { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 7 } } }, + { event: "message_stop", data: { type: "message_stop" } }, +]; +const SSE_TEXT = SSE_FRAMES.map(frame => `event: ${frame.event}\ndata: ${JSON.stringify(frame.data)}\n\n`).join(""); + +function ok(seenRequest: Seen): Response { + if (seenRequest.body.stream === true) { + return new Response(SSE_TEXT, { headers: { "content-type": "text/event-stream" } }); + } + return Response.json(MESSAGE_JSON); +} + +function startUpstream(reply: Reply = ok): number { + upstream = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(req) { + const entry = { path: new URL(req.url).pathname, headers: req.headers, body: await req.json() as Record }; + seen.push(entry); + return reply(entry, seen.length - 1); + } }); + return upstream.port!; +} + +function fixtureConfig(port: number, options: { on?: boolean; pool?: boolean; claudeCode?: OcxConfig["claudeCode"] } = {}): OcxConfig { + const { on = true, pool = false } = options; + releaseSpendHome ??= acquireOwnedSpendHome(); + const config = { + port: 0, + defaultProvider: "anth", + providers: { anth: { + adapter: "anthropic", + baseUrl: `http://127.0.0.1:${port}`, + authMode: "key", + apiKey: "fixture-key-alpha", + allowPrivateNetwork: true, + models: ["claude-x"], + ...(pool ? { apiKeyPool: [ + { id: "k1", key: "fixture-key-alpha", addedAt: 1 }, + { id: "k2", key: "fixture-key-beta", addedAt: 2 }, + ] } : {}), + } }, + ...(on ? { protocols: { rollout: { managedMessagesNative: true } } } : {}), + ...(options.claudeCode ? { claudeCode: options.claudeCode } : {}), + } as OcxConfig; + saveConfig(config); + return config; +} + +const SOURCE_BODY = { + model: "anth/claude-x", + max_tokens: 64, + top_k: 5, + temperature: 0.3, + thinking: { type: "enabled", budget_tokens: 1024 }, + metadata: { user_id: "fixture-user" }, + system: [{ type: "text", text: "fixture system", cache_control: { type: "ephemeral" } }], + messages: [{ role: "user", content: [{ type: "text", text: "fixture question", cache_control: { type: "ephemeral" } }] }], + tools: [{ name: "lookup", description: "fixture tool", input_schema: { type: "object", properties: {} } }], + // Not on the allowlist: must not reach the provider. + context_management: { edits: [] }, +}; + +function messagesRequest(body: Record, headers: Record = {}): Request { + return new Request("http://localhost/v1/messages", { + method: "POST", + headers: { "content-type": "application/json", ...headers }, + body: JSON.stringify(body), + }); +} + +function rowFor(requestId: string) { + const rows = getRequestLogEntries().filter(entry => entry.requestId === requestId); + expect(rows).toHaveLength(1); + return rows[0]!; +} + +async function send(config: OcxConfig, body: Record, headers?: Record) { + const requestId = `pf08-${crypto.randomUUID()}`; + const response = await handleClaudeMessages(messagesRequest(body, headers), config, { model: "", provider: "" }, + { requestId, start: Date.now() }); + const text = await response.text(); + return { requestId, response, text }; +} + +describe("managed native Messages", () => { + test("sends exactly the allowlisted source body with the provider key and no caller credential", async () => { + const config = fixtureConfig(startUpstream()); + const { requestId, response, text } = await send(config, { ...SOURCE_BODY, stream: false }, { + authorization: "Bearer fixture-admission-token", + "x-api-key": "fixture-caller-key", + "anthropic-beta": "fixture-beta", + }); + expect(response.status).toBe(200); + expect(JSON.parse(text)).toEqual(MESSAGE_JSON); + expect(seen).toHaveLength(1); + const sent = seen[0]!; + expect(sent.path).toBe("/v1/messages"); + const { context_management: _dropped, ...allowlisted } = SOURCE_BODY; + expect(sent.body).toEqual({ ...allowlisted, stream: false, model: "claude-x" }); + expect(sent.headers.get("x-api-key")).toBe("fixture-key-alpha"); + expect(sent.headers.get("authorization")).toBeNull(); + expect(sent.headers.get("anthropic-beta")).toBeNull(); + expect(sent.headers.get("anthropic-version")).toBe("2023-06-01"); + + const row = rowFor(requestId); + expect(row.status).toBe(200); + expect(row.usage).toMatchObject({ inputTokens: 11, outputTokens: 3 }); + expect(row.protocolTrace).toMatchObject({ inbound: "messages", mode: "native", requestPath: ["messages", "messages"] }); + expect(JSON.stringify(row)).not.toContain("fixture-key-alpha"); + }); + + test("relays the upstream stream unchanged and records its usage", async () => { + const config = fixtureConfig(startUpstream()); + const { requestId, response, text } = await send(config, { ...SOURCE_BODY, stream: true }); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("text/event-stream"); + expect(text).toBe(SSE_TEXT); + expect(seen[0]!.body.stream).toBe(true); + const row = rowFor(requestId); + expect(row.status).toBe(200); + expect(row.usage).toMatchObject({ inputTokens: 7, outputTokens: 7, cacheReadInputTokens: 2 }); + }); + + test("a 401 on the first pooled key fails over to the next key", async () => { + const config = fixtureConfig(startUpstream((entry, index) => index === 0 + ? Response.json({ type: "error", error: { type: "authentication_error", message: "invalid x-api-key" } }, { status: 401 }) + : ok(entry)), { pool: true }); + const { response } = await send(config, { ...SOURCE_BODY, stream: false }); + expect(response.status).toBe(200); + expect(seen.map(entry => entry.headers.get("x-api-key"))).toEqual(["fixture-key-alpha", "fixture-key-beta"]); + expect(seen[1]!.body).toEqual(seen[0]!.body); + }); + + test("a 429 on the first pooled key fails over to the next key", async () => { + const config = fixtureConfig(startUpstream((entry, index) => index === 0 + ? Response.json({ type: "error", error: { type: "rate_limit_error", message: "slow down" } }, + { status: 429, headers: { "retry-after": "30" } }) + : ok(entry)), { pool: true }); + const { response } = await send(config, { ...SOURCE_BODY, stream: false }); + expect(response.status).toBe(200); + expect(seen.map(entry => entry.headers.get("x-api-key"))).toEqual(["fixture-key-alpha", "fixture-key-beta"]); + }); + + test("an upstream error answers in Anthropic shape without the key", async () => { + const config = fixtureConfig(startUpstream(() => Response.json( + { type: "error", error: { type: "invalid_request_error", message: "bad fixture" } }, { status: 400 }))); + const { response, text } = await send(config, { ...SOURCE_BODY, stream: false }); + expect(response.status).toBe(400); + expect(JSON.parse(text)).toMatchObject({ type: "error", error: { type: "invalid_request_error", message: "bad fixture" } }); + expect(text).not.toContain("fixture-key-alpha"); + }); + + test("switch off: the same request still takes the Responses bridge", async () => { + const config = fixtureConfig(startUpstream(), { on: false }); + const { requestId, response } = await send(config, { ...SOURCE_BODY, stream: false }); + expect(response.status).toBe(200); + expect(seen).toHaveLength(1); + // The bridge rebuilds the body through the adapter, which has no top_k. + expect(seen[0]!.body).not.toHaveProperty("top_k"); + expect(seen[0]!.body.stream).toBe(true); + expect(rowFor(requestId).protocolTrace).toMatchObject({ inbound: "messages", mode: "legacy-bridge" }); + }); + + test("caller-forward passthrough is decided first and keeps the caller's credential", async () => { + callerForward = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(req) { + callerForwardSeen.push({ path: new URL(req.url).pathname, headers: req.headers, body: await req.json() as Record }); + return Response.json(MESSAGE_JSON); + } }); + const config = fixtureConfig(startUpstream(), { + claudeCode: { anthropicBaseUrl: `http://127.0.0.1:${callerForward.port}` } as OcxConfig["claudeCode"], + }); + const { response } = await send(config, { model: "claude-haiku-4-5", max_tokens: 16, stream: false, + messages: [{ role: "user", content: "fixture" }] }, { "x-api-key": "sk-ant-fixture-caller" }); + expect(response.status).toBe(200); + expect(seen).toHaveLength(0); + expect(callerForwardSeen).toHaveLength(1); + expect(callerForwardSeen[0]!.headers.get("x-api-key")).toBe("sk-ant-fixture-caller"); + }); + + test("count_tokens counts the body the native lane sends", async () => { + const config = fixtureConfig(startUpstream()); + await send(config, { ...SOURCE_BODY, stream: false }); + const sentBody = seen[0]!.body; + const response = await handleClaudeCountTokens(new Request("http://localhost/v1/messages/count_tokens", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(SOURCE_BODY), + }), config); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ input_tokens: estimateClaudeRequestTokens(sentBody, "anth/claude-x") }); + // Counting sends nothing. + expect(seen).toHaveLength(1); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 54be7f84079..9a702263cc3 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -247,6 +247,7 @@ "claude-management-api.test.ts": "claude-integration", "claude-manual-env.test.ts": "gui", "claude-messages-endpoint.test.ts": "claude-integration", + "messages-native.test.ts": "claude-integration", "messages-surface-matrix.test.ts": "claude-integration", "claude-model-info.test.ts": "claude-integration", "claude-models-discovery.test.ts": "claude-integration", From e84b404babb214808cb803d2635545c901138100 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 12:59:49 +0900 Subject: [PATCH 10/20] docs(structure): describe the managed native Messages lane Protocol paths owns the rule, builder and authority split; the Responses transport doc lists the lane among the native senders; the devlog inventory records what stays bridged. --- .../040_acceptance_and_rollout.md | 3 + structure/data-planes/protocol-paths.md | 68 ++++++++++++++++--- structure/transports/responses.md | 7 +- 3 files changed, 67 insertions(+), 11 deletions(-) diff --git a/devlog/_plan/260924_protocol_first_class/040_acceptance_and_rollout.md b/devlog/_plan/260924_protocol_first_class/040_acceptance_and_rollout.md index d95a6b0cf3b..c0f2c82648e 100644 --- a/devlog/_plan/260924_protocol_first_class/040_acceptance_and_rollout.md +++ b/devlog/_plan/260924_protocol_first_class/040_acceptance_and_rollout.md @@ -48,3 +48,6 @@ Kept current by each packet that migrates something. | Responses-only features on Chat/Messages | `previous_response_id`, `store`, `background`, compaction stay on the bridge | | Non-public-wire adapters (`other`) | translated through the IR; no feature claims | | OAuth native Chat | not planned in this unit | +| Messages → key-auth Anthropic | native behind `managedMessagesNative` (PF-08); bridge while off. Caller `anthropic-beta` is not forwarded (PF-10 allowlist); top-level fields outside the allowlist are dropped with no feature effect; the response `model` is the upstream's wire id | +| Messages → Anthropic OAuth | bridge until `managedMessagesNativeOAuth` (PF-10) | +| Messages native lane, translated-only steps | skill elision, `stabilizePromptCache`, pinned route effort and the Claude web-search/vision sidecars run only on the bridge; routes needing vision preprocessing stay there | diff --git a/structure/data-planes/protocol-paths.md b/structure/data-planes/protocol-paths.md index a68d33c9c25..a0e04c500e8 100644 --- a/structure/data-planes/protocol-paths.md +++ b/structure/data-planes/protocol-paths.md @@ -23,7 +23,9 @@ Existing spellings keep their names and map through explicit functions in the sa `nativeChatDeclineReason` in `src/server/chat-native-eligibility.ts` names, as one of these reason codes, the first rule that keeps a Chat request off the native Chat lane; `isNativeChatRouteEligible` is defined as "no reason", so the lane decision and the reason a plan -or trace reports cannot disagree. +or trace reports cannot disagree. `nativeMessagesDeclineReason` in +`src/server/messages-native-eligibility.ts` does the same for the managed native Messages lane +(below). `contract.ts`, `src/protocols/features.ts`, `src/protocols/baseline.ts`, `src/protocols/path.ts`, `src/protocols/dto.ts`, `src/protocols/plan.ts` and `src/protocols/guard.ts` are leaf modules: the dashboard imports them directly, so they import @@ -59,7 +61,8 @@ path the protocol-first-class work targets. Today Chat to Chat and Responses to native, Chat and Messages reach a Responses upstream through their codec directly, and every other Chat or Messages pair (including Messages to a proxy-managed Anthropic key) travels through `responses-internal`; every routed Chat or Messages path streams internally and folds for a -non-streaming client. No target cell contains `responses-internal`. +non-streaming client. No target cell contains `responses-internal`. `current` describes the +default configuration, with every rollout switch off. `tests/responses/protocol-baseline.test.ts` pins both sides. ## Plan and trace shapes @@ -75,8 +78,8 @@ exposes, within fixed limits, and both validators reject anything that is not ex `src/protocols/trace.ts` is server side and is not a leaf. The Chat Completions ingress (`src/server/chat-completions.ts`) marks the lane it chose, the reason code that declined the native lane, and the request's features; the Messages ingress (`src/server/claude-messages.ts`) -marks caller-forward passthrough as the native lane, the translated path as the bridge, and a -disabled surface or a compatibility reject as blocked. The Responses ingress needs no mark: its +marks caller-forward passthrough and the managed native lane as native, the translated path as +the bridge, and a disabled surface or a compatibility reject as blocked. The Responses ingress needs no mark: its path follows the final adapter's wire. Marks live in WeakMaps keyed by the request log context and the live attempt objects, and no mark function throws into the request. @@ -112,7 +115,10 @@ candidate's wire is settled the way the ingress settles it (`captureRouteStaticP original inbound, then `resolveWireProtocolOverride`), and a Chat candidate's native lane is judged by `nativeChatDeclineReason` against a structural body built from the requested features. With `nativeChatCombos` on, a combo's candidates are judged as the concrete routes they are, as the -combo loop judges them; policy candidates keep `combo-or-policy-route`. Messages +combo loop judges them; policy candidates keep `combo-or-policy-route`. With +`protocols.rollout.managedMessagesNative` on, a Messages candidate is judged the same way by +`nativeMessagesDeclineReason`; with it off Messages candidates carry no decline reason, exactly as +before the lane existed. Messages caller-forward passthrough depends on the caller's own credential, so it is reported as `caller-credential-required` and never assumed. The OpenCode Go session-lane transport is not modelled. `tests/responses/protocol-plan-snapshot.test.ts` pins the no-side-effect property against @@ -141,7 +147,8 @@ hop (the internal Responses body) still does. Only when `resolveProtocolSettings(config).unrepresentable === "reject"` do the Chat and Messages ingresses build an envelope and run the guard, after the route and its wire settle and before the request is sent: Chat on the native path when the native lane was chosen, otherwise on the -bridge path to the settled adapter's wire; Messages on the bridge path. Combo and policy routes +bridge path to the settled adapter's wire; Messages on the native path when the managed native +lane was chosen, otherwise on the bridge path. Combo and policy routes and an unroutable model are not judged at ingress; with `nativeChatCombos` on, a Chat combo's candidates are judged one by one inside the combo loop (below). A refusal answers 400 in the ingress's own error shape (Chat `invalid_request_error` / `unsupported_feature`; Anthropic @@ -199,6 +206,47 @@ per candidate. Policy routes select a single candidate in the router and stay on transport side (send budget, failover, logging) is in [Responses transport](../transports/responses.md#native-chat-candidates-in-combos). +## Managed native Messages + +Behind `protocols.rollout.managedMessagesNative` (default off). A Messages request whose settled +route is a direct, key-auth `anthropic` provider is sent as Messages instead of replaying through +Responses. `nativeMessagesDeclineReason` names the first rule that keeps a route off the lane: +`rollout-disabled`, `cross-wire-ir` (another adapter), `auth-mode-not-native` (OAuth, which is +PF-10, or `forward`), `combo-or-policy-route`, `effort-row` / `fast-row` (synthetic rows need the +adapter's wire rewrite), `vision-preprocessing` (an image for a model declared unable to read +it). The ingress, `count_tokens` and the planner all ask it. + +`src/server/claude-messages.ts` decides the lane after the route and its wire settle and after the +managed-client steps already applied to the body (alias/modelMap resolution, `ocx-route`, effort +directives). The caller-forward passthrough is decided earlier, on the caller's own credential, +and returns before this point; the two branches share no credential and no header. The body sent +is `envelope.freshBody()` when a source envelope exists, otherwise the ingress's own body. +`src/server/messages-native.ts` is imported lazily, only for an eligible route. + +`buildAnthropicMessagesPassthroughRequest` in `src/adapters/anthropic/passthrough.ts` builds the +request from that body: the top-level allowlist (`model, messages, system, max_tokens, metadata, +stop_sequences, stream, temperature, top_p, top_k, tools, tool_choice, thinking, output_config, +service_tier`), the wire model, and the URL, `anthropic-version`, client identity and key placement +the Anthropic adapter uses (`resolveAnthropicMessagesUrl`, `anthropicBaseRequestHeaders`, +`applyAnthropicKeyAuth`), plus the provider's configured headers. No caller header is read, so the +caller's `Authorization`, `x-api-key` and `anthropic-beta` never reach the provider; a beta +allowlist is PF-10. A dropped field has no name in the feature vocabulary, so it records no +feature effect. + +`handleNativeMessages` mirrors native Chat on the shared pieces: `beginInferenceAttempt`, +`createFinalRequestLog`, the request spend tracker charged per physical send, proactive key +selection, 401 and 429 key-pool rotation, same-target 429 replay, the reset/transient retry +policy and `sendWithConnectionPolicy`. Before sending it runs the image normalizer, the image +guard and the tool-call-id repair the caller-forward passthrough runs. A streaming caller gets the +upstream SSE relayed byte for byte through `tapAnthropicSseForLog`, which records usage and the +terminal and applies the body stall and size guards; a non-streaming caller gets the upstream JSON +(or a folded stream). Upstream errors answer in Anthropic shape with the translated lane's status +policy (transient 5xx as 529, replay refusals kept non-retryable). `count_tokens` estimates the +body the builder would send when the route is eligible, and sends nothing. +`tests/adapters/anthropic/anthropic-messages-passthrough.test.ts`, +`tests/responses/messages-native-eligibility.test.ts` and +`tests/claude-integration/messages-native.test.ts` pin the builder, the rule and the lane. + ## Settings `resolveApiSurfaceSettings` and `resolveProtocolSettings` in `src/protocols/settings.ts` are the @@ -207,9 +255,11 @@ always served. The Messages surface uses an explicit `apiSurfaces.messages.enabl present, closes when that value is present but malformed, and otherwise inherits `claudeCode.enabled !== false`. The unrepresentable policy defaults to `legacy` and every `protocols.rollout` switch defaults off; the OAuth native-Messages switch is effective only with -the key-auth one. The Chat and Messages ingresses read the unrepresentable policy (above), and -`directEncodersApply` reads `directEncoders` on both, and the Chat ingress reads `nativeChatCombos` for -combo routes (above); no request path reads the other rollout switches yet. +the key-auth one. The Chat and Messages ingresses read the unrepresentable policy (above); +`directEncodersApply` reads `directEncoders` on both; the Chat ingress reads `nativeChatCombos` +for combo routes (above); `managedMessagesNative` is read through `nativeMessagesDeclineReason` +by the Messages ingress, `count_tokens` and the planner (below). No request path reads the +other rollout switches yet. `claudeInboundDisabled` in `src/server/claude-messages.ts` is the Messages ingress reader: both `/v1/messages` and `/v1/messages/count_tokens` call it, so the two routes cannot disagree, and a diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 56df18ece65..0d26c402157 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -251,7 +251,8 @@ pick after the pin would leave the first attempt on the cooled key. Native Chat Completions is a separate entry path: `src/server/chat-completions.ts` routes eligible `openai-chat` requests to `src/server/chat-native.ts` and never through Responses -core, so that file repeats the same call before it binds the adapter. Native compact +core, so that file repeats the same call before it binds the adapter; the managed native +Messages lane (`src/server/messages-native.ts`) does the same. Native compact (`src/server/responses/compact.ts`) and the keyed Images relay (`src/server/images.ts`) do the same for the same reason. Request paths assign the Transport variant, not the bare `selectProactiveApiKey` snapshot: the snapshot is the persisted row, so it carries none of the @@ -467,7 +468,7 @@ reachable from `core.ts`. | Module | Contract | | --- | --- | | `context.ts` | `createInferenceSendBudget(req, logCtx)` is the one construction of an ingress-owned send holder: the default guarded policy with this request's spend tracker as observer. `handleResponses` calls it only when no holder was inherited, because attaching the tracker parks it on `logCtx`. | -| `final-log.ts` | `createFinalRequestLog(logIds, logCtx)` owns one request's final row: the first `finish(status, meta)` writes it, every later call is a no-op, and without log ids the claim settles with nothing written. The bridged Chat and Messages ingresses and native Chat finish through it. | +| `final-log.ts` | `createFinalRequestLog(logIds, logCtx)` owns one request's final row: the first `finish(status, meta)` writes it, every later call is a no-op, and without log ids the claim settles with nothing written. The bridged Chat and Messages ingresses, native Chat and native Messages finish through it. | | `attempt.ts` | `beginInferenceAttempt(logCtx, { provider, model, adapter })` opens the next attempt ordinal, makes it the active attempt with its start time, appends it to the request, and returns `seal(accountLabel?)` and `finish(status, usage?)`. | | `client-wire.ts` | `markClientWire(response, protocol)` / `clientWireOf(response)` record, per `Response` identity, that a body is already in a client's wire. `createClientWireLog` / `attachClientWireLog` / `clientWireLogOf` carry the request-log facts of such a body (a start payload, one terminal, a cancel), buffered until the deferred log subscribes. The combo marks a native Chat child's answer and its refusal of an all-unrepresentable combo as `chat` (below). | | `client-wire-log.ts` | `recordClientWireRequestLog` is the deferred request log of a client-wire response: `responseWithDeferredRequestLog` calls it instead of tapping the body, and it applies the Responses SSE tap's rules to the reported facts (payload inspection until the terminal, `terminal_sse` phase, `httpStatusForRequestLogTerminal`, 499 for a cancel before any terminal, one row). | @@ -567,6 +568,8 @@ in the Chat shape, blocked trace) with no send. `n > 1` is never emulated with s inferences. Policy routes resolve one candidate in `routeModel` and never reach this loop, so they are not migrated. `tests/responses/chat-native-combo.test.ts` pins the source body, failover within the shared budget, no resend after output, the switch-off path and both reject outcomes. +Managed native Messages ([Protocol Paths](../data-planes/protocol-paths.md#managed-native-messages)) +reuses `beginInferenceAttempt` and `createFinalRequestLog` with its own 401/429 key-rotation loop. ## Adapter-to-Responses bridge From e59e396461963863e794c01c91b029ec68bb22b0 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:03:00 +0900 Subject: [PATCH 11/20] feat(protocols): name operator policy that only the bridge applies A native lane that would skip a pinned effort, skill elision or a sidecar declines with bridge-only-policy. New reason code, so the contract version moves to 2026-09-25.1. --- src/protocols/contract.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/protocols/contract.ts b/src/protocols/contract.ts index 19c501fada7..375f72d5f55 100644 --- a/src/protocols/contract.ts +++ b/src/protocols/contract.ts @@ -9,7 +9,7 @@ */ /** Bumped when a reason code, hop, mode, or feature disposition changes meaning. */ -export const PROTOCOL_CONTRACT_VERSION = "2026-09-24.1"; +export const PROTOCOL_CONTRACT_VERSION = "2026-09-25.1"; export const PROTOCOLS = ["responses", "chat", "messages"] as const; /** A public inference API a client can speak to this proxy. */ @@ -72,6 +72,8 @@ export const PROTOCOL_REASON_CODES = [ "unknown-model", "upstream-other", "rollout-disabled", + /** Operator policy that only the bridge applies (pinned effort, skill elision, a sidecar). */ + "bridge-only-policy", ] as const; export type ProtocolReasonCode = (typeof PROTOCOL_REASON_CODES)[number]; From 5f6d60de619ee4c50a09d99a4259b59ff181a792 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:03:14 +0900 Subject: [PATCH 12/20] feat(claude): tell whether translation would elide a blocked skill A pure predicate over the same inputs the translator uses, so a caller can keep a request on the path that applies claudeCode.blockedSkills. --- src/claude/inbound.ts | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/src/claude/inbound.ts b/src/claude/inbound.ts index 547a8a1c9b7..318db951b6e 100644 --- a/src/claude/inbound.ts +++ b/src/claude/inbound.ts @@ -177,6 +177,28 @@ function blockedSkillCallIds(messages: readonly unknown[], blocked: readonly str return ids; } +/** + * Whether translating this Messages body would elide a blocked skill bundle: a user text block + * `maybeElideSkillText` would stub, or a tool_result answering a blocked Skill call. Pure; the + * managed native Messages lane asks it so a request whose bundle the operator blocked keeps the + * translated path that applies the block. + */ +export function anthropicBodyElidesBlockedSkill(body: unknown, cc?: Pick): boolean { + if (!isRec(body) || !Array.isArray(body.messages)) return false; + const names = effectiveBlockedSkillNames(cc); + if (names.length === 0) return false; + const callIds = blockedSkillCallIds(body.messages, names); + for (const msg of body.messages) { + if (!isRec(msg) || msg.role !== "user" || !Array.isArray(msg.content)) continue; + for (const block of msg.content) { + if (!isRec(block)) continue; + if (block.type === "text" && typeof block.text === "string" && maybeElideSkillText(block.text, names) !== block.text) return true; + if (block.type === "tool_result" && typeof block.tool_use_id === "string" && callIds.has(block.tool_use_id)) return true; + } + } + return false; +} + /** * Claude Code (observed 2026-07-11, real CLI smoke) sends `role:"system"` entries in * `messages` despite the published API having no system role. They are emitted as From 822a5a1708996c50c1b01f1eca66f39c507e3866 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:03:40 +0900 Subject: [PATCH 13/20] feat(claude): keep Messages on the bridge when bridge-only policy applies A pinned route effort, a blocked-skill bundle the translator would elide, or a web-search tool the sidecar could serve now declines the native lane with bridge-only-policy. --- src/server/messages-native-eligibility.ts | 59 ++++++++++++++++++++++- 1 file changed, 57 insertions(+), 2 deletions(-) diff --git a/src/server/messages-native-eligibility.ts b/src/server/messages-native-eligibility.ts index 1b1742b30a4..91ba350f203 100644 --- a/src/server/messages-native-eligibility.ts +++ b/src/server/messages-native-eligibility.ts @@ -5,12 +5,15 @@ * `count_tokens` and the protocol planner (`src/protocols/plan-snapshot.ts`) all ask here, so a * preview, a count and the real send cannot disagree about the lane. */ +import { anthropicBodyElidesBlockedSkill } from "../claude/inbound"; +import { isClaudeWebSearchToolName } from "../claude/outbound"; import type { ProtocolReasonCode } from "../protocols/contract"; import { featuresFromMessagesBody } from "../protocols/features"; import { resolveProtocolSettings } from "../protocols/settings"; import type { RouteResult } from "../router"; import type { OcxConfig } from "../types"; import { requiresVisionPreprocessing } from "../vision"; +import { resolvePinnedEffort } from "./effort-policy"; /** Why the managed native Messages lane declines a route, as a protocol reason code. */ export type NativeMessagesDeclineReason = Extract< @@ -22,12 +25,61 @@ export type NativeMessagesDeclineReason = Extract< | "effort-row" | "fast-row" | "vision-preprocessing" + | "bridge-only-policy" >; -/** Model-selector facts the body cannot carry: a synthetic effort or fast row was requested. */ +/** Request facts the body cannot carry. */ export interface NativeMessagesSelector { + /** A synthetic effort row was requested. */ effortRow?: boolean; + /** A synthetic fast row was requested. */ fastRow?: boolean; + /** The model id the bridge would parse (the translated body's `model`), for the effort pin. */ + routeSelector?: string; + /** The Claude settings this ingress reads (intercept bindings applied); defaults to config's. */ + claudeCode?: OcxConfig["claudeCode"]; +} + +type Rec = Record; +function isRec(value: unknown): value is Rec { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +/** + * Whether the bridge could hand this request to the web-search sidecar: a `web_search*` server + * tool the tool choice does not exclude, with the sidecar not disabled in the Claude replay config + * (`buildClaudeReplayConfig` merges `claudeCode.webSearchSidecar` over the global block). Whether + * a backend credential then exists is decided at dispatch and is not read here, so this errs + * toward the bridge, which is today's behavior for every such request. + */ +function webSearchSidecarMayEngage(body: Readonly, config: OcxConfig): boolean { + const sidecar = { ...config.webSearchSidecar, ...config.claudeCode?.webSearchSidecar }; + if (sidecar.enabled === false) return false; + if (!Array.isArray(body.tools)) return false; + const declared = body.tools.some(tool => isRec(tool) && typeof tool.type === "string" && tool.type.startsWith("web_search")); + if (!declared) return false; + const choice = body.tool_choice; + if (isRec(choice)) { + if (choice.type === "none") return false; + if (choice.type === "tool") return typeof choice.name === "string" && isClaudeWebSearchToolName(choice.name); + } + return true; +} + +/** + * Operator policy the translated path applies and the native lane would skip: a pinned + * reasoning effort for the route, a blocked-skill bundle the translator would elide, or the + * web-search sidecar. Any of them keeps the request on the bridge. + */ +function bridgeOnlyPolicyApplies( + route: RouteResult, + body: Readonly, + config: OcxConfig, + selector: NativeMessagesSelector, +): boolean { + if (resolvePinnedEffort(route, selector.routeSelector, config) !== undefined) return true; + if (anthropicBodyElidesBlockedSkill(body, selector.claudeCode ?? config.claudeCode)) return true; + return webSearchSidecarMayEngage(body, config); } /** @@ -39,7 +91,9 @@ export interface NativeMessagesSelector { * - the credential is not a proxy-managed key (OAuth is PF-10; `forward` belongs to the caller); * - a combo or policy route owns multi-candidate execution in the Responses pipeline; * - a synthetic effort or fast row needs the adapter that owns its wire rewrite; - * - an image would reach a model the operator declared unable to read it. + * - an image would reach a model the operator declared unable to read it; + * - operator policy only the bridge applies would engage (pinned effort, blocked-skill elision, + * the web-search sidecar). */ export function nativeMessagesDeclineReason( route: RouteResult, @@ -58,6 +112,7 @@ export function nativeMessagesDeclineReason( && requiresVisionPreprocessing(config, provider, route.modelId, route.providerName)) { return "vision-preprocessing"; } + if (bridgeOnlyPolicyApplies(route, body, config, selector)) return "bridge-only-policy"; return undefined; } From 8bf884a1e53cafaf827808047d34e04227643d2c Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:04:11 +0900 Subject: [PATCH 14/20] feat(claude): judge native Messages with the bridge's selector and Claude settings The ingress, count_tokens and the key-reselection recheck pass the translated model id and the ingress's Claude view, so the effort pin and blocked skills read what the bridge reads. --- src/server/claude-messages.ts | 19 ++++++++++++------- src/server/messages-native.ts | 7 +++++-- 2 files changed, 17 insertions(+), 9 deletions(-) diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index 5760b730192..b4f227b96b9 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -71,7 +71,7 @@ import { checkRepresentable, unrepresentableMessage } from "../protocols/guard"; import { requestPathForLane } from "../protocols/path"; import { resolveApiSurfaceSettings, resolveProtocolSettings } from "../protocols/settings"; import { markProtocolBlocked, markProtocolEntry } from "../protocols/trace"; -import { isNativeMessagesRouteEligible } from "./messages-native-eligibility"; +import { nativeMessagesDeclineReason, type NativeMessagesSelector } from "./messages-native-eligibility"; import { isApiAuthRequired, isDataPlaneAdmissionSecret, @@ -824,9 +824,11 @@ async function handleClaudeMessagesWithBudget( const sourceEnvelope = envelope; // The bridge entry mark below reads these before an effort override rewrites `thinking`, // which also fixes the envelope's cached features on the caller's own settings. + // Scanned once, like the envelope's cache, so a later mark reports the same features. + let scannedFeatures: ReadonlySet | undefined; messagesFeatures = sourceEnvelope ? () => sourceEnvelope.features() - : () => featuresFromMessagesBody(messagesBody); + : () => (scannedFeatures ??= featuresFromMessagesBody(messagesBody)); if (!effortRow && !fastRow && isRec(anthropicBody) && wantsNativePassthrough(req, config, requestPolicy, anthropicBody.model, cc)) { markProtocolEntry(logCtx, { inbound: "messages", lane: "native", features: messagesFeatures }); return await anthropicNativePassthrough(req, config, logCtx, logIds, anthropicBody, "/v1/messages"); @@ -986,10 +988,13 @@ async function handleClaudeMessagesWithBudget( // PF-08: a managed-key Anthropic route sends its Messages body natively. The caller-forward // passthrough above was decided on the caller's own credential and never reaches this point. - const nativeMessagesRoute = settledRoute && isRec(anthropicBody) - && isNativeMessagesRouteEligible(settledRoute, anthropicBody, config, { effortRow: !!effortRow, fastRow: !!fastRow }) - ? settledRoute - : undefined; + const nativeSelector: NativeMessagesSelector = { + effortRow: !!effortRow, fastRow: !!fastRow, routeSelector: String(internalBody.model ?? ""), claudeCode: cc, + }; + const nativeDecline = settledRoute && isRec(anthropicBody) + ? nativeMessagesDeclineReason(settledRoute, anthropicBody, config, nativeSelector) + : "rollout-disabled"; + const nativeMessagesRoute = settledRoute && nativeDecline === undefined ? settledRoute : undefined; // Combo and policy children are judged per candidate (PF-07); an unknown model has no route. if (envelope && settledRoute && !settledRoute.combo && settledRoute.routeKind !== "policy") { const verdict = checkRepresentable({ @@ -1019,7 +1024,7 @@ async function handleClaudeMessagesWithBudget( const { handleNativeMessages } = await import("./messages-native"); return await handleNativeMessages({ req, config, logCtx, ...(logIds ? { logIds } : {}), - route: nativeMessagesRoute, body: nativeBody, requestedModel, translatorBudget, + route: nativeMessagesRoute, body: nativeBody, requestedModel, translatorBudget, selector: nativeSelector, }); } diff --git a/src/server/messages-native.ts b/src/server/messages-native.ts index 3fb7a499b3f..11fc675b587 100644 --- a/src/server/messages-native.ts +++ b/src/server/messages-native.ts @@ -117,6 +117,8 @@ export interface HandleNativeMessagesOptions { /** The selector the client sent, echoed on the request log. */ requestedModel: string; translatorBudget: TranslatorBudget; + /** The facts the ingress judged eligibility with; re-applied if key selection changes. */ + selector?: NativeMessagesSelector; } type FinishLog = (status: number, message?: string, closeReason?: FinalRequestLogMeta["closeReason"]) => void; @@ -299,7 +301,7 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions) ? Math.max(0, requestTransientPolicy.attempts - transientSendsUsed) : Number.POSITIVE_INFINITY; const transientSendAvailable = (): boolean => remainingTransientSends() > 0; - const selector: NativeMessagesSelector = {}; + const selector: NativeMessagesSelector = options.selector ?? {}; const send = async (recovery?: "rate-limit-429" | "key-429" | "key-401"): Promise => { const remaining = remainingTransientSends(); @@ -596,7 +598,7 @@ export function nativeMessagesCountBody( config: OcxConfig, cc: OcxConfig["claudeCode"], body: Rec, - selector: NativeMessagesSelector, + rows: Pick, ): Rec | undefined { if (typeof body.model !== "string") return undefined; try { @@ -610,6 +612,7 @@ export function nativeMessagesCountBody( route.providerName, route.modelId, route.provider, route.staticPolicy.effectiveAlias, "anthropic", ); route.provider = resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, "anthropic", route.staticPolicy); + const selector: NativeMessagesSelector = { ...rows, routeSelector: selectorId, claudeCode: cc }; if (nativeMessagesDeclineReason(route, body, config, selector) !== undefined) return undefined; return buildAnthropicMessagesPassthroughRequest(route.provider, route.modelId, body, config).wireBody; } catch { From e86399360985f32a5964b9bc84105b6b8534d472 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:04:13 +0900 Subject: [PATCH 15/20] feat(claude): record why a Messages route declined the native lane With the switch on, the bridge entry mark carries the decline reason, so the trace explains the bridge; features are scanned once so both marks report the caller's own settings. --- src/server/claude-messages.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index b4f227b96b9..c9b58a77fd2 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -995,6 +995,15 @@ async function handleClaudeMessagesWithBudget( ? nativeMessagesDeclineReason(settledRoute, anthropicBody, config, nativeSelector) : "rollout-disabled"; const nativeMessagesRoute = settledRoute && nativeDecline === undefined ? settledRoute : undefined; + // With the switch on, a declined route says why on its bridge mark, as native Chat does. + if (nativeDecline && nativeDecline !== "rollout-disabled") { + markProtocolEntry(logCtx, { + inbound: "messages", + lane: "bridge", + reasonCodes: [...(effortRow ? ["effort-row" as const] : fastRow ? ["fast-row" as const] : []), nativeDecline], + features: messagesFeatures, + }); + } // Combo and policy children are judged per candidate (PF-07); an unknown model has no route. if (envelope && settledRoute && !settledRoute.combo && settledRoute.routeKind !== "policy") { const verdict = checkRepresentable({ From 9706f83081b8710f9ec490aed306cf81ea8681e6 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:04:30 +0900 Subject: [PATCH 16/20] feat(protocols): preview the bridge-only decline for Messages candidates The planner passes the resolved selector, so a pinned route effort reports bridge-only-policy; body-dependent policy (skills, web search) is not predictable from features. --- src/protocols/plan-snapshot.ts | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/src/protocols/plan-snapshot.ts b/src/protocols/plan-snapshot.ts index e7a9684648f..aaadb435e88 100644 --- a/src/protocols/plan-snapshot.ts +++ b/src/protocols/plan-snapshot.ts @@ -102,6 +102,8 @@ function messagesBodyForFeatures(features: ReadonlySet): Record interface SyntheticRows { effortRow: boolean; fastRow: boolean; + /** The resolved selector the bridge would parse; the effort pin reads it. */ + routeSelector: string; } function candidateFor( @@ -139,7 +141,12 @@ function candidateFor( } else if (inbound === "messages" && resolveProtocolSettings(config).rollout.managedMessagesNative) { // The runtime rule itself. With the switch off nothing is judged, so the default preview // is exactly what it was before the managed native lane existed. - const reason = nativeMessagesDeclineReason(settled, messagesBodyForFeatures(features), config, rows); + // Pinned effort is judged from config and the route; blocked-skill elision and the web-search + // sidecar depend on body content no feature describes, so a preview cannot predict them. + const reason = nativeMessagesDeclineReason(settled, messagesBodyForFeatures(features), config, { + ...rows, + claudeCode: config.claudeCode, + }); nativeEligible = reason === undefined; if (reason) declineReasons = [reason]; } @@ -205,7 +212,7 @@ export function buildProtocolPlanSnapshot( return { ...base, routeKind: settled.routeKind, - candidates: settled.routes.map(route => candidateFor(config, request.inbound, route, settled.routeKind, features, { effortRow, fastRow })), + candidates: settled.routes.map(route => candidateFor(config, request.inbound, route, settled.routeKind, features, { effortRow, fastRow, routeSelector: routeKey })), reasonCodes, }; } From 776e9a65596d5ce6169797e3be5f0de50db939a9 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:05:01 +0900 Subject: [PATCH 17/20] feat(claude): echo the requested model on native Messages responses The translated lane answers with the client's selector; the native lane now rewrites message_start (stream) or the message (JSON) the same way instead of the wire id. --- src/server/messages-native.ts | 85 ++++++++++++++++++++++++++++++++++- 1 file changed, 83 insertions(+), 2 deletions(-) diff --git a/src/server/messages-native.ts b/src/server/messages-native.ts index 11fc675b587..51a1ce3550b 100644 --- a/src/server/messages-native.ts +++ b/src/server/messages-native.ts @@ -84,6 +84,7 @@ import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, sendWithConnectio import { linkAbortSignal } from "./responses/core-lifetime"; import { attachRequestSpendTracker } from "./responses/request-spend"; import { workflowRefusalResponse } from "./workflow-refusal"; +import { sseFieldValue } from "../lib/sse-decoder"; export { isNativeMessagesRouteEligible, @@ -146,6 +147,82 @@ function observeFirstChunk(body: ReadableStream, onFirst: () => void }); } +/** How far into a stream the `message_start` frame is looked for before relaying untouched. */ +const MODEL_ECHO_SCAN_BYTES = 64 * 1024; + +/** The `message_start` frame with `message.model` set to `model`, or undefined for any other frame. */ +function messageStartWithModel(frame: string, model: string): string | undefined { + const data = frame + .split("\n") + .map(line => sseFieldValue(line, "data")) + .filter((value): value is string => value !== null) + .join(""); + if (!data) return undefined; + let parsed: unknown; + try { parsed = JSON.parse(data); } catch { return undefined; } + if (!isRec(parsed) || parsed.type !== "message_start" || !isRec(parsed.message)) return undefined; + parsed.message.model = model; + return `event: message_start\ndata: ${JSON.stringify(parsed)}\n\n`; +} + +/** + * Echo the client's selector in `message_start.message.model`, as the translated lane does + * (`responsesSseToAnthropicSse` is given the requested model). Works on bytes: frames before and + * including `message_start` are split on the blank line, and everything after is relayed as is. + */ +function echoRequestedModel(body: ReadableStream, model: string): ReadableStream { + const reader = body.getReader(); + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + let pending = new Uint8Array(0); + let scanning = true; + let scanned = 0; + const frameEnd = (bytes: Uint8Array): number => { + for (let i = 0; i + 1 < bytes.length; i++) if (bytes[i] === 10 && bytes[i + 1] === 10) return i + 2; + return -1; + }; + return new ReadableStream({ + async pull(controller) { + const { done, value } = await reader.read(); + if (done) { + if (pending.length > 0) controller.enqueue(pending); + pending = new Uint8Array(0); + controller.close(); + return; + } + if (!scanning) { + controller.enqueue(value); + return; + } + scanned += value.byteLength; + const joined = new Uint8Array(pending.length + value.byteLength); + joined.set(pending); + joined.set(value, pending.length); + pending = joined; + let end: number; + while (scanning && (end = frameEnd(pending)) !== -1) { + const frame = pending.subarray(0, end); + pending = pending.subarray(end); + const rewritten = messageStartWithModel(decoder.decode(frame), model); + if (rewritten !== undefined) { + controller.enqueue(encoder.encode(rewritten)); + scanning = false; + } else { + controller.enqueue(frame); + } + } + if (scanning && scanned > MODEL_ECHO_SCAN_BYTES) scanning = false; + if (!scanning && pending.length > 0) { + controller.enqueue(pending); + pending = new Uint8Array(0); + } + }, + cancel(reason) { + return reader.cancel(reason); + }, + }); +} + /** A minimal valid Messages stream for a streaming caller whose upstream answered JSON. */ function messageAsSse(message: Rec): string { const frames: string[] = []; @@ -445,7 +522,8 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions) const contentType = response.headers.get("content-type")?.toLowerCase() ?? ""; if (contentType.includes("text/event-stream") && response.body) { const bodyGuard = resolvePassthroughBodyGuard(config, req.signal); - const source = logIds ? observeFirstChunk(response.body, () => recordFirstOutput(logCtx, logIds.start)) : response.body; + const observed = logIds ? observeFirstChunk(response.body, () => recordFirstOutput(logCtx, logIds.start)) : response.body; + const source = echoRequestedModel(observed, requestedModel); if (requestedStream) { transferTurnToStream(); const relayed = tapAnthropicSseForLog(source, logCtx, (status, meta) => { @@ -530,6 +608,9 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions) } throw error; } + // The client's selector, not the wire id, as the translated lane answers. + message.model = requestedModel; + const serialized = JSON.stringify(message); finishLog(200); if (requestedStream) { return new Response(messageAsSse(message), { @@ -537,7 +618,7 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions) headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache" }, }); } - return new Response(read.text, { status: 200, headers: { "Content-Type": "application/json" } }); + return new Response(serialized, { status: 200, headers: { "Content-Type": "application/json" } }); } /** From f4f943ff020ebe7f70901bf3c0a1c40147ce16c6 Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:06:05 +0900 Subject: [PATCH 18/20] test(claude): expect the requested model on native Messages responses JSON and stream answers now carry the client's selector, matching the translated lane. --- tests/claude-integration/messages-native.test.ts | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/tests/claude-integration/messages-native.test.ts b/tests/claude-integration/messages-native.test.ts index 24eb3551070..37b19deddaf 100644 --- a/tests/claude-integration/messages-native.test.ts +++ b/tests/claude-integration/messages-native.test.ts @@ -74,7 +74,12 @@ const SSE_FRAMES = [ { event: "message_delta", data: { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 7 } } }, { event: "message_stop", data: { type: "message_stop" } }, ]; -const SSE_TEXT = SSE_FRAMES.map(frame => `event: ${frame.event}\ndata: ${JSON.stringify(frame.data)}\n\n`).join(""); +const sseText = (frames: readonly { event: string; data: unknown }[]) => frames.map(frame => `event: ${frame.event}\ndata: ${JSON.stringify(frame.data)}\n\n`).join(""); +const SSE_TEXT = sseText(SSE_FRAMES); +/** What the client receives: the upstream stream with its selector echoed in message_start. */ +const ECHOED_SSE_TEXT = sseText(SSE_FRAMES.map((frame, index) => index === 0 + ? { ...frame, data: { ...frame.data, message: { ...(frame.data as { message: Record }).message, model: "anth/claude-x" } } } + : frame)); function ok(seenRequest: Seen): Response { if (seenRequest.body.stream === true) { @@ -162,7 +167,8 @@ describe("managed native Messages", () => { "anthropic-beta": "fixture-beta", }); expect(response.status).toBe(200); - expect(JSON.parse(text)).toEqual(MESSAGE_JSON); + // The client's selector is echoed, as on the translated lane. + expect(JSON.parse(text)).toEqual({ ...MESSAGE_JSON, model: "anth/claude-x" }); expect(seen).toHaveLength(1); const sent = seen[0]!; expect(sent.path).toBe("/v1/messages"); @@ -180,12 +186,12 @@ describe("managed native Messages", () => { expect(JSON.stringify(row)).not.toContain("fixture-key-alpha"); }); - test("relays the upstream stream unchanged and records its usage", async () => { + test("relays the upstream stream (selector echoed in message_start) and records its usage", async () => { const config = fixtureConfig(startUpstream()); const { requestId, response, text } = await send(config, { ...SOURCE_BODY, stream: true }); expect(response.status).toBe(200); expect(response.headers.get("content-type")).toContain("text/event-stream"); - expect(text).toBe(SSE_TEXT); + expect(text).toBe(ECHOED_SSE_TEXT); expect(seen[0]!.body.stream).toBe(true); const row = rowFor(requestId); expect(row.status).toBe(200); From b6fc3300265f836160379af437bf009ea6e0ba3e Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:06:05 +0900 Subject: [PATCH 19/20] test(claude): pin bridge-only-policy declines and the decline trace Pinned effort, blocked-skill elision and the web-search sidecar decline the native lane; the planner reports the pin; a declined route's bridge trace names the reason only when on. --- scripts/test-layout/layout.json | 2 + .../messages-native-decline-trace.test.ts | 102 +++++++++++++++ tests/fixtures/test-layout-expected.json | 2 + .../messages-native-bridge-policy.test.ts | 122 ++++++++++++++++++ 4 files changed, 228 insertions(+) create mode 100644 tests/claude-integration/messages-native-decline-trace.test.ts create mode 100644 tests/responses/messages-native-bridge-policy.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index d246747b47d..b73b5f373c6 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -345,6 +345,7 @@ "protocol-direct-encoders-messages.test.ts": "responses", "chat-native-combo.test.ts": "responses", "messages-native-eligibility.test.ts": "responses", + "messages-native-bridge-policy.test.ts": "responses", "chat-inbound-reasoning-none.test.ts": "responses", "chat-native-decline-reason.test.ts": "responses", "chat-inbound-reasoning-replay.test.ts": "responses", @@ -422,6 +423,7 @@ "claude-manual-env.test.ts": "gui", "claude-messages-endpoint.test.ts": "claude-integration", "messages-native.test.ts": "claude-integration", + "messages-native-decline-trace.test.ts": "claude-integration", "messages-surface-matrix.test.ts": "claude-integration", "claude-model-info.test.ts": "claude-integration", "claude-models-discovery.test.ts": "claude-integration", diff --git a/tests/claude-integration/messages-native-decline-trace.test.ts b/tests/claude-integration/messages-native-decline-trace.test.ts new file mode 100644 index 00000000000..0a439614182 --- /dev/null +++ b/tests/claude-integration/messages-native-decline-trace.test.ts @@ -0,0 +1,102 @@ +/** + * With `protocols.rollout.managedMessagesNative` on, a Messages route that declines the native + * lane stays on the bridge and its trace names the rule that declined it. With the switch off + * the trace is unchanged: no decline reason is recorded. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { saveConfig } from "../../src/config"; +import { handleClaudeMessages } from "../../src/server/claude-messages"; +import { getRequestLogEntries } from "../../src/server/request-log"; +import type { OcxConfig } from "../../src/types"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +let upstream: ReturnType | undefined; +let bodies: Record[] = []; +let releaseSpendHome: (() => void) | undefined; +let testDir = ""; +let previousHome: string | undefined; + +const SSE = [ + { type: "message_start", message: { id: "msg_fixture", type: "message", role: "assistant", model: "claude-x", content: [], + stop_reason: null, stop_sequence: null, usage: { input_tokens: 3, output_tokens: 0 } } }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "fixture" } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, +].map(data => `event: ${data.type}\ndata: ${JSON.stringify(data)}\n\n`).join(""); + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + testDir = mkdtempSync(join(tmpdir(), "ocx-messages-decline-trace-")); + process.env.OPENCODEX_HOME = testDir; + bodies = []; + upstream = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(req) { + bodies.push(await req.json() as Record); + return new Response(SSE, { headers: { "content-type": "text/event-stream" } }); + } }); +}); + +afterEach(async () => { + releaseSpendHome?.(); + releaseSpendHome = undefined; + await upstream?.stop(true); + upstream = undefined; + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (testDir) removeTreeWithRetry(testDir); +}); + +function fixtureConfig(on: boolean): OcxConfig { + releaseSpendHome ??= acquireOwnedSpendHome(); + const config = { + port: 0, + defaultProvider: "anth", + providers: { anth: { + adapter: "anthropic", baseUrl: `http://127.0.0.1:${upstream!.port}`, authMode: "key", apiKey: "fixture-key", + allowPrivateNetwork: true, models: ["claude-x"], + // Operator policy only the bridge applies. + pinnedReasoningEffort: "high", + } }, + ...(on ? { protocols: { rollout: { managedMessagesNative: true } } } : {}), + } as OcxConfig; + saveConfig(config); + return config; +} + +async function send(config: OcxConfig) { + const requestId = `pf08-decline-${crypto.randomUUID()}`; + const response = await handleClaudeMessages(new Request("http://localhost/v1/messages", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "anth/claude-x", max_tokens: 32, top_k: 4, stream: false, + messages: [{ role: "user", content: "fixture" }] }), + }), config, { model: "", provider: "" }, { requestId, start: Date.now() }); + await response.text(); + const rows = getRequestLogEntries().filter(entry => entry.requestId === requestId); + expect(rows).toHaveLength(1); + return { response, row: rows[0]! }; +} + +describe("native Messages decline in the trace", () => { + test("switch on: a pinned effort keeps the bridge and the trace says why", async () => { + const { response, row } = await send(fixtureConfig(true)); + expect(response.status).toBe(200); + expect(bodies).toHaveLength(1); + // The bridge rebuilt the request through the adapter: no top_k, internal streaming. + expect(bodies[0]).not.toHaveProperty("top_k"); + expect(row.protocolTrace).toMatchObject({ inbound: "messages", mode: "legacy-bridge" }); + expect(row.protocolTrace?.reasonCodes).toContain("bridge-only-policy"); + }); + + test("switch off: no decline reason is recorded", async () => { + const { row } = await send(fixtureConfig(false)); + expect(row.protocolTrace).toMatchObject({ inbound: "messages", mode: "legacy-bridge" }); + expect(row.protocolTrace?.reasonCodes).not.toContain("bridge-only-policy"); + expect(row.protocolTrace?.reasonCodes).not.toContain("rollout-disabled"); + }); +}); diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 9a702263cc3..ad337cdd2ad 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -171,6 +171,7 @@ "protocol-direct-encoders-messages.test.ts": "responses", "chat-native-combo.test.ts": "responses", "messages-native-eligibility.test.ts": "responses", + "messages-native-bridge-policy.test.ts": "responses", "chat-inbound-reasoning-none.test.ts": "responses", "chat-native-decline-reason.test.ts": "responses", "chat-inbound-reasoning-replay.test.ts": "responses", @@ -248,6 +249,7 @@ "claude-manual-env.test.ts": "gui", "claude-messages-endpoint.test.ts": "claude-integration", "messages-native.test.ts": "claude-integration", + "messages-native-decline-trace.test.ts": "claude-integration", "messages-surface-matrix.test.ts": "claude-integration", "claude-model-info.test.ts": "claude-integration", "claude-models-discovery.test.ts": "claude-integration", diff --git a/tests/responses/messages-native-bridge-policy.test.ts b/tests/responses/messages-native-bridge-policy.test.ts new file mode 100644 index 00000000000..61ab7c4815b --- /dev/null +++ b/tests/responses/messages-native-bridge-policy.test.ts @@ -0,0 +1,122 @@ +/** + * Operator policy only the bridge applies keeps a Messages request off the managed native lane + * (`bridge-only-policy`): a pinned route effort, a blocked-skill bundle the translator would + * elide, and a web-search tool the sidecar could serve. The planner reports the config-only part + * of the same rule. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { anthropicBodyElidesBlockedSkill } from "../../src/claude/inbound"; +import { buildProtocolPlanSnapshot } from "../../src/protocols/plan-snapshot"; +import type { RouteResult } from "../../src/router"; +import { nativeMessagesDeclineReason } from "../../src/server/messages-native-eligibility"; +import type { OcxConfig } from "../../src/types"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; + +let testDir = ""; +let previousHome: string | undefined; + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + testDir = mkdtempSync(join(tmpdir(), "ocx-messages-bridge-policy-")); + process.env.OPENCODEX_HOME = testDir; +}); + +afterEach(() => { + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (testDir) removeTreeWithRetry(testDir); +}); + +function config(overrides: Partial = {}, provider: Record = {}): OcxConfig { + return { + port: 10100, + defaultProvider: "anth", + providers: { + anth: { adapter: "anthropic", baseUrl: "https://anth.example/v1", authMode: "key", apiKey: "ka", models: ["claude-x"], ...provider }, + }, + protocols: { rollout: { managedMessagesNative: true } }, + ...overrides, + } as OcxConfig; +} + +function route(provider: Record = {}): RouteResult { + return { + providerName: "anth", + modelId: "claude-x", + routeKind: "direct", + routeReason: "test", + provider: { adapter: "anthropic", baseUrl: "https://anth.example/v1", apiKey: "ka", ...provider }, + } as unknown as RouteResult; +} + +const TEXT_BODY = { messages: [{ role: "user", content: "fixture" }] }; + +const SKILL_BUNDLE = `Base directory for this skill: /fixture/skills/claude-api\n${"x".repeat(12_000)}`; + +describe("pinned route effort", () => { + test("a provider or model pin keeps the request on the bridge", () => { + expect(nativeMessagesDeclineReason(route({ pinnedReasoningEffort: "high" }), TEXT_BODY, config())) + .toBe("bridge-only-policy"); + expect(nativeMessagesDeclineReason(route({ modelPinnedReasoningEfforts: { "claude-x": "low" } }), TEXT_BODY, config())) + .toBe("bridge-only-policy"); + }); + + test("a global pin is read through the bridge's selector", () => { + const pinned = config({ modelPinnedEfforts: { "anth/claude-x": "medium" } } as Partial); + expect(nativeMessagesDeclineReason(route(), TEXT_BODY, pinned, { routeSelector: "anth/claude-x" })) + .toBe("bridge-only-policy"); + expect(nativeMessagesDeclineReason(route(), TEXT_BODY, config(), { routeSelector: "anth/claude-x" })).toBeUndefined(); + }); + + test("the planner reports the pin as the decline reason", () => { + const snapshot = buildProtocolPlanSnapshot(config({}, { pinnedReasoningEffort: "high" }), + { model: "anth/claude-x", inbound: "messages", features: [] }); + expect(snapshot.candidates[0]).toMatchObject({ nativeEligible: false, declineReasons: ["bridge-only-policy"] }); + }); +}); + +describe("blocked-skill elision", () => { + const bundleBody = { messages: [{ role: "user", content: [{ type: "text", text: SKILL_BUNDLE }] }] }; + const resultBody = { + messages: [ + { role: "assistant", content: [{ type: "tool_use", id: "toolu_skill", name: "Skill", input: { skill: "claude-api" } }] }, + { role: "user", content: [{ type: "tool_result", tool_use_id: "toolu_skill", content: "Launching skill" }] }, + ], + }; + + test("a bundle or a blocked Skill result the translator would stub declines the native lane", () => { + expect(anthropicBodyElidesBlockedSkill(bundleBody)).toBe(true); + expect(anthropicBodyElidesBlockedSkill(resultBody)).toBe(true); + expect(nativeMessagesDeclineReason(route(), bundleBody, config())).toBe("bridge-only-policy"); + expect(nativeMessagesDeclineReason(route(), resultBody, config())).toBe("bridge-only-policy"); + }); + + test("nothing to elide, or no blocked skills, stays native", () => { + expect(anthropicBodyElidesBlockedSkill(TEXT_BODY)).toBe(false); + const short = { messages: [{ role: "user", content: [{ type: "text", text: "Base directory for this skill: /x/claude-api" }] }] }; + expect(anthropicBodyElidesBlockedSkill(short)).toBe(false); + const unblocked = config({ claudeCode: { blockedSkills: [] } } as Partial); + expect(nativeMessagesDeclineReason(route(), bundleBody, unblocked)).toBeUndefined(); + }); +}); + +describe("web-search sidecar", () => { + const searchBody = { ...TEXT_BODY, tools: [{ type: "web_search_20250305", name: "web_search" }] }; + + test("a web_search server tool the sidecar could serve declines the native lane", () => { + expect(nativeMessagesDeclineReason(route(), searchBody, config())).toBe("bridge-only-policy"); + expect(nativeMessagesDeclineReason(route(), { ...searchBody, tool_choice: { type: "tool", name: "web_search" } }, config())) + .toBe("bridge-only-policy"); + }); + + test("a disabled sidecar or a tool choice that excludes search stays native", () => { + const off = config({ claudeCode: { webSearchSidecar: { enabled: false } } } as Partial); + expect(nativeMessagesDeclineReason(route(), searchBody, off)).toBeUndefined(); + expect(nativeMessagesDeclineReason(route(), { ...searchBody, tool_choice: { type: "none" } }, config())).toBeUndefined(); + expect(nativeMessagesDeclineReason(route(), { ...searchBody, tool_choice: { type: "tool", name: "lookup" } }, config())) + .toBeUndefined(); + }); +}); From 53e478328ba66580ecadd64b859a678fad13bdcb Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 25 Sep 2026 13:06:28 +0900 Subject: [PATCH 20/20] docs(structure): record bridge-only declines, the decline trace and model echo States which operator policy keeps Messages on the bridge, that stabilizePromptCache is a recorded gap rather than a decline rule, and that native answers echo the client's selector. --- .../040_acceptance_and_rollout.md | 4 +-- structure/data-planes/protocol-paths.md | 28 +++++++++++++++---- 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/devlog/_plan/260924_protocol_first_class/040_acceptance_and_rollout.md b/devlog/_plan/260924_protocol_first_class/040_acceptance_and_rollout.md index c0f2c82648e..d234573565b 100644 --- a/devlog/_plan/260924_protocol_first_class/040_acceptance_and_rollout.md +++ b/devlog/_plan/260924_protocol_first_class/040_acceptance_and_rollout.md @@ -48,6 +48,6 @@ Kept current by each packet that migrates something. | Responses-only features on Chat/Messages | `previous_response_id`, `store`, `background`, compaction stay on the bridge | | Non-public-wire adapters (`other`) | translated through the IR; no feature claims | | OAuth native Chat | not planned in this unit | -| Messages → key-auth Anthropic | native behind `managedMessagesNative` (PF-08); bridge while off. Caller `anthropic-beta` is not forwarded (PF-10 allowlist); top-level fields outside the allowlist are dropped with no feature effect; the response `model` is the upstream's wire id | +| Messages → key-auth Anthropic | native behind `managedMessagesNative` (PF-08); bridge while off. Caller `anthropic-beta` is not forwarded (PF-10 allowlist); top-level fields outside the allowlist are dropped with no feature effect | | Messages → Anthropic OAuth | bridge until `managedMessagesNativeOAuth` (PF-10) | -| Messages native lane, translated-only steps | skill elision, `stabilizePromptCache`, pinned route effort and the Claude web-search/vision sidecars run only on the bridge; routes needing vision preprocessing stay there | +| Messages native lane, translated-only steps | a pinned route effort, blocked-skill elision, the web-search sidecar and vision preprocessing keep the request on the bridge (`bridge-only-policy` / `vision-preprocessing`); `stabilizePromptCache` is a recorded gap — the native lane does not apply it | diff --git a/structure/data-planes/protocol-paths.md b/structure/data-planes/protocol-paths.md index a0e04c500e8..37d88cbec73 100644 --- a/structure/data-planes/protocol-paths.md +++ b/structure/data-planes/protocol-paths.md @@ -214,7 +214,20 @@ Responses. `nativeMessagesDeclineReason` names the first rule that keeps a route `rollout-disabled`, `cross-wire-ir` (another adapter), `auth-mode-not-native` (OAuth, which is PF-10, or `forward`), `combo-or-policy-route`, `effort-row` / `fast-row` (synthetic rows need the adapter's wire rewrite), `vision-preprocessing` (an image for a model declared unable to read -it). The ingress, `count_tokens` and the planner all ask it. +it), and `bridge-only-policy` when operator policy that only the translated path applies would +engage: a pinned reasoning effort for the route (`resolvePinnedEffort`, read with the translated +body's model id as the bridge reads it), a blocked-skill bundle the translator would elide +(`anthropicBodyElidesBlockedSkill` in `src/claude/inbound.ts`), or a `web_search*` server tool the +web-search sidecar could serve (not excluded by `tool_choice`, sidecar not disabled in the Claude +replay config; backend credentials are decided at dispatch, so this errs toward the bridge). The +ingress, `count_tokens` and the planner all ask it. The planner can judge only the config-and-route +parts: skill elision and web search depend on body content no feature describes. + +With the switch on, a declined route re-marks its bridge entry with the decline reason (after +any `effort-row` / `fast-row`), so the trace says why the bridge was taken; with it off nothing is +added. `claudeCode.stabilizePromptCache` is not a decline rule: it is a Claude-app cache +optimization rather than routing policy, applies only on the translated path, and is a recorded +gap of the native lane. `src/server/claude-messages.ts` decides the lane after the route and its wire settle and after the managed-client steps already applied to the body (alias/modelMap resolution, `ocx-route`, effort @@ -238,14 +251,19 @@ feature effect. selection, 401 and 429 key-pool rotation, same-target 429 replay, the reset/transient retry policy and `sendWithConnectionPolicy`. Before sending it runs the image normalizer, the image guard and the tool-call-id repair the caller-forward passthrough runs. A streaming caller gets the -upstream SSE relayed byte for byte through `tapAnthropicSseForLog`, which records usage and the +upstream SSE relayed through `tapAnthropicSseForLog`, which records usage and the terminal and applies the body stall and size guards; a non-streaming caller gets the upstream JSON -(or a folded stream). Upstream errors answer in Anthropic shape with the translated lane's status +(or a folded stream). Either way `model` is rewritten to the selector the client sent, as the +translated lane answers (`message_start.message.model` on a stream, found within the first 64 KiB; +everything else is relayed as is). Upstream errors answer in Anthropic shape with the translated lane's status policy (transient 5xx as 529, replay refusals kept non-retryable). `count_tokens` estimates the body the builder would send when the route is eligible, and sends nothing. `tests/adapters/anthropic/anthropic-messages-passthrough.test.ts`, -`tests/responses/messages-native-eligibility.test.ts` and -`tests/claude-integration/messages-native.test.ts` pin the builder, the rule and the lane. +`tests/responses/messages-native-eligibility.test.ts`, +`tests/responses/messages-native-bridge-policy.test.ts`, +`tests/claude-integration/messages-native.test.ts` and +`tests/claude-integration/messages-native-decline-trace.test.ts` pin the builder, the rule, the +lane and the decline trace. ## Settings