From ffedb701182d4469a296a1b37027ad444819660c Mon Sep 17 00:00:00 2001 From: devlikebear Date: Mon, 24 Aug 2026 09:12:15 +0900 Subject: [PATCH] feat(llm): add rolling cache breakpoints to anthropic message history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit toAnthropicWireMessages emitted the message array with no cache markers, so every turn paid full input rates for the whole transcript. The API allows four breakpoints and only two were used (system head, last tool). Place up to two message-level markers deterministically on completed turn boundaries: the newest on the last message of the most recent completed turn — making everything before the incoming turn one cacheable prefix — and an older marker one completed turn further back so a warm entry survives when the newest is invalidated. The trailing in-flight group (incoming user message plus any partial tool exchanges) never gets a marker, so placement is stable across tool-loop iterations and never splits a tool_use/tool_result pairing; a defensive guard also refuses to mark any tool_use-bearing content. The message budget is what remains of the provider-wide limit after the system and tool markers, so total breakpoints stay within four for every shape of request. Third-party Anthropic-compatible gateways (e.g. minimax) receive the same standard field; if one rejects it the request errors through the normal error path by design — no special-case retries. Closes #921 --- internal/llm/anthropic.go | 120 ++++++++++++- internal/llm/anthropic_test.go | 297 +++++++++++++++++++++++++++++++++ 2 files changed, 415 insertions(+), 2 deletions(-) diff --git a/internal/llm/anthropic.go b/internal/llm/anthropic.go index a34758ce..906ed453 100644 --- a/internal/llm/anthropic.go +++ b/internal/llm/anthropic.go @@ -15,6 +15,17 @@ import ( const anthropicAPIVersion = "2023-06-01" const anthropicPromptCachingBeta = "prompt-caching-2024-07-31" +const ( + // anthropicMaxCacheBreakpoints is the provider-wide limit on cache_control + // markers per request, across system blocks, tool definitions, and + // messages. + anthropicMaxCacheBreakpoints = 4 + // anthropicMaxMessageCacheBreakpoints caps how many of the remaining + // slots this client spends on the message array. Two are enough for a + // rolling window: the newest completed turn plus one older fallback. + anthropicMaxMessageCacheBreakpoints = 2 +) + type AnthropicClient struct { baseURL string apiKey string @@ -98,10 +109,14 @@ func (c *AnthropicClient) buildChatRequest(messages []ChatMessage, opts ChatOpti nonSystemMessages = append(nonSystemMessages, msg) } + tools := toAnthropicTools(opts.Tools) + wireMessages := toAnthropicWireMessages(nonSystemMessages) + applyAnthropicRollingCacheBreakpoints(wireMessages, nonSystemMessages, len(systemMessages) > 0, len(tools) > 0) + reqBody := map[string]any{ "model": c.model, "max_tokens": c.config.MaxTokens, - "messages": toAnthropicWireMessages(nonSystemMessages), + "messages": wireMessages, } if budget := effectiveThinkingBudget(c.config, opts); budget > 0 { reqBody["thinking"] = map[string]any{ @@ -112,7 +127,7 @@ func (c *AnthropicClient) buildChatRequest(messages []ChatMessage, opts ChatOpti if len(systemMessages) > 0 { reqBody["system"] = toAnthropicSystemBlocks(systemMessages) } - if tools := toAnthropicTools(opts.Tools); len(tools) > 0 { + if len(tools) > 0 { tools[len(tools)-1].CacheControl = map[string]any{"type": "ephemeral"} reqBody["tools"] = tools if choice := toAnthropicToolChoice(opts.ToolChoice); len(choice) > 0 { @@ -153,6 +168,107 @@ func toAnthropicSystemBlocks(systemMessages []string) []map[string]any { return blocks } +// applyAnthropicRollingCacheBreakpoints places cache_control markers on the +// message array so a growing conversation reuses its prefix instead of paying +// full input rates every turn. +// +// Markers land only on completed turns: the newest sits on the last message of +// the most recent completed turn — making everything before the incoming turn +// one cacheable prefix — and an older marker stays two completed turns back so +// a warm entry survives when the newest is invalidated. The trailing group of +// messages (the incoming user message plus any in-flight tool exchanges) never +// gets one: its content still changes within the request cycle, and marking it +// would write a fresh entry per loop iteration without ever being read. +// +// The provider-wide limit counts system and tool markers too, so the message +// budget is what remains of anthropicMaxCacheBreakpoints, capped at +// anthropicMaxMessageCacheBreakpoints. Short histories simply use fewer slots; +// an empty array gets none. +func applyAnthropicRollingCacheBreakpoints(wire []anthropicWireMessage, messages []ChatMessage, hasSystemBlocks bool, hasTools bool) { + if len(wire) == 0 || len(wire) != len(messages) { + return + } + budget := anthropicMaxCacheBreakpoints + if hasSystemBlocks { + budget-- + } + if hasTools { + budget-- + } + if budget > anthropicMaxMessageCacheBreakpoints { + budget = anthropicMaxMessageCacheBreakpoints + } + if budget <= 0 { + return + } + ends := anthropicCompletedTurnEndIndexes(messages) + if len(ends) > budget { + ends = ends[len(ends)-budget:] + } + for _, idx := range ends { + markAnthropicCacheBreakpoint(&wire[idx]) + } +} + +// anthropicCompletedTurnEndIndexes returns the index of the last message of +// each completed turn. A turn starts at a user-initiated message (role "user" +// that is not a tool result); assistant replies and tool results stay inside +// the turn that triggered them. The final group is the in-flight turn — the +// request exists to extend it — so it is excluded. +func anthropicCompletedTurnEndIndexes(messages []ChatMessage) []int { + ends := make([]int, 0, len(messages)) + prevGroupStart := -1 + for i := range messages { + if !anthropicIsUserTurnStart(messages[i]) { + continue + } + if prevGroupStart >= 0 { + ends = append(ends, i-1) + } else if i > 0 { + // Defensive: leading non-user messages form their own + // (completed) prologue group before the first user turn. + ends = append(ends, i-1) + } + prevGroupStart = i + } + return ends +} + +func anthropicIsUserTurnStart(msg ChatMessage) bool { + return msg.Role == "user" && strings.TrimSpace(msg.ToolCallID) == "" +} + +// markAnthropicCacheBreakpoint attaches cache_control to the last content +// block of one wire message, which makes Anthropic treat everything up to and +// including that block as the cached prefix. Plain string content is upgraded +// to a single text block; block arrays get the marker appended to their last +// entry. Messages that would split a tool-call/tool-result pairing (any +// tool_use-bearing content) or carry no markable content are skipped quietly. +func markAnthropicCacheBreakpoint(msg *anthropicWireMessage) { + switch content := msg.Content.(type) { + case string: + if strings.TrimSpace(content) == "" { + return + } + msg.Content = []map[string]any{ + { + "type": "text", + "text": content, + "cache_control": map[string]any{"type": "ephemeral"}, + }, + } + case []map[string]any: + if len(content) == 0 { + return + } + last := content[len(content)-1] + if blockType, _ := last["type"].(string); blockType == "tool_use" { + return + } + last["cache_control"] = map[string]any{"type": "ephemeral"} + } +} + func (c *AnthropicClient) chatNonStreamingResponse(body io.Reader) (ChatResponse, error) { respBody, err := io.ReadAll(body) if err != nil { diff --git a/internal/llm/anthropic_test.go b/internal/llm/anthropic_test.go index 84161145..812e0045 100644 --- a/internal/llm/anthropic_test.go +++ b/internal/llm/anthropic_test.go @@ -381,6 +381,303 @@ func TestToAnthropicSystemBlocks_SingleMessageKeepsWholePromptCached(t *testing. } } +// --- LP-002 rolling message breakpoints --- + +func userMsg(content string) ChatMessage { return ChatMessage{Role: "user", Content: content} } +func assistMsg(content string) ChatMessage { return ChatMessage{Role: "assistant", Content: content} } + +// wireCacheMarkedIndexes returns indexes of wire messages whose LAST content +// block carries cache_control. +func wireCacheMarkedIndexes(t *testing.T, wire []anthropicWireMessage) []int { + t.Helper() + marked := make([]int, 0) + for i, msg := range wire { + if anthropicWireMessageMarked(msg) { + marked = append(marked, i) + } + } + return marked +} + +func anthropicWireMessageMarked(msg anthropicWireMessage) bool { + blocks, ok := msg.Content.([]map[string]any) + if !ok || len(blocks) == 0 { + return false + } + _, marked := blocks[len(blocks)-1]["cache_control"] + return marked +} + +func TestAnthropicCompletedTurnEndIndexes(t *testing.T) { + tests := []struct { + name string + messages []ChatMessage + want []int + }{ + {"empty", nil, nil}, + {"single incoming user message", []ChatMessage{userMsg("hi")}, nil}, + { + "one completed turn plus incoming", + []ChatMessage{userMsg("q1"), assistMsg("r1"), userMsg("q2")}, + []int{1}, + }, + { + "three completed turns plus incoming", + []ChatMessage{userMsg("q1"), assistMsg("r1"), userMsg("q2"), assistMsg("r2"), userMsg("q3"), assistMsg("r3"), userMsg("q4")}, + []int{1, 3, 5}, + }, + { + "tool loop tail stays in-flight", + []ChatMessage{ + userMsg("q1"), assistMsg("r1"), + userMsg("q2"), + {Role: "assistant", ToolCalls: []ToolCall{{ID: "c1", Name: "exec", Arguments: "{}"}}}, + {Role: "tool", ToolCallID: "c1", Content: "out"}, + }, + []int{1}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := anthropicCompletedTurnEndIndexes(tt.messages) + if len(got) != len(tt.want) { + t.Fatalf("got %v want %v", got, tt.want) + } + for i := range got { + if got[i] != tt.want[i] { + t.Fatalf("got %v want %v", got, tt.want) + } + } + }) + } +} + +func TestApplyAnthropicRollingCacheBreakpoints_EmptyHistory(t *testing.T) { + wire := toAnthropicWireMessages(nil) + applyAnthropicRollingCacheBreakpoints(wire, nil, true, true) + if len(wire) != 0 { + t.Fatalf("expected empty wire messages, got %d", len(wire)) + } +} + +func TestApplyAnthropicRollingCacheBreakpoints_SingleIncomingMessage(t *testing.T) { + messages := []ChatMessage{userMsg("hi")} + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages, false, false) + if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 0 { + t.Fatalf("expected no breakpoints on bare history, got %v", marked) + } +} + +func TestApplyAnthropicRollingCacheBreakpoints_ShortHistoryMarksLastCompletedTurn(t *testing.T) { + messages := []ChatMessage{userMsg("q1"), assistMsg("r1"), userMsg("q2")} + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages, false, false) + if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 1 || marked[0] != 1 { + t.Fatalf("expected single breakpoint on the previous assistant reply, got %v", marked) + } + if !anthropicWireMessageMarked(wire[1]) { + t.Fatal("expected breakpoint on message index 1") + } +} + +func TestApplyAnthropicRollingCacheBreakpoints_LongHistoryUsesRollingWindow(t *testing.T) { + messages := []ChatMessage{ + userMsg("q1"), assistMsg("r1"), + userMsg("q2"), assistMsg("r2"), + userMsg("q3"), assistMsg("r3"), + userMsg("q4"), assistMsg("r4"), + userMsg("q5"), + } + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages, true, true) + if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 2 || marked[0] != 5 || marked[1] != 7 { + t.Fatalf("expected rolling breakpoints on the two newest completed turns, got %v", marked) + } + if anthropicWireMessageMarked(wire[8]) { + t.Fatal("the in-flight turn must not be marked") + } +} + +func TestApplyAnthropicRollingCacheBreakpoints_MidToolLoopKeepsStablePlacement(t *testing.T) { + messages := []ChatMessage{ + userMsg("q1"), assistMsg("r1"), + userMsg("q2"), assistMsg("r2"), + userMsg("q3"), + {Role: "assistant", ToolCalls: []ToolCall{{ID: "c1", Name: "exec", Arguments: "{}"}}}, + {Role: "tool", ToolCallID: "c1", Content: "out"}, + } + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages, true, true) + if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 2 || marked[0] != 1 || marked[1] != 3 { + t.Fatalf("expected breakpoints frozen on completed turns, got %v", marked) + } + for _, idx := range []int{4, 5, 6} { + if anthropicWireMessageMarked(wire[idx]) { + t.Fatalf("message %d must stay unmarked during the tool loop", idx) + } + } +} + +// A breakpoint may sit ON a tool_result message (both halves of the exchange +// land inside the cached prefix), but never between an assistant tool_use and +// its matching result. +func TestApplyAnthropicRollingCacheBreakpoints_TurnMayEndOnCompleteToolPair(t *testing.T) { + messages := []ChatMessage{ + userMsg("q1"), + {Role: "assistant", ToolCalls: []ToolCall{{ID: "c1", Name: "exec", Arguments: "{}"}}}, + {Role: "tool", ToolCallID: "c1", Content: "out"}, + userMsg("q2"), + } + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages, false, false) + if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 1 || marked[0] != 2 { + t.Fatalf("expected the breakpoint on the completed tool_result message, got %v", marked) + } + resultBlock, ok := wire[2].Content.([]map[string]any) + if !ok || resultBlock[0]["type"] != "tool_result" { + t.Fatalf("unexpected wire content at index 2: %+v", wire[2].Content) + } + if _, ok := resultBlock[0]["cache_control"]; !ok { + t.Fatalf("expected cache_control on the tool_result block, got %+v", resultBlock[0]) + } +} + +func TestMarkAnthropicCacheBreakpoint_RefusesUnmatchedToolUse(t *testing.T) { + msg := &anthropicWireMessage{ + Role: "assistant", + Content: []map[string]any{ + {"type": "tool_use", "id": "c1", "name": "exec", "input": map[string]any{}}, + }, + } + markAnthropicCacheBreakpoint(msg) + blocks := msg.Content.([]map[string]any) + if _, ok := blocks[len(blocks)-1]["cache_control"]; ok { + t.Fatal("must never place a breakpoint between tool_use and its tool_result") + } +} + +func TestMarkAnthropicCacheBreakpoint_SkipsBlankStringContent(t *testing.T) { + msg := &anthropicWireMessage{Role: "assistant", Content: ""} + markAnthropicCacheBreakpoint(msg) + if text, ok := msg.Content.(string); !ok || text != "" { + t.Fatalf("blank string content must be left untouched, got %#v", msg.Content) + } +} + +func TestAnthropicChat_TotalCacheBreakpointsWithinProviderLimit(t *testing.T) { + var captured map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := json.NewDecoder(r.Body).Decode(&captured); err != nil { + t.Fatalf("decode request: %v", err) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"content":[{"type":"text","text":"ok"}]}`)) + })) + defer srv.Close() + + client, err := NewAnthropicClient(srv.URL, "k", "claude-3-5-haiku-latest", 0) + if err != nil { + t.Fatalf("new client: %v", err) + } + messages := []ChatMessage{ + {Role: "system", Content: "stable body"}, + userMsg("q1"), assistMsg("r1"), + userMsg("q2"), assistMsg("r2"), + userMsg("q3"), assistMsg("r3"), + userMsg("q4"), assistMsg("r4"), + userMsg("q5"), + } + _, err = client.Chat(context.Background(), messages, ChatOptions{ + Tools: []ToolSchema{{ + Type: "function", + Function: ToolFunctionSchema{ + Name: "memory_search", + Description: "search memory", + Parameters: json.RawMessage(`{"type":"object"}`), + }, + }}, + }) + if err != nil { + t.Fatalf("chat: %v", err) + } + + countCacheControl := func(value any) int { + count := 0 + var walk func(any) + walk = func(v any) { + switch typed := v.(type) { + case map[string]any: + if _, ok := typed["cache_control"]; ok { + count++ + } + for _, child := range typed { + walk(child) + } + case []any: + for _, child := range typed { + walk(child) + } + } + } + walk(value) + return count + } + + systemRaw, _ := captured["system"].([]any) + toolsRaw, _ := captured["tools"].([]any) + messagesRaw, _ := captured["messages"].([]any) + total := countCacheControl(systemRaw) + countCacheControl(toolsRaw) + countCacheControl(messagesRaw) + if total > 4 { + t.Fatalf("provider allows at most 4 breakpoints, request carried %d", total) + } + if total != 4 { + t.Fatalf("expected system + tools + two rolling message markers = 4, got %d", total) + } +} + +// Turn N+1's newest message-level breakpoint must cover the whole transcript +// through turn N — everything before the incoming user message becomes one +// cacheable prefix, and its coverage grows as the conversation grows. +func TestAnthropicChat_CacheablePrefixGrowsWithConversation(t *testing.T) { + newestMarkedIndex := func(t *testing.T, history []ChatMessage) int { + t.Helper() + reqBody := buildTestChatRequest(t, history) + messagesRaw, ok := reqBody["messages"].([]anthropicWireMessage) + if !ok { + t.Fatalf("expected wire messages, got %+v", reqBody["messages"]) + } + marked := wireCacheMarkedIndexes(t, messagesRaw) + if len(marked) == 0 { + t.Fatalf("expected message breakpoints for history of %d messages", len(history)) + } + return marked[len(marked)-1] + } + + base := []ChatMessage{userMsg("q1"), assistMsg("r1"), userMsg("q2")} + turnTwoNewest := newestMarkedIndex(t, base) + + longer := append(append([]ChatMessage{}, base...), assistMsg("r2"), userMsg("q3")) + turnThreeNewest := newestMarkedIndex(t, longer) + + if turnThreeNewest <= turnTwoNewest { + t.Fatalf("cacheable prefix must grow with conversation length: turn2=%d turn3=%d", turnTwoNewest, turnThreeNewest) + } + incoming := len(longer) - 1 + if turnThreeNewest >= incoming { + t.Fatalf("newest breakpoint must not cover the incoming turn: newest=%d incoming=%d", turnThreeNewest, incoming) + } +} + +func buildTestChatRequest(t *testing.T, history []ChatMessage) map[string]any { + t.Helper() + client, err := newAnthropicClientWithConfig("https://api.anthropic.com", "k", "claude-3-5-haiku-latest", DefaultClientConfig()) + if err != nil { + t.Fatalf("new client: %v", err) + } + return client.buildChatRequest(history, ChatOptions{}, false) +} + func TestAnthropicChat_EmitsSystemTailOutsideCachedPrefix(t *testing.T) { var captured map[string]any srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {