Skip to content

Commit d3d8465

Browse files
TheGreatAxiosTheGreatAxios
authored
Merge pull request #1008 from corbitsdev/cl-6913-anthropic-conversation-cache-breakpoint-lands-on-ephemeral
fix(provider): keep the Anthropic cache breakpoint off ephemeral turns
2 parents 3eff6f9 + 09d3b96 commit d3d8465

3 files changed

Lines changed: 256 additions & 0 deletions

File tree

‎src/provider/anthropic-cache-breakpoint.test.ts‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
import { describe, expect, test } from "bun:test";
2+
import type { AdapterRegistry } from "@intx/inference";
3+
import { createBuiltinRegistry } from "@intx/inference/providers";
4+
import type {
5+
ConversationTurn,
6+
InferenceOptions,
7+
LastCycleSource,
8+
} from "@intx/types/runtime";
9+
import { withAnthropicCacheBreakpoint } from "./anthropic-cache-breakpoint.js";
10+
import { createOpenCodeGoAnthropicAdapter } from "./opencode-go-anthropic-adapter.js";
11+
import { createZenAnthropicAdapter } from "./zen-anthropic-adapter.js";
12+
13+
function sourceFor(provider: string): LastCycleSource {
14+
return { sourceId: `test-${provider}`, provider, model: "test-model" };
15+
}
16+
17+
const inner: AdapterRegistry = {
18+
has: (provider) => createBuiltinRegistry().has(provider),
19+
resolve: (source) => {
20+
if (source.provider === "zen-messages") {
21+
return createZenAnthropicAdapter(source);
22+
}
23+
if (source.provider === "opencode-go-messages") {
24+
return createOpenCodeGoAnthropicAdapter(source);
25+
}
26+
return createBuiltinRegistry().resolve(source);
27+
},
28+
};
29+
30+
const adapters = withAnthropicCacheBreakpoint(inner);
31+
32+
function userTurn(text: string): ConversationTurn {
33+
return {
34+
role: "user",
35+
timestamp: 0,
36+
content: [{ type: "text", text }],
37+
};
38+
}
39+
40+
function assistantTurn(text: string): ConversationTurn {
41+
return {
42+
role: "assistant",
43+
timestamp: 0,
44+
content: [{ type: "text", text }],
45+
};
46+
}
47+
48+
type WireBlock = {
49+
cache_control?: unknown;
50+
text?: string;
51+
name?: string;
52+
};
53+
54+
type WireBody = {
55+
messages: { role: string; content: WireBlock[] }[];
56+
system?: WireBlock[];
57+
tools?: WireBlock[];
58+
};
59+
60+
function wireBody(body: string): WireBody {
61+
return JSON.parse(body) as WireBody;
62+
}
63+
64+
function build(provider: string, options: InferenceOptions): WireBody {
65+
const persisted = [userTurn("q1"), assistantTurn("a1"), userTurn("q2")];
66+
const nudge = userTurn("wrap up soon");
67+
const request = adapters
68+
.resolve(sourceFor(provider))
69+
.buildRequest([...persisted, nudge], "test-model", {
70+
...options,
71+
ephemeralTurns: [nudge],
72+
} as InferenceOptions);
73+
return wireBody(request.body);
74+
}
75+
76+
describe("anthropic cache breakpoint with ephemeral turns", () => {
77+
for (const provider of [
78+
"anthropic",
79+
"zen-messages",
80+
"opencode-go-messages",
81+
]) {
82+
test(`${provider}: breakpoint lands on the last persisted user turn, not the ephemeral tail`, () => {
83+
const body = build(provider, {});
84+
85+
expect(body.messages).toHaveLength(4);
86+
expect(body.messages[3]?.content[0]?.text).toBe("wrap up soon");
87+
expect(
88+
body.messages[2]?.content[body.messages[2].content.length - 1]
89+
?.cache_control,
90+
).toEqual({ type: "ephemeral" });
91+
expect(
92+
body.messages[3]?.content.filter(
93+
(block) => block.cache_control !== undefined,
94+
),
95+
).toEqual([]);
96+
});
97+
}
98+
99+
test("system and tools breakpoints stay untouched while the nudge is attached", () => {
100+
const persisted = [
101+
{ ...userTurn("sys"), role: "system" as const },
102+
userTurn("q1"),
103+
assistantTurn("a1"),
104+
userTurn("q2"),
105+
];
106+
const nudge = userTurn("wrap up soon");
107+
const request = adapters
108+
.resolve(sourceFor("anthropic"))
109+
.buildRequest([...persisted, nudge], "test-model", {
110+
ephemeralTurns: [nudge],
111+
tools: [{ name: "run_shell", description: "run", inputSchema: {} }],
112+
} as InferenceOptions);
113+
const body = wireBody(request.body);
114+
115+
expect(body.system?.[0]?.cache_control).toEqual({ type: "ephemeral" });
116+
expect(body.tools?.[body.tools.length - 1]?.cache_control).toEqual({
117+
type: "ephemeral",
118+
});
119+
expect(
120+
body.messages[2]?.content[body.messages[2].content.length - 1]
121+
?.cache_control,
122+
).toEqual({ type: "ephemeral" });
123+
expect(
124+
body.messages[3]?.content.filter(
125+
(block) => block.cache_control !== undefined,
126+
),
127+
).toEqual([]);
128+
});
129+
130+
test("without ephemeral turns the request is byte-identical to the base adapter", () => {
131+
const turns = [userTurn("q1"), assistantTurn("a1"), userTurn("q2")];
132+
const base = inner
133+
.resolve(sourceFor("anthropic"))
134+
.buildRequest(turns, "test-model", {});
135+
const wrapped = adapters
136+
.resolve(sourceFor("anthropic"))
137+
.buildRequest(turns, "test-model", {});
138+
expect(wrapped.body).toBe(base.body);
139+
});
140+
141+
test("non-anthropic providers pass through untouched", () => {
142+
const persisted = [userTurn("q1"), assistantTurn("a1"), userTurn("q2")];
143+
const nudge = userTurn("wrap up soon");
144+
const options = { ephemeralTurns: [nudge] } as InferenceOptions;
145+
const base = inner
146+
.resolve(sourceFor("openai"))
147+
.buildRequest([...persisted, nudge], "test-model", options);
148+
const wrapped = adapters
149+
.resolve(sourceFor("openai"))
150+
.buildRequest([...persisted, nudge], "test-model", options);
151+
expect(wrapped.body).toBe(base.body);
152+
expect(wrapped.body).not.toContain("cache_control");
153+
});
154+
});

‎src/provider/anthropic-cache-breakpoint.ts‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
import type {
2+
AdapterRegistry,
3+
BuiltRequest,
4+
ExtendedInferenceOptions,
5+
ProviderAdapter,
6+
} from "@intx/inference";
7+
import { OPENCODE_GO_MESSAGES_PROVIDER } from "./opencode-go-anthropic-adapter.js";
8+
import { ZEN_MESSAGES_PROVIDER } from "./zen-anthropic-adapter.js";
9+
10+
const ANTHROPIC_MESSAGES_PROVIDERS: ReadonlySet<string> = new Set([
11+
"anthropic",
12+
ZEN_MESSAGES_PROVIDER,
13+
OPENCODE_GO_MESSAGES_PROVIDER,
14+
]);
15+
16+
type WireBlock = Record<string, unknown>;
17+
18+
type WireMessage = {
19+
role?: unknown;
20+
content?: unknown;
21+
};
22+
23+
function ephemeralSuffixLength(options: ExtendedInferenceOptions): number {
24+
const turns = options.ephemeralTurns;
25+
if (turns === undefined) return 0;
26+
return turns.filter((turn) => turn.role !== "system").length;
27+
}
28+
29+
function isWireBlock(block: unknown): block is WireBlock {
30+
return typeof block === "object" && block !== null;
31+
}
32+
33+
function stripBreakpoints(messages: WireMessage[], from: number): void {
34+
for (const message of messages.slice(from)) {
35+
if (!Array.isArray(message.content)) continue;
36+
for (const block of message.content) {
37+
if (isWireBlock(block)) delete block.cache_control;
38+
}
39+
}
40+
}
41+
42+
function placeBreakpoint(messages: WireMessage[], before: number): boolean {
43+
for (let index = before - 1; index >= 0; index -= 1) {
44+
const message = messages[index];
45+
if (message?.role === "assistant") continue;
46+
if (!Array.isArray(message?.content) || message.content.length === 0) {
47+
continue;
48+
}
49+
const last = message.content[message.content.length - 1];
50+
if (!isWireBlock(last)) continue;
51+
last.cache_control = { type: "ephemeral" };
52+
return true;
53+
}
54+
return false;
55+
}
56+
57+
function moveBreakpointToPersistedTail(
58+
built: BuiltRequest,
59+
options: ExtendedInferenceOptions,
60+
): BuiltRequest {
61+
const suffix = ephemeralSuffixLength(options);
62+
if (suffix === 0) return built;
63+
let body: Record<string, unknown>;
64+
try {
65+
body = JSON.parse(built.body) as Record<string, unknown>;
66+
} catch {
67+
return built;
68+
}
69+
if (!Array.isArray(body.messages)) return built;
70+
const messages = body.messages as WireMessage[];
71+
const split = messages.length - suffix;
72+
if (split <= 0) return built;
73+
stripBreakpoints(messages, split);
74+
if (!placeBreakpoint(messages, split)) return built;
75+
return { ...built, body: JSON.stringify(body) };
76+
}
77+
78+
function withMovedBreakpoint(adapter: ProviderAdapter): ProviderAdapter {
79+
return {
80+
...adapter,
81+
buildRequest: (turns, model, options) =>
82+
moveBreakpointToPersistedTail(
83+
adapter.buildRequest(turns, model, options),
84+
options,
85+
),
86+
};
87+
}
88+
89+
export function withAnthropicCacheBreakpoint(
90+
adapters: AdapterRegistry,
91+
): AdapterRegistry {
92+
return {
93+
has: (provider) => adapters.has(provider),
94+
resolve(source, quirks) {
95+
const adapter = adapters.resolve(source, quirks);
96+
if (!ANTHROPIC_MESSAGES_PROVIDERS.has(source.provider)) return adapter;
97+
return withMovedBreakpoint(adapter);
98+
},
99+
};
100+
}

‎src/provider/inference-dependencies.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import {
1717
withCodexContentTypeRepair,
1818
} from "./codex-responses.js";
1919
import { GROK_RESPONSES_PROVIDER } from "./grok-responses.js";
20+
import { withAnthropicCacheBreakpoint } from "./anthropic-cache-breakpoint.js";
2021
import { withReplaySanitizer } from "./replay-sanitizer.js";
2122
import { isPollOnlyPendingBatch } from "../subagent/poll-exempt.js";
2223
import { OPENCODE_GO_PROVIDER_ID } from "../../packages/opencode-go/src/index.js";
@@ -95,6 +96,7 @@ export function createInferenceDependencies(): Promise<Dependencies> {
9596
import: (specifier) => Promise.resolve(localModules[specifier]),
9697
})
9798
.then(withReplaySanitizer)
99+
.then(withAnthropicCacheBreakpoint)
98100
.then(createDependencies)
99101
.then((deps) => ({
100102
...deps,

0 commit comments

Comments
 (0)