fix(provider): keep the Anthropic cache breakpoint off ephemeral turns

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");
+  });
+});

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<string> = new Set([
+  "anthropic",
+  ZEN_MESSAGES_PROVIDER,
+  OPENCODE_GO_MESSAGES_PROVIDER,
+]);
+
+type WireBlock = Record<string, unknown>;
+
+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<string, unknown>;
+  try {
+    body = JSON.parse(built.body) as Record<string, unknown>;
+  } 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);
+    },
+  };
+}

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<Dependencies> {
       import: (specifier) => Promise.resolve(localModules[specifier]),
     })
       .then(withReplaySanitizer)
+      .then(withAnthropicCacheBreakpoint)
       .then(createDependencies)
       .then((deps) => ({
         ...deps,