From 495bb9ac2e922c07a6f51d1ff8bdd47d7fc549ae Mon Sep 17 00:00:00 2001 From: devlikebear Date: Mon, 24 Aug 2026 11:30:25 +0900 Subject: [PATCH] fix(llm): stop rolling cache breakpoints from forfeiting marker slots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up review fixes for #933 (#921). - Fill breakpoint slots newest-first and consume a slot only when a marker actually lands, so a turn that cannot carry one (empty content, trailing tool_use) falls back to an older completed turn instead of being dropped. Previously the budget was truncated before markability was known, so a long history could ship one marker or none. - Replace the unreachable budget arithmetic (hasSystemBlocks/hasTools bools could only ever yield 2) with a reserved-slot count plus a separate anthropicMessageCacheBudget helper, pinned per reservation level by test. - Collapse the two identical branches in anthropicCompletedTurnEndIndexes and drop the prevGroupStart variable that only chose between them. - Route all four cache_control literals through anthropicEphemeralCacheControl. - Correct the doc comment: the fallback marker sits one completed turn back, not two, and the in-flight exclusion is a conservative choice rather than a free one — agent.Loop re-sends the same slice, so a marker there would be read by the next iteration. - Add the CHANGELOG entries #933 never wrote. --- CHANGELOG.md | 4 + internal/llm/anthropic.go | 136 ++++++++++++++++++++++----------- internal/llm/anthropic_test.go | 105 +++++++++++++++++++++++-- 3 files changed, 193 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b46d88da..428cc9b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,10 +8,14 @@ The format is based on Keep a Changelog and the project follows Semantic Version ### Added +- **Anthropic 메시지 배열 rolling cache breakpoint (#921)** — 완료된 턴 경계에 최대 2개의 `cache_control` 마커를 찍어 대화가 길어져도 이전 transcript를 캐시 prefix로 재사용한다. 가장 새 마커는 직전 완료 턴의 마지막 메시지에, 두 번째는 그 앞 턴에 놓여 이전 요청이 이미 데운 위치를 fallback으로 유지한다. 진행 중인 턴(수신 user 메시지 + in-flight tool 교환)에는 찍지 않으므로 tool 루프 반복 사이에 배치가 흔들리지 않고 `tool_use`/`tool_result` 쌍도 갈라지지 않는다. system/tool 마커를 포함한 요청 전체 breakpoint는 프로바이더 상한 4개를 넘지 않는다. + - **`antigravity-cli` provider** — Reuses a locally installed `agy` CLI the way `claude-code-cli` reuses `claude`, via `agy --print --output-format stream-json`. The CLI owns its Google login in the system keyring, so the kind takes no `api_key` and no `base_url` and no credential passes through TARS. Multi-turn resumes through the stream's `conversation_id` (`--conversation`), `ReasoningEffort` maps to `--effort`, and a `json_schema` response format maps to `--json-schema`. Tools remain the CLI's own and are reported on `ChatResponse.ProviderExecutedTools` for audit only, never re-dispatched through TARS' registry. `AGY_CLI_MODE` accepts only `accept-edits` or `plan`; there is deliberately no path to `--dangerously-skip-permissions`. Requires Antigravity CLI 1.1.12 or newer. ### Fixed +- **rolling cache breakpoint가 마커를 조용히 흘리던 문제 (#921)** — 예산은 마킹 가능 여부를 따지기 전에 최신 2개 턴으로 잘려 있어서, 가장 새 턴이 마커를 못 받는 형태(내용이 빈 assistant 메시지, 끝이 `tool_use`인 블록)면 그 슬롯을 더 오래된 턴으로 넘기지 않고 그냥 버렸다. 이제 최신 턴부터 역순으로 훑으며 **실제로 마커가 찍힌 경우에만** 예산을 소모하므로, 마킹 불가능한 턴은 건너뛰고 그 앞 턴이 fallback 자리를 채운다. 아울러 breakpoint 예산 계산에서 실행되지 않던 분기를 걷어내고(`hasSystemBlocks`/`hasTools` bool 두 개 → 예약 슬롯 수 `int` 하나), `anthropicMessageCacheBudget`으로 분리해 예약 수준별로 테스트한다. `cache_control` 리터럴 4곳은 `anthropicEphemeralCacheControl()` 한 곳으로 모았다. + - **프롬프트 캐시가 매 턴 무효화되던 문제 (#920)** — 시스템 프롬프트 첫 줄이 초 단위 wall-clock 타임스탬프라 어떤 프로바이더의 prefix 캐시에도 걸리지 않았고, 정적 본문 전체가 매 턴 write 요금으로 재과금됐다. 시각은 이제 프롬프트 맨 뒤 `## Current Time` 블록으로 내려가 분 단위로 truncate되고, `## Prior Context` 회상도 skills/style/goal/critic 등 세션 고정 섹션 **뒤**로 재배치된다. Anthropic 요청은 system을 블록 배열로 보내 `cache_control` breakpoint를 안정 블록에만 찍으므로 동적 tail이 캐시 prefix를 깨지 않는다. 정렬 규칙은 `prompt.BuildResultFor`에 불변식으로 문서화했다. - **메모리 prefetch가 세션 작업 디렉터리를 지우던 문제 (#920)** — 채팅 메모리 캐시가 조립된 프롬프트를 통째로 저장하는 바람에, work dir을 모르는 prefetch 고루틴의 결과가 캐시에 올라가면 그 다음 턴 프롬프트에서 `## Working Directories` 섹션이 통째로 사라졌다. 이제 캐시는 회상 payload만 보관하고 프롬프트는 항상 live 옵션으로 다시 조립한다 — 캐시 hit과 miss가 구조적으로 동일한 프롬프트를 낸다. diff --git a/internal/llm/anthropic.go b/internal/llm/anthropic.go index 906ed453..8ee62164 100644 --- a/internal/llm/anthropic.go +++ b/internal/llm/anthropic.go @@ -26,6 +26,37 @@ const ( anthropicMaxMessageCacheBreakpoints = 2 ) +// anthropicEphemeralCacheControl builds one cache_control marker. Every +// breakpoint this client emits — system block, tool definition, message — +// goes through here so they cannot drift apart. +// +// It must return a fresh map per call: callers store the result into blocks +// they do not own, and a shared map would let one edit rewrite every +// breakpoint in the request. +func anthropicEphemeralCacheControl() map[string]any { + return map[string]any{"type": "ephemeral"} +} + +// anthropicMessageCacheBudget returns how many cache_control markers the +// message array may carry, given how many the request already spends +// elsewhere. The provider counts system blocks and tool definitions against +// the same per-request limit, so whatever they reserve is unavailable here. +// +// The message array is additionally capped at +// anthropicMaxMessageCacheBreakpoints: a rolling window needs two, and extra +// markers would just buy more cache writes at 1.25x without extending the +// cached prefix any further. +func anthropicMessageCacheBudget(reserved int) int { + budget := anthropicMaxCacheBreakpoints - reserved + if budget > anthropicMaxMessageCacheBreakpoints { + budget = anthropicMaxMessageCacheBreakpoints + } + if budget < 0 { + budget = 0 + } + return budget +} + type AnthropicClient struct { baseURL string apiKey string @@ -111,7 +142,17 @@ func (c *AnthropicClient) buildChatRequest(messages []ChatMessage, opts ChatOpti tools := toAnthropicTools(opts.Tools) wireMessages := toAnthropicWireMessages(nonSystemMessages) - applyAnthropicRollingCacheBreakpoints(wireMessages, nonSystemMessages, len(systemMessages) > 0, len(tools) > 0) + // toAnthropicSystemBlocks marks one block and the tool array marks its + // last entry, so each contributes at most one breakpoint to the + // provider-wide limit. + reservedBreakpoints := 0 + if len(systemMessages) > 0 { + reservedBreakpoints++ + } + if len(tools) > 0 { + reservedBreakpoints++ + } + applyAnthropicRollingCacheBreakpoints(wireMessages, nonSystemMessages, reservedBreakpoints) reqBody := map[string]any{ "model": c.model, @@ -128,7 +169,7 @@ func (c *AnthropicClient) buildChatRequest(messages []ChatMessage, opts ChatOpti reqBody["system"] = toAnthropicSystemBlocks(systemMessages) } if len(tools) > 0 { - tools[len(tools)-1].CacheControl = map[string]any{"type": "ephemeral"} + tools[len(tools)-1].CacheControl = anthropicEphemeralCacheControl() reqBody["tools"] = tools if choice := toAnthropicToolChoice(opts.ToolChoice); len(choice) > 0 { reqBody["tool_choice"] = choice @@ -161,7 +202,7 @@ func toAnthropicSystemBlocks(systemMessages []string) []map[string]any { "text": text, } if i == 0 { - block["cache_control"] = map[string]any{"type": "ephemeral"} + block["cache_control"] = anthropicEphemeralCacheControl() } blocks = append(blocks, block) } @@ -174,39 +215,41 @@ func toAnthropicSystemBlocks(systemMessages []string) []map[string]any { // // 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. +// one cacheable prefix — and the second sits on the turn before it. That +// second marker is the fallback: it is the position the previous request +// marked as its newest, so it is already warm, and it survives a change that +// invalidates only the newer prefix. // -// 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) { +// The trailing group of messages (the incoming user message plus any in-flight +// tool exchanges) never gets one. Note this is a deliberately conservative +// choice, not a free one: agent.Loop appends each tool exchange to the same +// slice and re-sends it, so a marker on a completed in-flight tool_result +// WOULD be read by the next loop iteration. Placing one there is the obvious +// next step for cutting tool-loop cost; it is kept out of scope here so the +// placement rule stays "completed turns only". +// +// Budget comes from anthropicMessageCacheBudget, so a request that already +// spends breakpoints on system blocks and tools simply places fewer here. +// Short histories use fewer slots; an empty array gets none. +// +// Slots are filled newest-first and a slot is only consumed when a marker +// actually lands, so a turn that cannot carry one (see +// markAnthropicCacheBreakpoint) falls through to an older completed turn +// instead of being dropped. +func applyAnthropicRollingCacheBreakpoints(wire []anthropicWireMessage, messages []ChatMessage, reservedBreakpoints int) { 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 { + budget := anthropicMessageCacheBudget(reservedBreakpoints) + if budget == 0 { return } ends := anthropicCompletedTurnEndIndexes(messages) - if len(ends) > budget { - ends = ends[len(ends)-budget:] - } - for _, idx := range ends { - markAnthropicCacheBreakpoint(&wire[idx]) + placed := 0 + for i := len(ends) - 1; i >= 0 && placed < budget; i-- { + if markAnthropicCacheBreakpoint(&wire[ends[i]]) { + placed++ + } } } @@ -217,19 +260,14 @@ func applyAnthropicRollingCacheBreakpoints(wire []anthropicWireMessage, messages // 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. + // Every user turn start closes the group before it. i == 0 has no + // preceding group; any later start closes either a previous turn or + // (if the history opens with assistant/tool messages) the leading + // prologue group. + if i > 0 && anthropicIsUserTurnStart(messages[i]) { ends = append(ends, i-1) } - prevGroupStart = i } return ends } @@ -243,30 +281,36 @@ func anthropicIsUserTurnStart(msg ChatMessage) bool { // 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) { +// tool_use-bearing content) or carry no markable content are skipped. +// +// It reports whether a marker was placed so the caller can spend the freed +// budget on an older turn rather than silently shipping fewer breakpoints. +func markAnthropicCacheBreakpoint(msg *anthropicWireMessage) bool { switch content := msg.Content.(type) { case string: if strings.TrimSpace(content) == "" { - return + return false } msg.Content = []map[string]any{ { "type": "text", "text": content, - "cache_control": map[string]any{"type": "ephemeral"}, + "cache_control": anthropicEphemeralCacheControl(), }, } + return true case []map[string]any: if len(content) == 0 { - return + return false } last := content[len(content)-1] if blockType, _ := last["type"].(string); blockType == "tool_use" { - return + return false } - last["cache_control"] = map[string]any{"type": "ephemeral"} + last["cache_control"] = anthropicEphemeralCacheControl() + return true } + return false } func (c *AnthropicClient) chatNonStreamingResponse(body io.Reader) (ChatResponse, error) { diff --git a/internal/llm/anthropic_test.go b/internal/llm/anthropic_test.go index 812e0045..6b0cdb1c 100644 --- a/internal/llm/anthropic_test.go +++ b/internal/llm/anthropic_test.go @@ -454,16 +454,109 @@ func TestAnthropicCompletedTurnEndIndexes(t *testing.T) { func TestApplyAnthropicRollingCacheBreakpoints_EmptyHistory(t *testing.T) { wire := toAnthropicWireMessages(nil) - applyAnthropicRollingCacheBreakpoints(wire, nil, true, true) + applyAnthropicRollingCacheBreakpoints(wire, nil, 2) if len(wire) != 0 { t.Fatalf("expected empty wire messages, got %d", len(wire)) } } +// The index-alignment guard is the only thing keeping a marker off an +// unrelated message if the wire conversion ever stops being 1:1. +func TestApplyAnthropicRollingCacheBreakpoints_LengthMismatchMarksNothing(t *testing.T) { + messages := []ChatMessage{userMsg("q1"), assistMsg("r1"), userMsg("q2")} + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages[:2], 0) + if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 0 { + t.Fatalf("mismatched lengths must mark nothing, got %v", marked) + } +} + +// anthropicMessageCacheBudget is what keeps the request under the provider's +// total breakpoint limit, so every reservation level needs to be pinned. +func TestAnthropicMessageCacheBudget(t *testing.T) { + tests := []struct { + reserved int + want int + }{ + {0, 2}, + {1, 2}, + {2, 2}, + {3, 1}, + {4, 0}, + {5, 0}, + } + for _, tt := range tests { + if got := anthropicMessageCacheBudget(tt.reserved); got != tt.want { + t.Fatalf("reserved=%d: got budget %d want %d", tt.reserved, got, tt.want) + } + if total := tt.reserved + anthropicMessageCacheBudget(tt.reserved); tt.reserved <= anthropicMaxCacheBreakpoints && total > anthropicMaxCacheBreakpoints { + t.Fatalf("reserved=%d: total %d exceeds provider limit %d", tt.reserved, total, anthropicMaxCacheBreakpoints) + } + } +} + +// A heavily reserved request must spend fewer slots on messages rather than +// blow past the provider limit. +func TestApplyAnthropicRollingCacheBreakpoints_ReservedSlotsShrinkBudget(t *testing.T) { + messages := []ChatMessage{ + userMsg("q1"), assistMsg("r1"), + userMsg("q2"), assistMsg("r2"), + userMsg("q3"), assistMsg("r3"), + userMsg("q4"), + } + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages, 3) + if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 1 || marked[0] != 5 { + t.Fatalf("expected a single breakpoint on the newest completed turn, got %v", marked) + } +} + +// When the newest completed turn cannot carry a marker, the slot must fall +// back to an older markable turn instead of being forfeited. +func TestApplyAnthropicRollingCacheBreakpoints_FallsBackWhenNewestTurnUnmarkable(t *testing.T) { + messages := []ChatMessage{ + userMsg("q1"), assistMsg("r1"), + userMsg("q2"), assistMsg("r2"), + userMsg("q3"), assistMsg(""), // unmarkable: no content to hang cache_control on + userMsg("q4"), + } + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages, 2) + marked := wireCacheMarkedIndexes(t, wire) + if len(marked) != 2 || marked[0] != 1 || marked[1] != 3 { + t.Fatalf("expected the budget to fall back to older markable turns, got %v", marked) + } +} + +// The point of the second marker is that the previous turn's newest +// breakpoint — already warm — is retained as the fallback on the next turn. +func TestApplyAnthropicRollingCacheBreakpoints_WindowRollsAcrossTurns(t *testing.T) { + markersFor := func(messages []ChatMessage) []int { + wire := toAnthropicWireMessages(messages) + applyAnthropicRollingCacheBreakpoints(wire, messages, 2) + return wireCacheMarkedIndexes(t, wire) + } + + turnA := []ChatMessage{userMsg("q1"), assistMsg("r1"), userMsg("q2"), assistMsg("r2"), userMsg("q3")} + turnB := append(append([]ChatMessage{}, turnA...), assistMsg("r3"), userMsg("q4")) + + markedA := markersFor(turnA) + markedB := markersFor(turnB) + if len(markedA) != 2 || len(markedB) != 2 { + t.Fatalf("expected two markers on both turns, got %v and %v", markedA, markedB) + } + if markedB[0] != markedA[1] { + t.Fatalf("turn B's fallback (%d) must reuse turn A's newest breakpoint (%d) so it is already warm", markedB[0], markedA[1]) + } + if markedB[1] <= markedA[1] { + t.Fatalf("turn B's newest breakpoint (%d) must advance past turn A's (%d)", markedB[1], markedA[1]) + } +} + func TestApplyAnthropicRollingCacheBreakpoints_SingleIncomingMessage(t *testing.T) { messages := []ChatMessage{userMsg("hi")} wire := toAnthropicWireMessages(messages) - applyAnthropicRollingCacheBreakpoints(wire, messages, false, false) + applyAnthropicRollingCacheBreakpoints(wire, messages, 0) if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 0 { t.Fatalf("expected no breakpoints on bare history, got %v", marked) } @@ -472,7 +565,7 @@ func TestApplyAnthropicRollingCacheBreakpoints_SingleIncomingMessage(t *testing. func TestApplyAnthropicRollingCacheBreakpoints_ShortHistoryMarksLastCompletedTurn(t *testing.T) { messages := []ChatMessage{userMsg("q1"), assistMsg("r1"), userMsg("q2")} wire := toAnthropicWireMessages(messages) - applyAnthropicRollingCacheBreakpoints(wire, messages, false, false) + applyAnthropicRollingCacheBreakpoints(wire, messages, 0) if marked := wireCacheMarkedIndexes(t, wire); len(marked) != 1 || marked[0] != 1 { t.Fatalf("expected single breakpoint on the previous assistant reply, got %v", marked) } @@ -490,7 +583,7 @@ func TestApplyAnthropicRollingCacheBreakpoints_LongHistoryUsesRollingWindow(t *t userMsg("q5"), } wire := toAnthropicWireMessages(messages) - applyAnthropicRollingCacheBreakpoints(wire, messages, true, true) + applyAnthropicRollingCacheBreakpoints(wire, messages, 2) 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) } @@ -508,7 +601,7 @@ func TestApplyAnthropicRollingCacheBreakpoints_MidToolLoopKeepsStablePlacement(t {Role: "tool", ToolCallID: "c1", Content: "out"}, } wire := toAnthropicWireMessages(messages) - applyAnthropicRollingCacheBreakpoints(wire, messages, true, true) + applyAnthropicRollingCacheBreakpoints(wire, messages, 2) 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) } @@ -530,7 +623,7 @@ func TestApplyAnthropicRollingCacheBreakpoints_TurnMayEndOnCompleteToolPair(t *t userMsg("q2"), } wire := toAnthropicWireMessages(messages) - applyAnthropicRollingCacheBreakpoints(wire, messages, false, false) + applyAnthropicRollingCacheBreakpoints(wire, messages, 0) 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) }