diff --git a/apps/x/packages/core/docs/turn-runtime-design.md b/apps/x/packages/core/docs/turn-runtime-design.md index 5f462bd1..2b0e54fa 100644 --- a/apps/x/packages/core/docs/turn-runtime-design.md +++ b/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 diff --git a/apps/x/packages/core/src/models/prompt-caching.test.ts b/apps/x/packages/core/src/models/prompt-caching.test.ts new file mode 100644 index 00000000..3d3fd8f7 --- /dev/null +++ b/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); + }); +}); diff --git a/apps/x/packages/core/src/models/prompt-caching.ts b/apps/x/packages/core/src/models/prompt-caching.ts new file mode 100644 index 00000000..089cc3ff --- /dev/null +++ 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 = { + 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/" 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] }; +} diff --git a/apps/x/packages/core/src/turns/bridges/real-model-registry.test.ts b/apps/x/packages/core/src/turns/bridges/real-model-registry.test.ts index ea0fe990..8fbf1a3e 100644 --- a/apps/x/packages/core/src/turns/bridges/real-model-registry.test.ts +++ b/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( diff --git a/apps/x/packages/core/src/turns/bridges/real-model-registry.ts b/apps/x/packages/core/src/turns/bridges/real-model-registry.ts index 36935d8a..40665a44 100644 --- a/apps/x/packages/core/src/turns/bridges/real-model-registry.ts +++ b/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 { const tools: ToolSet = {}; for (const descriptor of request.tools) { @@ -121,10 +127,17 @@ export class RealModelRegistry implements IModelRegistry { let usage: z.infer = {}; 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,