diff --git a/src/adapters/anthropic.ts b/src/adapters/anthropic.ts index 0b0f36739d5..4e25e263bf6 100644 --- a/src/adapters/anthropic.ts +++ b/src/adapters/anthropic.ts @@ -122,6 +122,8 @@ function applyCacheControlToLastText(blocks: Array>, cc: type PromptCachingOptions = { maxExplicitBreakpoints?: number; skipLastUser?: boolean; + /** The last system block is the caller's uncached per-turn tail; mark the block before it. */ + uncachedSystemTail?: boolean; }; /** Place explicit cache_control breakpoints on the built Anthropic body. */ @@ -158,7 +160,8 @@ function applyPromptCaching( // 2. system const system = body.system as Array> | undefined; if (system && system.length > 0) { - applyCacheControlToLast(system, cc); + const i = system.length - (options.uncachedSystemTail && system.length > 1 ? 2 : 1); + system[i] = { ...system[i], cache_control: cc }; used++; } if (used >= explicitLimit || !messages) return; @@ -759,7 +762,7 @@ function messagesToAnthropicFormat( parsed: OcxParsedRequest, toolNames: { toWire: (name: string) => string }, serverReplay?: ReadonlyMap[]>, -): { system: string | undefined; messages: unknown[] } { +): { system: string[]; messages: unknown[] } { // The client's server-tool exchanges go back ahead of the block they preceded. Their anchors are // unique in the caller's history (replayedAnthropicServerBlocks drops repeated ones); each is used once. const pendingReplay = new Map(serverReplay); @@ -787,9 +790,13 @@ function messagesToAnthropicFormat( tool => toolNames.toWire(namespacedToolName(tool.namespace, tool.name)), ); const systemParts = [...(parsed.context.systemPrompt ?? []), ...(toolCatalogNudge ? [toolCatalogNudge] : [])]; - const system = systemParts.length - ? identifyRoutedModel(systemParts.join("\n\n"), parsed.modelId) || undefined - : undefined; + // End a block at the caller's cached prefix so its breakpoint skips the per-turn tail. + const prefix = parsed._anthropicSystemCachePrefix; + const head = systemParts[0]; + const texts = prefix && head?.startsWith(`${prefix}\n\n`) + ? [prefix, [head.slice(prefix.length + 2), ...systemParts.slice(1)].join("\n\n")] + : [systemParts.join("\n\n")]; + const system = texts.map(text => identifyRoutedModel(text, parsed.modelId)).filter(Boolean); const messages: unknown[] = []; for (let i = 0; i < parsed.context.messages.length; i++) { @@ -1078,10 +1085,10 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti // Claude OAuth (Pro/Max) requires the first system block to be the Claude Code identity. body.system = [ { type: "text", text: CLAUDE_CODE_SYSTEM_INSTRUCTION }, - ...(system ? [{ type: "text", text: system }] : []), + ...system.map(text => ({ type: "text", text })), ]; - } else if (system) { - body.system = [{ type: "text", text: system }]; + } else if (system.length > 0) { + body.system = system.map(text => ({ type: "text", text })); } if (tools) body.tools = tools; if (parsed.options.temperature !== undefined) body.temperature = parsed.options.temperature; @@ -1277,6 +1284,7 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti applyPromptCaching(body, cc, { maxExplicitBreakpoints: explicitLimit, skipLastUser: !!automaticPromptCaching, + uncachedSystemTail: system.length > 1, }); enforceCacheControlLimit(body, explicitLimit); normalizeTtlOrdering(body); diff --git a/src/claude/inbound-content-options.ts b/src/claude/inbound-content-options.ts index fdcf6051585..575d763b8e9 100644 --- a/src/claude/inbound-content-options.ts +++ b/src/claude/inbound-content-options.ts @@ -40,6 +40,18 @@ export function systemToInstructions(system: unknown): string | undefined { return undefined; } +/** + * The `instructions` prefix ending at the caller's last cache-marked system block, when blocks + * follow it. Claude Code ends its system with a per-turn env block; joining everything into one + * Anthropic system block moved the breakpoint onto those changing bytes, so system and every + * later breakpoint missed on each turn. The Anthropic adapter splits its system here instead. + */ +export function systemCachePrefix(system: unknown): string | undefined { + if (!Array.isArray(system)) return undefined; + const last = system.findLastIndex(block => isRec(block) && block.cache_control != null); + return last >= 0 && last < system.length - 1 ? systemToInstructions(system.slice(0, last + 1)) : undefined; +} + export function toolsToResponses(tools: unknown): Rec[] | undefined { if (!Array.isArray(tools) || tools.length === 0) return undefined; // The shared hosted-search lane cannot enforce these constraints on every destination or diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index 1a7192e91f4..193fcb67b84 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -40,7 +40,7 @@ import { stripOneMillionMarker } from "../claude/context-windows"; import { captureAdmissionPolicy } from "./capture-admission-policy"; import type { CapturedAdmissionPolicy } from "../types/admission-policy"; import { captureClaudeInbound } from "../claude/inbound-debug"; -import { declaredAnthropicServerTools, replayedAnthropicServerBlocks } from "../claude/inbound-content-options"; +import { declaredAnthropicServerTools, replayedAnthropicServerBlocks, systemCachePrefix } from "../claude/inbound-content-options"; import { claudeCodeForIngress } from "../claude/intercept/model-bindings"; import { classifyInterceptClient } from "../claude/intercept/client-class"; import { analyzeClaudeCompatibility, isClaudeCompatibilityMode } from "../claude/compatibility"; @@ -1253,6 +1253,7 @@ async function handleClaudeMessagesWithBudget( // Out of body like the replay above: the core hands these to an Anthropic destination only. anthropicServerTools: declaredAnthropicServerTools(anthropicBody), anthropicServerReplay: replayedAnthropicServerBlocks(anthropicBody), + anthropicSystemCachePrefix: isRec(anthropicBody) ? systemCachePrefix(anthropicBody.system) : undefined, // Preserve the original catalog through late fallback; only the physical adapter projects it. anthropicToolLoadingSource: cacheKeySource === "metadata" && isRec(anthropicBody) && Array.isArray(cc?.appendOnlyToolLoadingModels) && cc.appendOnlyToolLoadingModels.length > 0 diff --git a/src/server/responses/core-options.ts b/src/server/responses/core-options.ts index d6687bb0143..be3aa8d2791 100644 --- a/src/server/responses/core-options.ts +++ b/src/server/responses/core-options.ts @@ -198,6 +198,8 @@ export interface HandleResponsesOptions { anthropicServerTools?: readonly Record[]; /** The Claude client's replayable server-tool exchanges (`replayedAnthropicServerBlocks`); same scope. */ anthropicServerReplay?: ReadonlyMap[]>; + /** The caller's cached system prefix (`systemCachePrefix`); same scope. */ + anthropicSystemCachePrefix?: string; /** Metadata-qualified Messages discovery evidence, never caller JSON or a preselected destination. */ anthropicToolLoadingSource?: Readonly>; /** Droid's per-request effort default; each concrete combo or policy target applies it only if its ladder allows it. */ diff --git a/src/server/responses/request-prepare.ts b/src/server/responses/request-prepare.ts index 9d373415c04..0fc824890dc 100644 --- a/src/server/responses/request-prepare.ts +++ b/src/server/responses/request-prepare.ts @@ -515,6 +515,7 @@ export async function prepareResponsesRequest( parsed._nativeReasoningMint = options.nativeReasoningMint; parsed._anthropicServerTools = options.anthropicServerTools; parsed._anthropicServerReplay = options.anthropicServerReplay; + parsed._anthropicSystemCachePrefix = options.anthropicSystemCachePrefix; if (parsed.options.textFormat?.type === "json_schema") parsed._structuredOutputConformance = true; } parsed._promptCacheKeyIsSharedCohort = options.promptCacheKeyIsSharedCohort; diff --git a/src/types/request.ts b/src/types/request.ts index c5423f1a568..1e5878e0b5f 100644 --- a/src/types/request.ts +++ b/src/types/request.ts @@ -86,6 +86,11 @@ export interface OcxParsedRequest { * followed them (`call:` / `text:`). Same audience as `_anthropicServerTools`. */ _anthropicServerReplay?: ReadonlyMap[]>; + /** + * The caller's `instructions` prefix through its last cache-marked system block. The anthropic + * adapter ends a system block there so the breakpoint stays off the per-turn tail. + */ + _anthropicSystemCachePrefix?: string; /** * Boundary between replayed history and this turn's newly appended input. Usually the * items the proxy restored from local previous_response_id state; also set when the diff --git a/tests/claude-integration/claude-translated-cache-ttl.test.ts b/tests/claude-integration/claude-translated-cache-ttl.test.ts index 154423fa9f4..37a58d2db1e 100644 --- a/tests/claude-integration/claude-translated-cache-ttl.test.ts +++ b/tests/claude-integration/claude-translated-cache-ttl.test.ts @@ -29,6 +29,19 @@ afterAll(() => { removeTreeWithRetry(home); }); +function sseReply(): Response { + const frames = [ + { type: "message_start", message: { id: "msg_x", type: "message", role: "assistant", model: MODEL, content: [], stop_reason: null, usage: { input_tokens: 1, 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: "ok" } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]; + return new Response(frames.map(f => `event: ${f.type}\ndata: ${JSON.stringify(f)}\n\n`).join(""), + { headers: { "content-type": "text/event-stream" } }); +} + async function wireTtls( callerTtl: "1h" | undefined, cacheRetention?: "short" | "long", @@ -39,16 +52,7 @@ async function wireTtls( const wires: string[] = []; const fetch = (async (_input: unknown, init?: RequestInit) => { wires.push(String(init?.body)); - const frames = [ - { type: "message_start", message: { id: "msg_x", type: "message", role: "assistant", model: MODEL, content: [], stop_reason: null, usage: { input_tokens: 1, 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: "ok" } }, - { type: "content_block_stop", index: 0 }, - { type: "message_delta", delta: { stop_reason: "end_turn", stop_sequence: null }, usage: { output_tokens: 1 } }, - { type: "message_stop" }, - ]; - return new Response(frames.map(f => `event: ${f.type}\ndata: ${JSON.stringify(f)}\n\n`).join(""), - { headers: { "content-type": "text/event-stream" } }); + return sseReply(); }) as typeof globalThis.fetch; const config = { port: 0, defaultProvider: "anth", @@ -153,3 +157,47 @@ describe("callerCacheRetention walks every block container", () => { })).toBeUndefined(); }); }); + +describe("translated Claude lane system breakpoint", () => { + // Claude Code marks its stable system block and ends with a per-turn env block. The single + // joined system block put the breakpoint on the env bytes, so no turn could read the cache. + async function wireSystem(env: string): Promise { + const wires: string[] = []; + const fetch = (async (_input: unknown, init?: RequestInit) => { + wires.push(String(init?.body)); + return sseReply(); + }) as typeof globalThis.fetch; + const config = { + port: 0, defaultProvider: "anth", claudeCode: { nativePassthrough: false }, + providers: { anth: { adapter: "anthropic", baseUrl: "https://api.anthropic.com", authMode: "key", apiKey: "sk-ant-fixture", models: [MODEL], fetch } }, + } as unknown as OcxConfig; + const response = await handleClaudeMessages(new Request("http://localhost/v1/messages", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: MODEL, max_tokens: 64, + system: [ + { type: "text", text: "You are Claude Code." }, + { type: "text", text: "Stable instructions.", cache_control: { type: "ephemeral" } }, + { type: "text", text: env }, + ], + messages: [{ role: "user", content: "Hi." }], + }), + }), config, { model: "", provider: "" }); + expect(response.status).toBe(200); + await response.text(); + expect(wires).toHaveLength(1); + return JSON.parse(wires[0]!).system; + } + + test("the breakpoint ends at the caller's cached block, before the per-turn tail", async () => { + const first = await wireSystem("Today: 2026-11-03"); + const second = await wireSystem("Today: 2026-11-04"); + const cached = (system: unknown[]) => system.filter(b => (b as { cache_control?: unknown }).cache_control); + expect(cached(first)).toHaveLength(1); + expect(cached(first)).toEqual(cached(second)); + expect(JSON.stringify(cached(first))).toContain("Stable instructions."); + expect(JSON.stringify(cached(first))).not.toContain(""); + expect(JSON.stringify(first.at(-1))).toContain("2026-11-03"); + }); +});