diff --git a/src/provider/anthropic-cache-breakpoint.test.ts b/src/provider/anthropic-cache-breakpoint.test.ts new file mode 100644 index 000000000..b56e7186d --- /dev/null +++ b/src/provider/anthropic-cache-breakpoint.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, test } from "bun:test"; +import type { AdapterRegistry } from "@intx/inference"; +import { createBuiltinRegistry } from "@intx/inference/providers"; +import type { + ConversationTurn, + InferenceOptions, + LastCycleSource, +} from "@intx/types/runtime"; +import { withAnthropicCacheBreakpoint } from "./anthropic-cache-breakpoint.js"; +import { createOpenCodeGoAnthropicAdapter } from "./opencode-go-anthropic-adapter.js"; +import { createZenAnthropicAdapter } from "./zen-anthropic-adapter.js"; + +function sourceFor(provider: string): LastCycleSource { + return { sourceId: `test-${provider}`, provider, model: "test-model" }; +} + +const inner: AdapterRegistry = { + has: (provider) => createBuiltinRegistry().has(provider), + resolve: (source) => { + if (source.provider === "zen-messages") { + return createZenAnthropicAdapter(source); + } + if (source.provider === "opencode-go-messages") { + return createOpenCodeGoAnthropicAdapter(source); + } + return createBuiltinRegistry().resolve(source); + }, +}; + +const adapters = withAnthropicCacheBreakpoint(inner); + +function userTurn(text: string): ConversationTurn { + return { + role: "user", + timestamp: 0, + content: [{ type: "text", text }], + }; +} + +function assistantTurn(text: string): ConversationTurn { + return { + role: "assistant", + timestamp: 0, + content: [{ type: "text", text }], + }; +} + +type WireBlock = { + cache_control?: unknown; + text?: string; + name?: string; +}; + +type WireBody = { + messages: { role: string; content: WireBlock[] }[]; + system?: WireBlock[]; + tools?: WireBlock[]; +}; + +function wireBody(body: string): WireBody { + return JSON.parse(body) as WireBody; +} + +function build(provider: string, options: InferenceOptions): WireBody { + const persisted = [userTurn("q1"), assistantTurn("a1"), userTurn("q2")]; + const nudge = userTurn("wrap up soon"); + const request = adapters + .resolve(sourceFor(provider)) + .buildRequest([...persisted, nudge], "test-model", { + ...options, + ephemeralTurns: [nudge], + } as InferenceOptions); + return wireBody(request.body); +} + +describe("anthropic cache breakpoint with ephemeral turns", () => { + for (const provider of [ + "anthropic", + "zen-messages", + "opencode-go-messages", + ]) { + test(`${provider}: breakpoint lands on the last persisted user turn, not the ephemeral tail`, () => { + const body = build(provider, {}); + + expect(body.messages).toHaveLength(4); + expect(body.messages[3]?.content[0]?.text).toBe("wrap up soon"); + expect( + body.messages[2]?.content[body.messages[2].content.length - 1] + ?.cache_control, + ).toEqual({ type: "ephemeral" }); + expect( + body.messages[3]?.content.filter( + (block) => block.cache_control !== undefined, + ), + ).toEqual([]); + }); + } + + test("system and tools breakpoints stay untouched while the nudge is attached", () => { + const persisted = [ + { ...userTurn("sys"), role: "system" as const }, + userTurn("q1"), + assistantTurn("a1"), + userTurn("q2"), + ]; + const nudge = userTurn("wrap up soon"); + const request = adapters + .resolve(sourceFor("anthropic")) + .buildRequest([...persisted, nudge], "test-model", { + ephemeralTurns: [nudge], + tools: [{ name: "run_shell", description: "run", inputSchema: {} }], + } as InferenceOptions); + const body = wireBody(request.body); + + expect(body.system?.[0]?.cache_control).toEqual({ type: "ephemeral" }); + expect(body.tools?.[body.tools.length - 1]?.cache_control).toEqual({ + type: "ephemeral", + }); + expect( + body.messages[2]?.content[body.messages[2].content.length - 1] + ?.cache_control, + ).toEqual({ type: "ephemeral" }); + expect( + body.messages[3]?.content.filter( + (block) => block.cache_control !== undefined, + ), + ).toEqual([]); + }); + + test("without ephemeral turns the request is byte-identical to the base adapter", () => { + const turns = [userTurn("q1"), assistantTurn("a1"), userTurn("q2")]; + const base = inner + .resolve(sourceFor("anthropic")) + .buildRequest(turns, "test-model", {}); + const wrapped = adapters + .resolve(sourceFor("anthropic")) + .buildRequest(turns, "test-model", {}); + expect(wrapped.body).toBe(base.body); + }); + + test("non-anthropic providers pass through untouched", () => { + const persisted = [userTurn("q1"), assistantTurn("a1"), userTurn("q2")]; + const nudge = userTurn("wrap up soon"); + const options = { ephemeralTurns: [nudge] } as InferenceOptions; + const base = inner + .resolve(sourceFor("openai")) + .buildRequest([...persisted, nudge], "test-model", options); + const wrapped = adapters + .resolve(sourceFor("openai")) + .buildRequest([...persisted, nudge], "test-model", options); + expect(wrapped.body).toBe(base.body); + expect(wrapped.body).not.toContain("cache_control"); + }); +}); diff --git a/src/provider/anthropic-cache-breakpoint.ts b/src/provider/anthropic-cache-breakpoint.ts new file mode 100644 index 000000000..5a3b55fd2 --- /dev/null +++ b/src/provider/anthropic-cache-breakpoint.ts @@ -0,0 +1,100 @@ +import type { + AdapterRegistry, + BuiltRequest, + ExtendedInferenceOptions, + ProviderAdapter, +} from "@intx/inference"; +import { OPENCODE_GO_MESSAGES_PROVIDER } from "./opencode-go-anthropic-adapter.js"; +import { ZEN_MESSAGES_PROVIDER } from "./zen-anthropic-adapter.js"; + +const ANTHROPIC_MESSAGES_PROVIDERS: ReadonlySet = new Set([ + "anthropic", + ZEN_MESSAGES_PROVIDER, + OPENCODE_GO_MESSAGES_PROVIDER, +]); + +type WireBlock = Record; + +type WireMessage = { + role?: unknown; + content?: unknown; +}; + +function ephemeralSuffixLength(options: ExtendedInferenceOptions): number { + const turns = options.ephemeralTurns; + if (turns === undefined) return 0; + return turns.filter((turn) => turn.role !== "system").length; +} + +function isWireBlock(block: unknown): block is WireBlock { + return typeof block === "object" && block !== null; +} + +function stripBreakpoints(messages: WireMessage[], from: number): void { + for (const message of messages.slice(from)) { + if (!Array.isArray(message.content)) continue; + for (const block of message.content) { + if (isWireBlock(block)) delete block.cache_control; + } + } +} + +function placeBreakpoint(messages: WireMessage[], before: number): boolean { + for (let index = before - 1; index >= 0; index -= 1) { + const message = messages[index]; + if (message?.role === "assistant") continue; + if (!Array.isArray(message?.content) || message.content.length === 0) { + continue; + } + const last = message.content[message.content.length - 1]; + if (!isWireBlock(last)) continue; + last.cache_control = { type: "ephemeral" }; + return true; + } + return false; +} + +function moveBreakpointToPersistedTail( + built: BuiltRequest, + options: ExtendedInferenceOptions, +): BuiltRequest { + const suffix = ephemeralSuffixLength(options); + if (suffix === 0) return built; + let body: Record; + try { + body = JSON.parse(built.body) as Record; + } catch { + return built; + } + if (!Array.isArray(body.messages)) return built; + const messages = body.messages as WireMessage[]; + const split = messages.length - suffix; + if (split <= 0) return built; + stripBreakpoints(messages, split); + if (!placeBreakpoint(messages, split)) return built; + return { ...built, body: JSON.stringify(body) }; +} + +function withMovedBreakpoint(adapter: ProviderAdapter): ProviderAdapter { + return { + ...adapter, + buildRequest: (turns, model, options) => + moveBreakpointToPersistedTail( + adapter.buildRequest(turns, model, options), + options, + ), + }; +} + +export function withAnthropicCacheBreakpoint( + adapters: AdapterRegistry, +): AdapterRegistry { + return { + has: (provider) => adapters.has(provider), + resolve(source, quirks) { + const adapter = adapters.resolve(source, quirks); + if (!ANTHROPIC_MESSAGES_PROVIDERS.has(source.provider)) return adapter; + return withMovedBreakpoint(adapter); + }, + }; +} diff --git a/src/provider/inference-dependencies.ts b/src/provider/inference-dependencies.ts index d1be4a7c4..95cd82380 100644 --- a/src/provider/inference-dependencies.ts +++ b/src/provider/inference-dependencies.ts @@ -17,6 +17,7 @@ import { withCodexContentTypeRepair, } from "./codex-responses.js"; import { GROK_RESPONSES_PROVIDER } from "./grok-responses.js"; +import { withAnthropicCacheBreakpoint } from "./anthropic-cache-breakpoint.js"; import { withReplaySanitizer } from "./replay-sanitizer.js"; import { isPollOnlyPendingBatch } from "../subagent/poll-exempt.js"; import { OPENCODE_GO_PROVIDER_ID } from "../../packages/opencode-go/src/index.js"; @@ -95,6 +96,7 @@ export function createInferenceDependencies(): Promise { import: (specifier) => Promise.resolve(localModules[specifier]), }) .then(withReplaySanitizer) + .then(withAnthropicCacheBreakpoint) .then(createDependencies) .then((deps) => ({ ...deps,