fix(llm): stop rolling cache breakpoints from forfeiting marker slots

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 <text> --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가 구조적으로 동일한 프롬프트를 낸다.
 

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) {

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)
 	}