feat(protocols): native Anthropic Messages support for combos

scripts/test-layout/layout-replay.json

@@ -7,5 +7,6 @@
   "combo-replay-refusal-attribution.test.ts": "responses",
   "responses-replay-metadata-principal.test.ts": "responses",
   "responses-combo-replay-principal.test.ts": "responses",
-  "responses-translated-replay-principal.test.ts": "responses"
+  "responses-translated-replay-principal.test.ts": "responses",
+  "messages-native-combo.test.ts": "claude-integration"
 }

src/protocols/envelope.ts

@@ -19,6 +19,11 @@ export interface ProtocolEnvelope {
   features(): ReadonlySet<ProtocolFeature>;
   /** A structured clone of the source body, charged to the translator budget. */
   freshBody(): Record<string, unknown>;
+  /**
+   * The source body by reference, uncharged and read-only: for a consumer that only inspects the
+   * request (a candidate eligibility check) and so needs no copy. Never mutate or retain it.
+   */
+  peekBody(): Readonly<Record<string, unknown>>;
 }
 
 export function createProtocolEnvelope(input: {
@@ -34,6 +39,9 @@ export function createProtocolEnvelope(input: {
       features ??= featuresFromBody(inbound, body);
       return features;
     },
+    peekBody() {
+      return body;
+    },
     freshBody() {
       // Charged before the copy exists, so an over-budget request fails without allocating it.
       translatorBudget.chargeRetained(jsonUtf8Bytes(body), { kind: "request_copies" });

src/protocols/plan-snapshot.ts

@@ -127,10 +127,13 @@ function candidateFor(
   let declineReasons: ProtocolReasonCode[] = [];
   let nativeEligible = false;
   let directEligible = 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;
+  // With `nativeChatCombos` on, or with `managedMessagesNative` on for a Messages combo candidate,
+  // 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" && (
+    (inbound === "chat" && resolveProtocolSettings(config).rollout.nativeChatCombos)
+    || (inbound === "messages" && resolveProtocolSettings(config, route.providerName).rollout.managedMessagesNative)
+  );
   const settled: RouteResult = {
     ...route,
     provider,

src/server/claude-messages.ts

@@ -87,11 +87,12 @@ import {
   sessionLaneIdFromRequest,
 } from "./request-log-conversation";
 import { responseWithDeferredRequestLog } from "./relay";
-import { clientWireOf } from "./inference/client-wire";
+import { clientWireLogOf, clientWireOf } from "./inference/client-wire";
 import { directEncodersApply } from "./inference/client-encoder-delivery";
 import type { ClientEncoderOption } from "./responses/core-options";
 import { handleResponses } from "./responses";
 import { previewXaiOauthWireModel } from "./responses/core-normalize";
+import { getCombo } from "../combos";
 import { upstreamWireForAdapter } from "../protocols/contract";
 import { createProtocolEnvelope, type ProtocolEnvelope } from "../protocols/envelope";
 import { featuresFromMessagesBody, type ProtocolFeature } from "../protocols/features";
@@ -638,8 +639,9 @@ async function handleClaudeMessagesWithBudget(
   let directPlan: MessagesDirectPlan | undefined;
   /** The retained charge for the settings-only body, handed back if the plan is left. */
   let directSettingsBodyBytes = 0;
-  // Built only under the reject policy; the legacy default leaves this request untouched.
+  // Built only under the reject policy or for native combos; the legacy default leaves this request untouched.
   let envelope: ProtocolEnvelope | undefined;
+  let nativeMessagesCombo = false;
   let messagesFeatures: () => Iterable<ProtocolFeature> = () => [];
   // The Advisor caller policy is the admitted snapshot: capture it before the first await (#7859).
   const advisorAdmissionPolicy = logIds?.advisorAdmissionPolicy ?? captureAdmissionPolicy(config, logIds?.admission);
@@ -916,6 +918,13 @@ async function handleClaudeMessagesWithBudget(
     route.provider = resolveWireProtocolOverride(route.providerName, route.modelId, route.provider, "anthropic", route.staticPolicy);
     logCtx.routeDecision = route.routeDecision;
     settledRoute = route;
+    const comboConfig = settledRoute?.combo ? getCombo(config, settledRoute.combo.comboId) : undefined;
+    nativeMessagesCombo = !!comboConfig && !effortRow && !fastRow && comboConfig.targets.some(
+      target => resolveProtocolSettings(config, target.provider).rollout.managedMessagesNative,
+    );
+    if (!envelope && isRec(anthropicBody) && nativeMessagesCombo) {
+      envelope = createProtocolEnvelope({ inbound: "messages", body: anthropicBody, translatorBudget });
+    }
     // Before any route-specific shaping, so a request that leaves the lane is shaped as the bridge's.
     if (directPlan && !messagesDirectPlanHolds(directPlan, route, anthropicBody as Rec, config)) directLeaveError = leaveDirectPlan();
     if (directEncodersApply(config, route)) {
@@ -1269,15 +1278,37 @@ async function handleClaudeMessagesWithBudget(
     ...(logIds ? { onFirstOutput: () => recordFirstOutput(logCtx, logIds.start) } : {}),
     onNativePassthroughTerminal: status => finalizeNativeLog(httpStatusForRequestLogTerminal(status, logCtx), { terminalStatus: status, closeReason: "terminal" }),
     onNativePassthroughCancel: () => finalizeNativeLog(499, { closeReason: "client_cancel" }),
+    ...(envelope && nativeMessagesCombo ? {
+      protocolSource: (await import("./messages-native")).createNativeMessagesComboSource({
+        req,
+        config,
+        envelope,
+        selector: nativeSelector,
+        requestedModel: clientFacingModelText(clientModelLabel),
+        translatorBudget,
+        callerAnthropicBeta: req.headers.get("anthropic-beta"),
+        clientIdentity: captureAnthropicClientIdentity(req.headers),
+        sessionKey: anthropicSessionKeyFromParts({
+          sessionIdHeader: req.headers.get("session_id")?.trim() || req.headers.get("x-claude-code-session-id"),
+          threadIdHeader: req.headers.get("thread_id"),
+          clientThreadId: conversationIdFromClaudeMetadata(isRec(anthropicBody) && isRec(anthropicBody.metadata) ? anthropicBody.metadata : undefined),
+          promptCacheKey: typeof internalBody.prompt_cache_key === "string" ? internalBody.prompt_cache_key : null,
+          promptCacheKeyIsSharedCohort: cacheKeySource === "system",
+        }),
+        modelScope: resolveAdmissionModelScope(config, logIds?.admission),
+      }),
+    } : {}),
     clientModelLabel,
     ...(clientEncoder ? { clientEncoder } : {}),
     ...(directRoute && directPlan
       ? { directRequest: messagesDirectRequestOption(directPlan, anthropicBody, cc, translatorBudget) }
       : {}),
   });
+  if (clientWireOf(upstream) === "messages") {
+    if (!logIds || (upstream.ok && !clientWireLogOf(upstream))) return upstream;
+    return responseWithDeferredRequestLog(upstream, logIds.requestId, logIds.start, logCtx);
+  }
   const response = logIds ? responseWithDeferredRequestLog(upstream, logIds.requestId, logIds.start, logCtx) : upstream;
-  // Already in the Messages wire (direct encoder): no conversion.
-  if (clientWireOf(upstream) === "messages") return response;
 
   if (!response.ok) {
     // Read the shared provenance verdict before consuming and re-wrapping the body. A refusal

src/server/messages-native.ts

@@ -1,6 +1,8 @@
 import { captureAccountAutoPauseAttempt, wasAccountAutoPaused } from "../oauth/account-auto-pause";
-import { createPhysicalSendReporter } from "../lib/request-execution-budget";
+import { createPhysicalSendReporter, type RequestExecutionBudget, type SingleUseDispatchPermit } from "../lib/request-execution-budget";
 import { createInferenceSendBudget } from "./inference/context";
+import type { ComboProtocolSource } from "./responses/core-combo-native";
+import type { ProtocolEnvelope } from "../protocols/envelope";
 /**
  * Managed native Messages lane (PF-08, behind `protocols.rollout.managedMessagesNative`).
  *
@@ -96,7 +98,7 @@ import {
   sanitizePassthroughToolCallIds,
   tapAnthropicSseForLog,
 } from "./messages-shared";
-import { beginInferenceAttempt } from "./inference/attempt";
+import { beginInferenceAttempt, type InferenceAttempt } from "./inference/attempt";
 import { createFinalRequestLog, type FinalRequestLogMeta } from "./inference/final-log";
 import { registerTurn, unregisterTurn } from "./lifecycle";
 import { nativeMessagesDeclineReason, type NativeMessagesSelector } from "./messages-native-eligibility";
@@ -185,7 +187,20 @@ export interface HandleNativeMessagesOptions {
   modelScope?: AdmissionModelScope;
 }
 
-type FinishLog = (status: number, message?: string, meta?: FinalRequestLogMeta) => void;
+export type NativeMessagesFinishLog = (
+  status: number,
+  message?: string,
+  meta?: FinalRequestLogMeta | "non_stream" | "terminal" | "client_cancel",
+) => void;
+type FinishLog = NativeMessagesFinishLog;
+
+export interface NativeMessagesExecution extends HandleNativeMessagesOptions {
+  finishLog: NativeMessagesFinishLog;
+  sendBudget?: RequestExecutionBudget;
+  comboDispatchPermit?: SingleUseDispatchPermit;
+  onFirstOutput?: () => void;
+  turnAdmissionLease?: AdmissionLease;
+}
 
 /** Relay an upstream body unchanged, recording first output on its first non-empty chunk. */
 function observeFirstChunk(body: ReadableStream<Uint8Array>, onFirst: () => void): ReadableStream<Uint8Array> {
@@ -320,7 +335,35 @@ async function prepareNativeBody(body: Rec, signal: AbortSignal): Promise<void>
  * log row, and answers in the Messages wire (SSE or JSON per the caller's own `stream`).
  */
 export async function handleNativeMessages(options: HandleNativeMessagesOptions): Promise<Response> {
-  const { req, config, logCtx, logIds, route, body, requestedModel, translatorBudget } = options;
+  const { logCtx, logIds, route } = options;
+  logCtx.inboundProtocol = "messages";
+  const attemptHandle = beginInferenceAttempt(logCtx, {
+    provider: route.providerName,
+    model: route.modelId,
+    adapter: "anthropic",
+  });
+  attemptHandle.seal(logCtx.accountLogLabel);
+  const finalLog = createFinalRequestLog(logIds, logCtx);
+  const finishLog: NativeMessagesFinishLog = (status, message, meta = { closeReason: "non_stream" }) => {
+    if (finalLog.finished()) return;
+    if (message) logCtx.upstreamError = redactSecretString(message).slice(0, 500);
+    const resolvedMeta: FinalRequestLogMeta = typeof meta === "string" ? { closeReason: meta } : meta;
+    finalLog.finish(status, resolvedMeta);
+  };
+  return runNativeMessagesAttempt({ ...options, finishLog }, attemptHandle);
+}
+
+/**
+ * Run one native Messages attempt on an already-open attempt row.
+ * Used for both standalone native Messages requests and native combo children.
+ */
+export async function runNativeMessagesAttempt(
+  execution: NativeMessagesExecution,
+  attemptHandle: InferenceAttempt,
+): Promise<Response> {
+  const options = execution;
+  const { req, config, logCtx, logIds, route, body, requestedModel, translatorBudget, finishLog } = execution;
+  const { sendBudget } = execution;
   const nativeInstance = configuredAnthropicInstance(config, route.providerName);
   const requestedStream = body.stream === true;
   const clientDeclaredEmptyToolCatalog = Array.isArray(body.tools) && body.tools.length === 0;
@@ -351,20 +394,8 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
     logCtx.spendOutputCeilingTokens = Math.trunc(body.max_tokens);
   }
 
-  const attemptHandle = beginInferenceAttempt(logCtx, {
-    provider: route.providerName,
-    model: route.modelId,
-    adapter: "anthropic",
-  });
-  attemptHandle.seal(logCtx.accountLogLabel);
   recordAttemptRequestedEffort(logCtx);
   const { attempt } = attemptHandle;
-  const finalLog = createFinalRequestLog(logIds, logCtx);
-  const finishLog: FinishLog = (status, message, meta = { closeReason: "non_stream" }) => {
-    if (finalLog.finished()) return;
-    if (message) logCtx.upstreamError = redactSecretString(message).slice(0, 500);
-    finalLog.finish(status, meta);
-  };
   const bindUsage = (usage: OcxUsage | undefined) => {
     if (!usage) return;
     if (!recordKeyWireAttemptUsage(logCtx, usage)) {
@@ -418,7 +449,7 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
   const cleanupAbort = linkAbortSignal(upstream, req.signal);
   let streamTurnRegistered = false;
   const transferTurnToStream = () => {
-    const lease = logIds?.turnAdmissionLease;
+    const lease = execution.turnAdmissionLease ?? logIds?.turnAdmissionLease;
     if (!lease || typeof (lease as { bindAbortController?: unknown }).bindAbortController !== "function") return;
     registerTurn(upstream, lease);
     streamTurnRegistered = true;
@@ -464,8 +495,9 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
   let activeProvider: OcxProviderConfig = oauthBinding ? oauthProvider(oauthBinding) : route.provider;
   stampApiKeyAccountLabel(logCtx, route.providerName, activeProvider);
   if (oauthBinding) logCtx.provider = formatAnthropicProviderForLog(route.providerName, oauthBinding.snapshot.accountId, config);
-  const spendTracker = attachRequestSpendTracker(req, logCtx);
-  const physicalBudget = createInferenceSendBudget(req, logCtx);
+  const spendTracker = sendBudget ? undefined : attachRequestSpendTracker(req, logCtx);
+  const physicalBudget = sendBudget ?? createInferenceSendBudget(req, logCtx);
+  const onFirstOutput = execution.onFirstOutput ?? (logIds ? () => recordFirstOutput(logCtx, logIds.start) : undefined);
   // Observe-only spend is already booked by the tracker; count billing sends without booking again.
   let physicalSendsUsed = 0;
   let activeRequest: AnthropicMessagesPassthroughRequest;
@@ -630,16 +662,18 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
             let dispatched: Response;
             const spendReport = createPhysicalSendReporter(physicalBudget, () => ({
               poolId: logCtx.spendPoolId ?? route.providerName, identityId: logCtx.accountLogLabel,
-            }));
+            }), execution.comboDispatchPermit);
             let sendOwner: AnthropicPhysicalSendOwnership | null = null;
             try {
               if (init.signal?.aborted) throw init.signal.reason;
               // Capture before either accounting path so this request retains its start policy.
               if (spendReport.beforeSend && !spendReport.beforeSend()) {
-                if (spendTracker.refusals) throw new NativeMessagesSpendRefusal();
+                if (spendTracker?.refusals || (logCtx.spendTracker as { refusals?: number } | undefined)?.refusals) {
+                  throw new NativeMessagesSpendRefusal();
+                }
                 throw new SendBudgetExhaustedError();
               }
-              if (!physicalBudget.spendEnforced && !spendTracker.charge()) throw new NativeMessagesSpendRefusal();
+              if (spendTracker && !physicalBudget.spendEnforced && !spendTracker.charge()) throw new NativeMessagesSpendRefusal();
               noteProviderAttemptSend(logCtx, route.providerName, activeProvider, logCtx.usageLogInputTokens, transportRecovery ?? recovery);
               // Last synchronous ownership reservation before the physical fetch. Retain this
               // exact incarnation across its await; a returned response can never reserve one.
@@ -896,7 +930,7 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
       ...resolvePassthroughBodyGuard(config, upstream.signal),
       ...(activeProvider.anthropicEofTolerance === true ? { eofTolerant: true } : {}),
     };
-    const observed = logIds ? observeFirstChunk(response.body, () => recordFirstOutput(logCtx, logIds.start)) : response.body;
+    const observed = onFirstOutput ? observeFirstChunk(response.body, onFirstOutput) : response.body;
     const renamed = activeRequest.oauthToolNames
       ? restoreOAuthToolNamesInSse(observed, activeRequest.oauthToolNames, translatorBudget)
       : observed;
@@ -1009,7 +1043,7 @@ export async function handleNativeMessages(options: HandleNativeMessagesOptions)
     return fail(502, undeclaredToolCallMessage(), "api_error", UNDECLARED_TOOL_CALL_ERROR_CODE);
   }
   bindUsage(anthropicUsageToOcx(isRec(message.usage) ? message.usage : undefined));
-  if (logIds) recordFirstOutput(logCtx, logIds.start);
+  if (onFirstOutput) onFirstOutput();
   try {
     translatorBudget.chargeRetained(Buffer.byteLength(read.text) * 2, { kind: "live_transient" });
   } catch (error) {
@@ -1150,3 +1184,44 @@ export function nativeMessagesCountBody(
     return undefined;
   }
 }
+
+/**
+ * The Messages source a combo hands its native children. A child runs on the attempt the
+ * combo opened and reports through the combo's callbacks, so the final row stays the parent's.
+ */
+export function createNativeMessagesComboSource(input: {
+  req: Request;
+  config: OcxConfig;
+  envelope: ProtocolEnvelope;
+  requestedModel: string;
+  translatorBudget: TranslatorBudget;
+  callerAnthropicBeta?: string | null;
+  clientIdentity?: AnthropicClientIdentity;
+  sessionKey?: string | null;
+  modelScope?: AdmissionModelScope;
+  selector?: NativeMessagesSelector;
+}): ComboProtocolSource {
+  return {
+    inbound: "messages",
+    ...(input.selector ? { selector: input.selector } : {}),
+    envelope: input.envelope,
+    dispatchNativeChild: child => runNativeMessagesAttempt({
+      req: input.req,
+      config: input.config,
+      logCtx: child.childLog,
+      route: child.route,
+      body: child.body,
+      requestedModel: input.requestedModel,
+      translatorBudget: input.translatorBudget,
+      finishLog: child.finishLog,
+      sendBudget: child.sendBudget,
+      comboDispatchPermit: child.comboDispatchPermit,
+      onFirstOutput: child.onFirstOutput,
+      callerAnthropicBeta: input.callerAnthropicBeta,
+      clientIdentity: input.clientIdentity,
+      sessionKey: input.sessionKey,
+      modelScope: input.modelScope,
+      ...(child.turnAdmissionLease ? { turnAdmissionLease: child.turnAdmissionLease } : {}),
+    }, child.attemptHandle),
+  };
+}