feat(core): Anthropic prompt-cache breakpoints in the model registry

Anthropic caching is opt-in, and nothing sent breakpoints: observed
sessions show 0% cache hits on Claude models (~1.27M of 1.36M sampled
input tokens billed at full rate) vs ~82% implicit hits on Gemini. Two
ephemeral breakpoints fix that: the system prompt (whose cache prefix
also covers the tool schemas — both immutable per turn by construction)
and the last message (Anthropic's incremental-conversation pattern).
Conservative simulation on the sampled traffic floors the saving at 44%
with no cross-turn reuse; realistic reuse lands 70-85%.

Applied in the model registry bridge just before streamText, gated by
provider flavor or model id (covers direct Anthropic, OpenRouter, and
the gateways — the installed OpenRouter provider reads the same
providerOptions.anthropic key). Transport-only: nothing is persisted,
message content is untouched, and non-Anthropic requests pass through
byte-identical. Verify with cachedInputTokens on model_call_completed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

apps/x/packages/core/docs/turn-runtime-design.md

@@ -418,6 +418,12 @@ bytes than were transmitted; the inspect CLI prints the active policy so
 the divergence is visible. If exact-bytes replay ever becomes a hard
 requirement, record the applied policy on the turn and compose from that.
 
+Relatedly, Anthropic-family requests get cache_control breakpoints stamped
+at the transport layer (`models/prompt-caching.ts`, applied inside the
+model registry bridge just before streamText). These are provider metadata
+only — message content is untouched, nothing is persisted, and inspect
+does not render them; non-Anthropic requests pass through byte-identical.
+
 ### 6.7 Agent snapshot inheritance
 
 Session turns whose resolved system prompt and tool set are byte-identical

apps/x/packages/core/src/models/prompt-caching.test.ts

@@ -0,0 +1,100 @@
+import { describe, expect, it } from "vitest";
+import { applyPromptCaching, isAnthropicModel } from "./prompt-caching.js";
+
+const BREAKPOINT = { anthropic: { cacheControl: { type: "ephemeral" } } };
+
+function prompt(messageCount = 3) {
+    return {
+        system: "SYS",
+        messages: Array.from({ length: messageCount }, (_, i) => ({
+            role: i % 2 === 0 ? "user" : "assistant",
+            content: `m${i}`,
+        })),
+    };
+}
+
+describe("isAnthropicModel", () => {
+    it("matches the direct provider regardless of model id", () => {
+        expect(isAnthropicModel("anthropic", "whatever")).toBe(true);
+    });
+
+    it("matches aggregator ids by prefix or claude name", () => {
+        expect(isAnthropicModel("rowboat", "anthropic/claude-opus-4.8")).toBe(true);
+        expect(isAnthropicModel("openrouter", "anthropic/claude-3.5-sonnet")).toBe(true);
+        expect(isAnthropicModel("aigateway", "claude-sonnet-5")).toBe(true);
+    });
+
+    it("rejects everything else", () => {
+        expect(isAnthropicModel("openai", "gpt-test")).toBe(false);
+        expect(isAnthropicModel("rowboat", "google/gemini-3.5-flash")).toBe(false);
+        expect(isAnthropicModel("ollama", "llama3")).toBe(false);
+        expect(isAnthropicModel("openai-compatible", "qwen3")).toBe(false);
+    });
+});
+
+describe("applyPromptCaching", () => {
+    it("returns non-Anthropic prompts unchanged (same references)", () => {
+        const input = prompt();
+        const output = applyPromptCaching("openai", "gpt-test", input);
+        expect(output).toBe(input);
+    });
+
+    it("moves the system prompt into a breakpointed system message", () => {
+        const { system, messages } = applyPromptCaching("anthropic", "m", prompt());
+        expect(system).toBeUndefined();
+        expect(messages[0]).toEqual({
+            role: "system",
+            content: "SYS",
+            providerOptions: BREAKPOINT,
+        });
+    });
+
+    it("marks only the last message; middle messages stay untouched", () => {
+        const input = prompt(3);
+        const { messages } = applyPromptCaching("anthropic", "m", input);
+        // [system, m0, m1, m2]
+        expect(messages).toHaveLength(4);
+        expect(messages[1]).toBe(input.messages[0]);
+        expect(messages[2]).toBe(input.messages[1]);
+        expect(messages[3]).toEqual({
+            role: "user",
+            content: "m2",
+            providerOptions: BREAKPOINT,
+        });
+    });
+
+    it("does not mutate the input prompt", () => {
+        const input = prompt(2);
+        const snapshot = JSON.parse(JSON.stringify(input));
+        applyPromptCaching("anthropic", "m", input);
+        expect(input).toEqual(snapshot);
+    });
+
+    it("merges with pre-existing providerOptions on the last message", () => {
+        const input = {
+            system: "SYS",
+            messages: [
+                {
+                    role: "user",
+                    content: "hi",
+                    providerOptions: { anthropic: { foo: 1 }, openai: { bar: 2 } },
+                },
+            ],
+        };
+        const { messages } = applyPromptCaching("anthropic", "m", input);
+        expect(messages[1]).toEqual({
+            role: "user",
+            content: "hi",
+            providerOptions: {
+                openai: { bar: 2 },
+                anthropic: { foo: 1, cacheControl: { type: "ephemeral" } },
+            },
+        });
+    });
+
+    it("is deterministic (same input, same bytes)", () => {
+        const a = applyPromptCaching("anthropic", "m", prompt());
+        const b = applyPromptCaching("anthropic", "m", prompt());
+        expect(a).toEqual(b);
+    });
+});

apps/x/packages/core/src/models/prompt-caching.ts

@@ -0,0 +1,92 @@
+import type { JsonValue } from "@x/shared/dist/turns.js";
+
+// Anthropic prompt caching is opt-in: without explicit cache_control
+// breakpoints every call re-bills the full prefix at the base input rate
+// (observed sessions: 0% cache hits on Claude vs ~82% implicit hits on
+// Gemini, whose caching is automatic). Two breakpoints of the 4 allowed:
+//
+//   1. the system prompt — Anthropic's cache prefix is tools -> system ->
+//      messages, so this single breakpoint also covers the tool schemas;
+//      both are immutable for the life of a turn (agent snapshot),
+//   2. the last message — the incremental-conversation pattern: each call
+//      writes a cache entry at the conversation tip, and the next call's
+//      lookup reuses the longest previously written prefix.
+//
+// Writes bill 1.25x on the newly cached segment and reads bill 0.1x, so a
+// segment pays for itself on its first re-read; only tokens written and
+// never read again (a session's final tail) cost extra, capped at 25%.
+//
+// This is a transport-layer decoration ONLY: nothing is persisted, so the
+// durable log stays provider-agnostic and recomposition is unaffected.
+// Non-Anthropic requests pass through byte-identical. Failure degrades
+// soft: the AI SDK Anthropic provider validates placement and ignores
+// (with a warning) misplaced or excess breakpoints, and the OpenRouter
+// provider reads the same providerOptions.anthropic key.
+const CACHE_CONTROL: Record<string, JsonValue> = {
+    cacheControl: { type: "ephemeral" },
+};
+
+// Anthropic models arrive three ways: the direct provider (flavor
+// "anthropic", any model id), and aggregators (openrouter, aigateway, the
+// rowboat gateway) that address them as "anthropic/<model>" or "claude-*".
+export function isAnthropicModel(flavor: string, modelId: string): boolean {
+    if (flavor === "anthropic") {
+        return true;
+    }
+    const id = modelId.toLowerCase();
+    return id.startsWith("anthropic/") || id.includes("claude");
+}
+
+export interface PromptShape {
+    system?: string;
+    messages: JsonValue[];
+}
+
+export function applyPromptCaching(
+    flavor: string,
+    modelId: string,
+    prompt: { system: string; messages: JsonValue[] },
+): PromptShape {
+    if (!isAnthropicModel(flavor, modelId)) {
+        return prompt;
+    }
+    // The system string moves into the message array so it can carry
+    // per-message providerOptions (the AI SDK's documented breakpoint
+    // mechanism); the provider reassembles it into the system field.
+    const systemMessage: JsonValue = {
+        role: "system",
+        content: prompt.system,
+        providerOptions: { anthropic: CACHE_CONTROL },
+    };
+    const messages = prompt.messages.map((message, index) => {
+        if (index !== prompt.messages.length - 1) {
+            return message;
+        }
+        const existing =
+            typeof message === "object" && message !== null && !Array.isArray(message)
+                ? message
+                : {};
+        const existingOptions =
+            "providerOptions" in existing &&
+            typeof existing.providerOptions === "object" &&
+            existing.providerOptions !== null &&
+            !Array.isArray(existing.providerOptions)
+                ? existing.providerOptions
+                : {};
+        return {
+            ...existing,
+            providerOptions: {
+                ...existingOptions,
+                anthropic: {
+                    ...(typeof existingOptions.anthropic === "object" &&
+                    existingOptions.anthropic !== null &&
+                    !Array.isArray(existingOptions.anthropic)
+                        ? existingOptions.anthropic
+                        : {}),
+                    ...CACHE_CONTROL,
+                },
+            },
+        };
+    });
+    return { messages: [systemMessage, ...messages] };
+}

apps/x/packages/core/src/turns/bridges/real-model-registry.test.ts

@@ -188,6 +188,64 @@ describe("RealModelRegistry", () => {
         });
     });
 
+    it("applies cache breakpoints for Anthropic-family models", async () => {
+        const capture: InvokerOptions[] = [];
+        const fakeModel = { modelId: "claude-test" } as unknown as LanguageModel;
+        const registry = new RealModelRegistry({
+            resolveProvider: async () => ({ flavor: "anthropic" }),
+            createProviderImpl: (() => ({
+                languageModel: () => fakeModel,
+            })) as never,
+            invoke: (options) => {
+                capture.push(options);
+                return {
+                    fullStream: (async function* () {
+                        yield { type: "finish-step", finishReason: "stop", usage: {} };
+                    })(),
+                };
+            },
+        });
+        const model = await registry.resolve({
+            provider: "anthropic",
+            model: "claude-test",
+        });
+        const drained: unknown[] = [];
+        for await (const event of model.stream(request())) {
+            drained.push(event);
+        }
+        expect(drained.length).toBeGreaterThan(0);
+
+        const [options] = capture;
+        // System prompt rides the message array with a breakpoint; no
+        // separate system string is sent.
+        expect(options.system).toBeUndefined();
+        expect(options.messages[0]).toMatchObject({
+            role: "system",
+            content: "SYS",
+            providerOptions: { anthropic: { cacheControl: { type: "ephemeral" } } },
+        });
+        expect(options.messages[options.messages.length - 1]).toMatchObject({
+            role: "user",
+            providerOptions: { anthropic: { cacheControl: { type: "ephemeral" } } },
+        });
+    });
+
+    it("leaves non-Anthropic requests byte-identical", async () => {
+        const capture: InvokerOptions[] = [];
+        const registry = makeRegistry(
+            [{ type: "finish-step", finishReason: "stop", usage: {} }],
+            capture,
+        );
+        const req = request();
+        const originalMessages = JSON.parse(JSON.stringify(req.messages));
+        await collect(registry, req);
+
+        const [options] = capture;
+        expect(options.system).toBe("SYS");
+        expect(options.messages).toEqual(originalMessages);
+        expect(JSON.stringify(options.messages)).not.toContain("cacheControl");
+    });
+
     it("stops promptly when the signal aborts mid-stream", async () => {
         const controller = new AbortController();
         const registry = makeRegistry(

apps/x/packages/core/src/turns/bridges/real-model-registry.ts

@@ -14,6 +14,7 @@ import type { JsonValue, ModelDescriptor, TurnUsage } from "@x/shared/dist/turns
 import { convertFromMessages } from "../../agents/runtime.js";
 import { resolveProviderConfig } from "../../models/defaults.js";
 import { createProvider } from "../../models/models.js";
+import { applyPromptCaching } from "../../models/prompt-caching.js";
 import { applyLocalModelSettings } from "../../models/local.js";
 import type {
     IModelRegistry,
@@ -26,7 +27,9 @@ import type {
 // provider. The bridge always requests exactly one step.
 export type StreamTextInvoker = (options: {
     model: LanguageModel;
-    system: string;
+    // Absent when prompt caching moved the system prompt into messages
+    // (Anthropic-family models; see models/prompt-caching.ts).
+    system?: string;
     messages: ModelMessage[];
     tools: ToolSet;
     abortSignal: AbortSignal;
@@ -79,13 +82,16 @@ export class RealModelRegistry implements IModelRegistry {
             // per-message, so composed requests are byte-stable.
             encodeMessages: (messages) =>
                 convertFromMessages(messages) as unknown as JsonValue[],
-            stream: (request) => this.run(model, request),
+            stream: (request) =>
+                this.run(model, request, providerConfig.flavor, descriptor.model),
         };
     }
 
     private async *run(
         model: LanguageModel,
         request: ModelStreamRequest,
+        flavor: string,
+        modelId: string,
     ): AsyncGenerator<LlmStreamEvent, void, void> {
         const tools: ToolSet = {};
         for (const descriptor of request.tools) {
@@ -121,10 +127,17 @@ export class RealModelRegistry implements IModelRegistry {
         let usage: z.infer<typeof TurnUsage> = {};
         let providerMetadata: JsonValue | undefined;
 
+        // Anthropic-family models get cache_control breakpoints; everything
+        // else passes through byte-identical (transport-only, not persisted).
+        const prompt = applyPromptCaching(flavor, modelId, {
+            system: request.systemPrompt,
+            messages: request.messages,
+        });
+
         const result = this.invoke({
             model,
-            system: request.systemPrompt,
-            messages: request.messages as ModelMessage[],
+            ...(prompt.system === undefined ? {} : { system: prompt.system }),
+            messages: prompt.messages as ModelMessage[],
             tools,
             abortSignal: request.signal,
             ...generationParams,