From 7cd563ba3c686be4427e4fdfd45d2d6f03dc32d4 Mon Sep 17 00:00:00 2001 From: Osher-Elhadad Date: Mon, 10 Aug 2026 03:07:17 +0000 Subject: [PATCH 1/7] fix(cache): slide the store TTL, pin frozen decisions, and repair lost ones MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A frozen compaction could expire mid-task and hand the provider a DIFFERENT representation of an already-cached message, forcing a re-write of the whole suffix at 11.5x the cache-read price. Store.Get refreshed LRU recency but never e.expires, and the default TTL was 1800s — under Terminal-Bench's ~1975s mean wall clock, so a decision died roughly every 69 turns however often it was replayed. Three changes, one per failure mode: - Sliding TTL: Get refreshes expires. The TTL reclaims state for FINISHED sessions; a decision an ongoing session replays every turn is by definition live. An entry nobody reads still expires on its original deadline. - Default TTL 1800s -> 10000s (store.DefaultTTL), still ttl_seconds-configurable. Steady-state memory is unchanged: the 1000-entry cap, not the TTL, bounds it. - cg:frz: entries are pinned against LRU eviction, capped at half the entry cap so one session cannot pin the cache and starve the rewind stashes expand needs. The third part is the design question. A Get miss cannot distinguish "no frozen decision" from "the decision existed and was lost", and the two want OPPOSITE behavior: fail-open means forwarding the original, but once the provider has cached the compacted bytes, forwarding the original IS the destructive act. So the store keeps the FACT of a dropped freeze (store.FrozenLoser, a bounded key set — only the knowledge has to survive, not the payload) and mask/failed_run lift the depth restriction for exactly those keys. Re-deriving is safe because an offloader's replacement is a pure function of (content, config) and the marker key is sha256(original), so it reproduces the same bytes the provider cached and re-establishes the freeze; the never-worse and kept-verbatim guards still apply, so nothing new is ever dropped. Observability, without which the fix is unverifiable on a benchmark run: /stats gains frozen_hits, frozen_misses, frozen_dropped, frozen_repaired and frozen_flips (= dropped - repaired, the drops that actually cost a cache-write). Fields are added only — deploy/harbor parses this endpoint. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: Osher-Elhadad --- components/offload/failed_run.go | 8 +- components/offload/freeze_test.go | 153 ++++++++++++++++++++++++++++++ components/offload/mask.go | 8 +- components/offload/state.go | 52 +++++++++- docs/design.md | 44 ++++++++- docs/how-to/recover-context.md | 4 +- docs/reference/config.md | 4 +- metrics/metrics.go | 14 +++ proxy/proxy.go | 7 ++ store/store.go | 136 +++++++++++++++++++++++--- store/store_test.go | 125 ++++++++++++++++++++++++ 11 files changed, 533 insertions(+), 22 deletions(-) create mode 100644 components/offload/freeze_test.go diff --git a/components/offload/failed_run.go b/components/offload/failed_run.go index 690a58d..b965b47 100644 --- a/components/offload/failed_run.go +++ b/components/offload/failed_run.go @@ -110,8 +110,12 @@ func (fr *FailedRun) Offload(req *schemas.BifrostChatRequest, rep *components.Re // transitions on SWE-50). On a cached agent the superseded run already bills at // the cheap cache-read rate, so collapsing it doesn't pay: skip NEW collapses // entirely (frozen ones are still reapplied above for stability). With caching - // OFF, collapse freely — there the content cut is a direct saving. - if c.CacheAware { + // OFF, collapse freely — there the content cut is a direct saving. The one + // exception is a freeze this session established and the store then LOST: the + // provider already holds the collapsed bytes for this run, so re-deriving them + // (deterministic) preserves its cache, while leaving the run verbatim is what + // forces the suffix re-write. + if c.CacheAware && !repairLostFreeze(c, fr.Name(), content) { continue } newText, key, eff, ok := tryMark(c, fr.mode, content, " [full output: call "+expand.ToolName+"]", diff --git a/components/offload/freeze_test.go b/components/offload/freeze_test.go new file mode 100644 index 0000000..ef95e04 --- /dev/null +++ b/components/offload/freeze_test.go @@ -0,0 +1,153 @@ +package offload + +import ( + "strings" + "testing" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" +) + +func tool(s string) bschemas.ChatMessage { + t := s + return bschemas.ChatMessage{Role: bschemas.ChatMessageRoleTool, + Content: &bschemas.ChatMessageContent{ContentStr: &t}} +} + +// maskFor builds a mask offloader with a low floor so short fixtures still qualify. +func maskFor(t *testing.T) *Mask { + t.Helper() + comp, err := newMask([]byte("keep_recent: 1\nmin_tokens: 5\n")) + if err != nil { + t.Fatal(err) + } + return comp.(*Mask) +} + +// A long session replays a frozen mask on every turn. With a write-only TTL the +// decision died mid-session and the message flipped masked→full inside the provider's +// cached prefix; with the sliding TTL it must stay byte-identical for the whole run. +func TestFrozenMaskSurvivesLongSession(t *testing.T) { + now := time.Unix(0, 0) + st := store.NewMemory(store.Options{TTLSeconds: 10}) + st.SetClock(func() time.Time { return now }) + m := maskFor(t) + body := strings.Repeat("verbose tool output line\n", 30) + + var first string + for turn := 0; turn < 200; turn++ { + // Every turn the agent re-sends the ORIGINAL history plus a new tail message. + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{ + tool(body), tool("new tail output"), + }} + // Turn 0 is the tail turn (nothing cached yet) — that is where a NEW mask is + // allowed. Every later turn has the output in the already-cached prefix, so only + // the frozen replay can keep it masked. + maxCached := 0 + if turn == 0 { + maxCached = -1 + } + c := &components.Ctx{Session: "s", Store: st, CacheAware: true, MaxCachedIdx: maxCached} + var rep components.Report + if _, err := m.Offload(req, &rep, c); err != nil { + t.Fatal(err) + } + got := schema.MessageText(req.Input[0]) + if turn == 0 { + if got == body { + t.Fatal("turn 0 must mask the older output") + } + first = got + } + if got != first { + t.Fatalf("turn %d flipped representation (cache-destructive):\n want %q\n got %q", + turn, first, got) + } + now = now.Add(9 * time.Second) // ~9s/turn, 200 turns = 1800s ≫ the 10s TTL + } +} + +// TestFrozenMaskNewDecisionStillTailGated: the repair path must not become a general +// license to mutate at depth — content that was NEVER frozen stays verbatim in the +// cached prefix. +func TestNewMaskStillTailGated(t *testing.T) { + st := store.NewMemory(store.Options{}) + m := maskFor(t) + body := strings.Repeat("verbose tool output line\n", 30) + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{tool(body), tool("tail")}} + c := &components.Ctx{Session: "s", Store: st, CacheAware: true, MaxCachedIdx: 0} + var rep components.Report + if _, err := m.Offload(req, &rep, c); err != nil { + t.Fatal(err) + } + if schema.MessageText(req.Input[0]) != body { + t.Fatal("a never-frozen message inside the cached prefix must stay verbatim") + } +} + +// The core of part C: when the store LOSES an established frozen decision, the provider +// still holds the masked bytes. Re-deriving them is cache-preserving; leaving the message +// verbatim is the destructive flip. So a forced store miss must not flip representation. +func TestForcedStoreMissDoesNotFlipEstablishedCompaction(t *testing.T) { + now := time.Unix(0, 0) + st := store.NewMemory(store.Options{TTLSeconds: 10}) + st.SetClock(func() time.Time { return now }) + m := maskFor(t) + body := strings.Repeat("verbose tool output line\n", 30) + + run := func(maxCached int) string { + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{tool(body), tool("tail")}} + c := &components.Ctx{Session: "s", Store: st, CacheAware: true, MaxCachedIdx: maxCached} + var rep components.Report + if _, err := m.Offload(req, &rep, c); err != nil { + t.Fatal(err) + } + return schema.MessageText(req.Input[0]) + } + + masked := run(-1) // turn 1: the output is in the tail, so it gets masked and frozen + if masked == body { + t.Fatal("turn 1 must mask") + } + + // Force the loss: nothing reads the entry for longer than the TTL. From here on the + // message sits in the CACHED PREFIX (MaxCachedIdx=0), which the tail gate forbids + // mutating — the exact situation that used to revert masked→full. + now = now.Add(11 * time.Second) + if got := run(0); got != masked { + t.Fatalf("a lost freeze flipped an established compaction:\n want %q\n got %q", masked, got) + } + // And the repair is recorded as a repair, not as an unrepaired flip. + dropped, repaired := st.FrozenLossStats() + if dropped == 0 || dropped != repaired { + t.Fatalf("want every dropped decision repaired, got dropped=%d repaired=%d", dropped, repaired) + } +} + +// A Nop store (no FrozenLoser) must degrade to the legacy behavior rather than panic or +// mutate at depth on every message. +func TestRepairLostFreezeNoopStore(t *testing.T) { + c := &components.Ctx{Session: "s", Store: store.Nop{}} + if repairLostFreeze(c, "mask", "anything") { + t.Fatal("a store that cannot report losses must not authorize depth mutation") + } +} + +// The replay counters have to move, or the fix is unverifiable in a benchmark run. +func TestFrozenCountersMove(t *testing.T) { + h0, m0 := FrozenStats() + st := store.NewMemory(store.Options{}) + c := &components.Ctx{Session: "sCount", Store: st} + msg := tool("some tool output") + reapplyFrozen(c, "mask", &msg) // miss: nothing frozen yet + freeze(c, "mask", "some tool output", "short") + msg2 := tool("some tool output") + reapplyFrozen(c, "mask", &msg2) // hit + h1, m1 := FrozenStats() + if h1 <= h0 || m1 <= m0 { + t.Fatalf("hits/misses must both advance: %d->%d, %d->%d", h0, h1, m0, m1) + } +} diff --git a/components/offload/mask.go b/components/offload/mask.go index 9867304..9830b24 100644 --- a/components/offload/mask.go +++ b/components/offload/mask.go @@ -83,8 +83,12 @@ func (m *Mask) Offload(req *bschemas.BifrostChatRequest, rep *components.Report, } // A NEW mask only in the uncached tail: masking content the provider already // cached flips it full→masked and forces a cache-write of the suffix. Frozen masks - // are replayed everywhere above; new ones stay in the tail. - if !c.TailOnly(i) { + // are replayed everywhere above; new ones stay in the tail. The one exception is a + // freeze this session established and the store then LOST — there the provider + // already holds the masked bytes, so re-deriving them (deterministic: same content + // + config ⇒ same text and same sha256 key) PRESERVES the cache and leaving the + // output verbatim is what destroys it. + if !c.TailOnly(i) && !repairLostFreeze(c, m.Name(), content) { continue } prefix := "[older tool output masked] " diff --git a/components/offload/state.go b/components/offload/state.go index 957e004..d57169a 100644 --- a/components/offload/state.go +++ b/components/offload/state.go @@ -4,6 +4,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "sync/atomic" bschemas "github.com/maximhq/bifrost/core/schemas" "github.com/rossoctl/context-guru/components" @@ -44,7 +45,9 @@ func putResult(c *components.Ctx, id string, v []byte) { // REAPPLIES it on every turn, regardless of the tail boundary. New decisions are still // gated to the tail; frozen ones are replayed everywhere. -func frozenKey(session, comp, ck string) string { return "cg:frz:" + session + ":" + comp + ":" + ck } +func frozenKey(session, comp, ck string) string { + return store.FrozenPrefix + session + ":" + comp + ":" + ck +} // freeze records the replacement text a component produced for an original content, so // later turns replay it byte-for-byte. @@ -52,6 +55,28 @@ func freeze(c *components.Ctx, comp, original, replacement string) { c.Store.Put(frozenKey(c.Session, comp, contentKey(original)), []byte(replacement)) } +// frozenLost reports that this component DID freeze a decision for this content and the +// store has since dropped it (TTL expiry / pin cap). It is the counterpart to a plain +// reapplyFrozen miss, which is indistinguishable from "never frozen" — and the two call +// for OPPOSITE behavior: +// +// - never frozen: obey the tail gate. Compacting content the provider already cached +// flips it and forces a suffix cache-write, so NEW decisions stay in the tail. +// - frozen, then lost: the provider ALREADY holds the compacted bytes for this message, +// so leaving it verbatim is itself the cache-destructive move. Re-derive the decision +// even at depth. This is not a fail-open exception — an offloader's replacement text +// is a pure function of (content, component config), and the marker key is +// sha256(original), so re-deriving reproduces the SAME bytes the provider cached and +// re-establishes the freeze. For an ESTABLISHED compaction the safe direction is the +// opposite of the usual "forward the original" (see docs/design.md). +// +// Only the FACT of the freeze has to survive, not its payload — one key in a bounded set, +// which is why the signal lives in the store instead of a second content index. +func frozenLost(c *components.Ctx, comp, content string) bool { + fl, ok := c.Store.(store.FrozenLoser) + return ok && fl.FrozenLost(frozenKey(c.Session, comp, contentKey(content))) +} + // reapplyFrozen replays a component's frozen decision for the message at m, if one // exists and still shrinks it. It also refreshes the expand originals for any markers // in the replacement (the agent re-sent the full original as m's content), so @@ -63,8 +88,10 @@ func reapplyFrozen(c *components.Ctx, comp string, m *bschemas.ChatMessage) ([]s } repl, ok := c.Store.Get(frozenKey(c.Session, comp, contentKey(content))) if !ok { + frozenMisses.Add(1) return nil, 0, false } + frozenHits.Add(1) rs := string(repl) saved := schema.TextTokens(content) - schema.TextTokens(rs) if saved <= 0 { @@ -78,6 +105,29 @@ func reapplyFrozen(c *components.Ctx, comp string, m *bschemas.ChatMessage) ([]s return keys, saved, true } +// repairLostFreeze reports whether an offloader may compact this message even though the +// cache-tail gate would forbid it, because a freeze for it was established and then lost. +// Re-deriving reproduces the bytes the provider already cached; NOT re-deriving is what +// flips the representation and re-writes the suffix. The caller's own never-worse and +// skipReduce guards still apply, so this only ever LIFTS the depth restriction. +func repairLostFreeze(c *components.Ctx, comp, content string) bool { + return frozenLost(c, comp, content) +} + +// Freeze-replay counters: how often a replay landed vs found nothing. Cache-write is the +// largest cost line on long-horizon traffic and a lost freeze is the mechanism that +// produces it, so the store counts the drops and the repairs (a re-Put of a dropped +// frozen key) itself — exactly once each, regardless of how many turns observe them — +// while these two count the replay path. /stats reports all of them together. +var ( + frozenHits atomic.Int64 + frozenMisses atomic.Int64 +) + +// FrozenStats returns the cumulative freeze-replay hits and misses since process start. +// Exported for the host's /stats, which pairs them with the store's drop/repair counts. +func FrozenStats() (hits, misses int64) { return frozenHits.Load(), frozenMisses.Load() } + // contentKey is a marker/whitespace-insensitive content hash (shared with extract's // result cache), so the same output re-sent across turns maps to one frozen decision. func contentKey(s string) string { return extract.ContentKey(s) } diff --git a/docs/design.md b/docs/design.md index 3d5c3c0..bf104c8 100644 --- a/docs/design.md +++ b/docs/design.md @@ -190,7 +190,8 @@ sequenceDiagram An expired/evicted original resolves to an explicit placeholder rather than being omitted (the provider requires one `tool_result` per `tool_call_id`). A miss silently turns a lossless offload -lossy — the known TTL edge. +lossy — the known TTL edge, much narrower now the TTL slides on every read (see +[Freeze lifetime](#freeze-lifetime-and-which-way-to-fail)). ### The loop on a streaming response @@ -218,7 +219,7 @@ bytes* (`expand.rawMarkerRe`, used by the host's streaming decision) must accept ## State: the Store -One `Store` interface, in-memory TTL+LRU default (both hosts share it). Defaults: **1800s TTL, +One `Store` interface, in-memory TTL+LRU default (both hosts share it). Defaults: **10000s TTL, 1000 entries, 100 sticky sessions**. It carries, keyed per session: - **Rewind** — `cache_key → original bytes` (what the expand loop resolves). @@ -227,6 +228,43 @@ One `Store` interface, in-memory TTL+LRU default (both hosts share it). Defaults SQLite/Redis slot in behind the same interface when a durable/multi-replica deployment is real. +### Freeze lifetime, and which way to fail + +The TTL exists to reclaim state for **finished** sessions. Applying it to a *live* one is a bug +with a price tag: a frozen compaction (`cg:frz:…`, the exact replacement bytes an offloader must +replay so an already-cached message stays byte-identical) that dies mid-task makes that message +flip representation inside the provider's cached prefix, and the whole suffix is re-written at +**11.5x** the cache-read price. So the store treats a *read* as proof of life: + +- **Sliding TTL** — `Get` refreshes `expires`, not just LRU recency. An entry being replayed every + turn never ages out; one nobody reads still expires on its original deadline. +- **Default 10000s** — Terminal-Bench tasks average ~1975s of wall clock and run to 4h, so the + old 1800s default expired live decisions mid-task. Still `store.ttl_seconds`. +- **Frozen decisions are pinned** against LRU eviction (they are a marker line each), capped at + half the entry cap so one pathological session cannot pin the whole cache and starve the rewind + stashes the expand loop needs. + +**The fail direction inverts for an established compaction.** Fail-open normally means "forward the +original", and for a *new* compaction that is right. But once the provider has cached the compacted +bytes, forwarding the original **is** the destructive act. A plain `Get` miss can't tell those cases +apart, so the store keeps the *fact* of a dropped freeze (`FrozenLoser.FrozenLost`, a bounded key +set — the payload need not survive, only the knowledge that it existed): + +- **never frozen** → obey the tail gate; a new compaction stays in the uncached tail. +- **frozen, then lost** → re-derive it even at depth. An offloader's replacement text is a pure + function of `(content, component config)` and the marker key is `sha256(original)`, so + re-deriving reproduces the *same* bytes the provider cached and re-establishes the freeze. The + component's own never-worse and kept-verbatim guards still apply, so this only ever lifts the + depth restriction — it never authorizes new content loss. + +`/stats` reports `frozen_hits`, `frozen_misses`, `frozen_dropped`, `frozen_repaired`, and +`frozen_flips` (= dropped − repaired, the drops that actually cost a cache-write; it should be 0). + +The related fail-*open* on `MaxCachedIdx`: `prevLen` returning 0 on a store miss yields +`MaxCachedIdx = -1`, and `Ctx.TailOnly` then permits mutating any index (measured on 11.2% of +Terminal-Bench requests). The sliding TTL shrinks that window — `cg:len:` is read every turn, so it +no longer expires mid-session — but inverting `TailOnly` to fail *closed* is a separate change. + ## Session keying `session.Resolve(explicit, system, firstUser)`: an explicit host id wins; otherwise a stable @@ -265,7 +303,7 @@ pipeline: [format, dedup, failed_run, cmdfilter, cacheinject] # order + enable components: collapse: { max_tokens: 2000, head_lines: 20, tail_lines: 20 } smartcrush: { min_items: 5, keep_first: 3, keep_last: 2 } -store: { ttl_seconds: 1800, max_entries: 1000 } +store: { ttl_seconds: 10000, max_entries: 1000 } ``` A component registers its constructor + config type via `init()`; adding one makes it diff --git a/docs/how-to/recover-context.md b/docs/how-to/recover-context.md index cbbd133..6819820 100644 --- a/docs/how-to/recover-context.md +++ b/docs/how-to/recover-context.md @@ -100,7 +100,9 @@ from the model loop. ## Reversibility requires the store The store is the whole reversibility mechanism. It defaults to an in-memory TTL+LRU backend — -**1800s TTL, 1000 entries, 100 sticky sessions** — shared by every host. It holds, per session: +**10000s sliding TTL, 1000 entries, 100 sticky sessions** — shared by every host. The TTL is +refreshed on every read, so a stash an active session keeps touching does not expire under it. +It holds, per session: - **Rewind** — `cache_key → original bytes`, what the expand loop resolves. - **Sticky** — the set of content ids already reduced on prior turns (byte-stable output across diff --git a/docs/reference/config.md b/docs/reference/config.md index e377bfe..2372387 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -14,7 +14,7 @@ The document has four top-level fields (from the `Config` struct in | `preset` | string | Named default pipeline (see [Presets](presets.md)). | | `pipeline` | `[]string` | Ordered component names — controls **order + enablement**. Overrides the preset's pipeline when present. | | `components:` | map | Each component's typed config block, handed to its constructor verbatim. | -| `store` | object | State store options (`enabled`, `ttl_seconds`, `max_entries`, …). | +| `store` | object | State store options (`enabled`, `ttl_seconds` (default **10000**, sliding), `max_entries`, …). | !!! warning "Strict: unknown keys are rejected" The YAML loader runs with `KnownFields(true)`, so a typo'd key fails loudly @@ -28,7 +28,7 @@ pipeline: [format, dedup, failed_run, cmdfilter, cacheinject] # order + enable components: collapse: { max_tokens: 2000, head_lines: 20, tail_lines: 20 } smartcrush: { min_items: 5, keep_first: 3, keep_last: 2 } -store: { ttl_seconds: 1800, max_entries: 1000 } +store: { ttl_seconds: 10000, max_entries: 1000 } ``` A component registers its constructor + config type via `init()`, so adding one diff --git a/metrics/metrics.go b/metrics/metrics.go index 443d76f..efdc8b1 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -277,6 +277,20 @@ type Snapshot struct { // requests", not as a latency to compare against sse_ttfb_ms_avg. SSETTFBMsAvgBuf float64 `json:"sse_ttfb_ms_avg_buffered"` SSEBufferedPct float64 `json:"sse_buffered_pct"` + // Freeze-replay health — the cache-WRITE cost line. A frozen decision replayed + // (frozen_hits) keeps an already-cached message byte-identical. A decision the store + // DROPS (frozen_dropped: TTL expiry / pin cap) would flip that message's + // representation and force the provider to re-write the whole suffix at 11.5x the + // read price — unless it is re-derived (frozen_repaired). frozen_flips = + // dropped − repaired is the count that actually cost cache-writes; it should be 0. + // frozen_misses counts every replay lookup that found nothing, which is dominated by + // the normal "never frozen yet" case — read it beside frozen_dropped, not instead. + // Filled by the host at serve time (offload + store live below metrics). + FrozenHits int64 `json:"frozen_hits"` + FrozenMisses int64 `json:"frozen_misses"` + FrozenDropped int64 `json:"frozen_dropped"` + FrozenRepaired int64 `json:"frozen_repaired"` + FrozenFlips int64 `json:"frozen_flips"` } // Snapshot returns a point-in-time copy of the rollups. diff --git a/proxy/proxy.go b/proxy/proxy.go index f5ad0a7..77d5a79 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -590,6 +590,13 @@ func (h *Handler) stats(w http.ResponseWriter, _ *http.Request) { // Fill the CG components' own LLM cost (cheap-model usage) — kept out of the // metrics package (layering) and merged here at serve time. snap.LLMCalls, snap.LLMInputTokens, snap.LLMOutputTokens = cheapmodel.Usage() + // Freeze-replay health, same layering: the counters live with the code that owns + // them (offload for the replay path, the store for dropped/repaired decisions). + snap.FrozenHits, snap.FrozenMisses = offload.FrozenStats() + if fl, ok := h.store.(*store.Memory); ok { + snap.FrozenDropped, snap.FrozenRepaired = fl.FrozenLossStats() + snap.FrozenFlips = snap.FrozenDropped - snap.FrozenRepaired + } json.NewEncoder(w).Encode(snap) } diff --git a/store/store.go b/store/store.go index c6f8fe9..b663db1 100644 --- a/store/store.go +++ b/store/store.go @@ -13,6 +13,7 @@ package store import ( "container/list" + "strings" "sync" "time" ) @@ -34,15 +35,39 @@ type Store interface { Persists() bool } +// FrozenLoser is an OPTIONAL Store capability: reporting that a frozen decision +// under key was dropped (TTL expiry / pin cap) rather than never taken. A bare Get +// miss cannot tell those apart, and they call for opposite behavior — "never frozen" +// means obey the tail gate, "was frozen, now lost" means re-derive the same bytes so +// the cached prefix does not flip. Stores that don't implement it degrade to the +// legacy indistinguishable behavior. +type FrozenLoser interface { + // FrozenLost reports whether a frozen entry under key existed and was dropped. + FrozenLost(key string) bool +} + +// FrozenPrefix namespaces a component's FROZEN decision — the exact replacement +// bytes it must replay on every later turn to keep an already-cached message +// byte-identical (see components/offload/state.go). Entries under this prefix are +// PINNED: exempt from LRU eviction, because losing one is not a cache miss, it is a +// cache-DESTRUCTIVE event (the message flips representation inside the provider's +// cached prefix and the whole suffix is re-written at 11.5x the read price). They are +// small (a marker line), still honor the sliding TTL, and the exemption is capped at +// half the entry cap so a pathological session can never pin the whole cache. +const FrozenPrefix = "cg:frz:" + type entry struct { key string payload []byte expires time.Time + pinned bool // exempt from LRU eviction (frozen decision); TTL still applies } // Memory is an in-memory Store: a TTL+LRU cache for rewind payloads plus a -// bounded per-session sticky-id set. Defaults mirror headroom's CCR store -// (1800s TTL, 1000 entries). +// bounded per-session sticky-id set. The TTL is SLIDING (refreshed on Get), and +// the default (DefaultTTL) is sized past a long-horizon agent task rather than +// mirroring headroom's 1800s CCR store: a frozen compaction that dies mid-task is +// a cache-destructive event, not a saving. type Memory struct { mu sync.Mutex ttl time.Duration @@ -52,6 +77,13 @@ type Memory struct { sticky map[string]map[string]struct{} maxStick int now func() time.Time // injectable for tests + pinnedN int // live pinned (frozen) entries, capped at max/2 + // lostFrozen remembers keys whose FROZEN entry was dropped anyway (TTL expiry, or + // the pin cap). It is the "was frozen, now LOST" signal a caller cannot otherwise + // distinguish from "never frozen" — see FrozenLost. Bounded like sticky. + lostFrozen map[string]struct{} + lostN int64 + repairedN int64 } // Options configures a Memory store; the zero value yields sane defaults. @@ -77,12 +109,18 @@ func (Nop) Sticky(string) map[string]struct{} { return nil } func (Nop) MarkSticky(string, string) {} func (Nop) Persists() bool { return false } +// DefaultTTL is the store's default (sliding) entry lifetime. Terminal-Bench tasks +// averaged 1975s of wall clock and run up to 4h, so the old 1800s default expired +// live frozen decisions mid-task; ~2.8h covers a long-horizon task's idle gaps +// (test suites, training runs) with the sliding refresh doing the rest. +const DefaultTTL = 10000 * time.Second + // NewMemory builds an in-memory store. Zero/negative option fields fall back to -// defaults (1800s TTL, 1000 entries, 100 sessions of sticky sets). +// defaults (DefaultTTL, 1000 entries, 100 sessions of sticky sets). func NewMemory(o Options) *Memory { ttl := time.Duration(o.TTLSeconds) * time.Second if o.TTLSeconds <= 0 { - ttl = 1800 * time.Second + ttl = DefaultTTL } max := o.MaxEntries if max <= 0 { @@ -95,13 +133,23 @@ func NewMemory(o Options) *Memory { return &Memory{ ttl: ttl, max: max, maxStick: stick, ll: list.New(), items: map[string]*list.Element{}, - sticky: map[string]map[string]struct{}{}, - now: time.Now, + sticky: map[string]map[string]struct{}{}, + lostFrozen: map[string]struct{}{}, + now: time.Now, } } func (*Memory) Persists() bool { return true } +// SetClock replaces the store's time source. For TESTS only — TTL behavior over a +// multi-hour agent session is not testable in real time, and the freeze lifetime is +// exactly what this store gets wrong when it's wrong. +func (m *Memory) SetClock(now func() time.Time) { + m.mu.Lock() + defer m.mu.Unlock() + m.now = now +} + func (m *Memory) Put(key string, payload []byte) { m.mu.Lock() defer m.mu.Unlock() @@ -113,10 +161,59 @@ func (m *Memory) Put(key string, payload []byte) { return } e := &entry{key: key, payload: payload, expires: m.now().Add(m.ttl)} + // Pin frozen decisions, but never more than half the cache: past that the marginal + // pin protects one message while starving the rewind stashes the expand loop needs. + if strings.HasPrefix(key, FrozenPrefix) { + if m.pinnedN < m.max/2 { + e.pinned = true + m.pinnedN++ + } else { + m.noteLost(key) // pin cap reached: this decision is evictable, and losing it is visible + } + } + if _, wasLost := m.lostFrozen[key]; wasLost { + delete(m.lostFrozen, key) // re-frozen: the dropped decision was repaired + m.repairedN++ + } m.items[key] = m.ll.PushFront(e) for m.ll.Len() > m.max { - m.evictOldest() + if !m.evictOldest() { + break // everything left is pinned + } + } +} + +// noteLost records that a frozen decision under key is gone, so a later Get miss is +// distinguishable from "never frozen". Bounded by the entry cap. +func (m *Memory) noteLost(key string) { + if len(m.lostFrozen) >= m.max { + for k := range m.lostFrozen { + delete(m.lostFrozen, k) + break + } } + m.lostFrozen[key] = struct{}{} + m.lostN++ +} + +// FrozenLost reports whether a frozen entry under key existed and was dropped (TTL +// expiry or the pin cap) — the "was frozen, now lost" signal. See FrozenLoser. +func (m *Memory) FrozenLost(key string) bool { + m.mu.Lock() + defer m.mu.Unlock() + _, ok := m.lostFrozen[key] + return ok +} + +// FrozenLossStats returns how many frozen decisions this store has DROPPED since start +// (TTL expiry / pin cap) and how many of those were later re-Put — repaired to the same +// bytes, so no representation flip reached the provider. dropped−repaired is the count of +// flips that actually cost a suffix cache-write. Both count each key once, however many +// turns observe it. +func (m *Memory) FrozenLossStats() (dropped, repaired int64) { + m.mu.Lock() + defer m.mu.Unlock() + return m.lostN, m.repairedN } func (m *Memory) Get(key string) ([]byte, bool) { @@ -131,6 +228,12 @@ func (m *Memory) Get(key string) ([]byte, bool) { m.remove(el) return nil, false } + // Sliding TTL: an entry still being read is still live. The TTL exists to reclaim + // state for FINISHED sessions, not to kill a decision an ongoing session replays + // every turn — expiring a frozen compaction mid-task flips an already-cached + // message's representation and forces the provider to re-write the whole suffix + // (one cache-write costs 11.5 cache-reads). Recency and lifetime refresh together. + e.expires = m.now().Add(m.ttl) m.ll.MoveToFront(el) return e.payload, true } @@ -165,13 +268,24 @@ func (m *Memory) MarkSticky(session, id string) { s[id] = struct{}{} } -func (m *Memory) evictOldest() { - if el := m.ll.Back(); el != nil { - m.remove(el) +// evictOldest drops the least-recently-used UNPINNED entry, walking back over pinned +// (frozen) ones. Reports false when nothing is evictable. +func (m *Memory) evictOldest() bool { + for el := m.ll.Back(); el != nil; el = el.Prev() { + if !el.Value.(*entry).pinned { + m.remove(el) + return true + } } + return false } func (m *Memory) remove(el *list.Element) { + e := el.Value.(*entry) + if e.pinned { + m.pinnedN-- + m.noteLost(e.key) // a frozen decision is disappearing — make it detectable + } m.ll.Remove(el) - delete(m.items, el.Value.(*entry).key) + delete(m.items, e.key) } diff --git a/store/store_test.go b/store/store_test.go index b655cf4..46f9feb 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -71,3 +71,128 @@ func TestStickyBoundedAndCopied(t *testing.T) { t.Fatal("Sticky must return a defensive copy") } } + +// A frozen decision that is still being replayed every turn must not die of old age: +// the TTL reclaims state for FINISHED sessions, and expiring a live decision flips an +// already-cached message's representation (a suffix cache-write at 11.5x the read price). +func TestMemorySlidingTTLOnGet(t *testing.T) { + now := time.Unix(0, 0) + m := NewMemory(Options{TTLSeconds: 10}) + m.now = func() time.Time { return now } + m.Put("k", []byte("v")) + // Read just before each expiry, 100 times = 900s ≫ the 10s TTL. + for i := 0; i < 100; i++ { + now = now.Add(9 * time.Second) + if _, ok := m.Get("k"); !ok { + t.Fatalf("a continuously-read entry expired at iteration %d", i) + } + } + // ... but stopping the reads still expires it (the TTL is not disabled). + now = now.Add(11 * time.Second) + if _, ok := m.Get("k"); ok { + t.Fatal("an unread entry must still expire") + } +} + +// The sliding refresh must not keep an entry nobody reads alive just because OTHER +// entries are being read. +func TestMemoryUnreadEntryStillExpires(t *testing.T) { + now := time.Unix(0, 0) + m := NewMemory(Options{TTLSeconds: 10}) + m.now = func() time.Time { return now } + m.Put("hot", []byte("1")) + m.Put("cold", []byte("2")) + for i := 0; i < 5; i++ { + now = now.Add(9 * time.Second) + m.Get("hot") + } + if _, ok := m.Get("cold"); ok { + t.Fatal("an entry that was never read must expire on its original deadline") + } + if _, ok := m.Get("hot"); !ok { + t.Fatal("the continuously-read entry must survive") + } +} + +// The default TTL must outlast a long-horizon agent task (TB averaged 1975s, up to 4h) +// — the old 1800s default expired live frozen decisions mid-task. +func TestDefaultTTLCoversLongTask(t *testing.T) { + if DefaultTTL < 2*time.Hour { + t.Fatalf("default TTL %v is too short for a long-horizon task", DefaultTTL) + } + m := NewMemory(Options{}) + if m.ttl != DefaultTTL { + t.Fatalf("zero TTLSeconds should yield DefaultTTL, got %v", m.ttl) + } + if m2 := NewMemory(Options{TTLSeconds: 42}); m2.ttl != 42*time.Second { + t.Fatalf("ttl_seconds must stay configurable, got %v", m2.ttl) + } +} + +// Frozen decisions are pinned against LRU eviction: losing one is not a cache miss but +// a cache-DESTRUCTIVE event. Ordinary entries still evict normally. +func TestFrozenEntriesExemptFromLRU(t *testing.T) { + m := NewMemory(Options{MaxEntries: 4}) // pin cap = max/2 = 2 + m.Put(FrozenPrefix+"s:mask:aaa", []byte("frozen")) + for i := 0; i < 20; i++ { + m.Put(string(rune('a'+i)), []byte("x")) + } + if _, ok := m.Get(FrozenPrefix + "s:mask:aaa"); !ok { + t.Fatal("a frozen decision must survive LRU pressure") + } + if m.ll.Len() > 4 { + t.Fatalf("cache still has to respect the entry cap, len=%d", m.ll.Len()) + } +} + +// The eviction exemption must be capped so a pathological session cannot pin the whole +// cache and starve the rewind stashes the expand loop needs. +func TestFrozenPinCapped(t *testing.T) { + m := NewMemory(Options{MaxEntries: 10}) // pin cap = 5 + for i := 0; i < 20; i++ { + m.Put(FrozenPrefix+"s:mask:"+string(rune('a'+i)), []byte("f")) + } + if m.pinnedN > 5 { + t.Fatalf("pinned entries %d exceed the max/2 cap", m.pinnedN) + } + if m.ll.Len() > 10 { + t.Fatalf("entry cap breached, len=%d", m.ll.Len()) + } + // Beyond the cap the decisions are evictable — but their loss stays VISIBLE, which + // is the whole point of the signal. + if dropped, _ := m.FrozenLossStats(); dropped == 0 { + t.Fatal("frozen decisions dropped past the pin cap must be counted") + } +} + +// A dropped frozen decision must be distinguishable from one that never existed: the two +// call for opposite behavior (re-derive at depth vs obey the tail gate). +func TestFrozenLostIsDistinguishable(t *testing.T) { + now := time.Unix(0, 0) + m := NewMemory(Options{TTLSeconds: 10}) + m.now = func() time.Time { return now } + k := FrozenPrefix + "s:mask:aaa" + if m.FrozenLost(k) { + t.Fatal("a key that was never frozen must not report as lost") + } + m.Put(k, []byte("masked")) + now = now.Add(11 * time.Second) + if _, ok := m.Get(k); ok { + t.Fatal("expired entry must miss") + } + if !m.FrozenLost(k) { + t.Fatal("an EXPIRED frozen decision must report as lost, not as never-frozen") + } + dropped, repaired := m.FrozenLossStats() + if dropped != 1 || repaired != 0 { + t.Fatalf("want dropped=1 repaired=0, got %d/%d", dropped, repaired) + } + // Re-freezing the same key repairs it: no flip reached the provider. + m.Put(k, []byte("masked")) + if m.FrozenLost(k) { + t.Fatal("a re-frozen decision is no longer lost") + } + if dropped, repaired = m.FrozenLossStats(); dropped != 1 || repaired != 1 { + t.Fatalf("want dropped=1 repaired=1, got %d/%d", dropped, repaired) + } +} From c4b91cf8ee0e347efdccbb141f3224f936c010d5 Mon Sep 17 00:00:00 2001 From: Osher-Elhadad Date: Mon, 10 Aug 2026 03:33:23 +0000 Subject: [PATCH 2/7] fix(cache): extend freeze-lifetime protection to extract_llm's result cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The freeze-replay contract is implemented twice under different names. mask and failed_run use freeze/reapplyFrozen (cg:frz:); extract_llm uses its own per-content result cache (cg:res:, plus cg:sum1: for the summary line it re-emits) — and that is the one that carries the load in the shipped coding config, where mask is absent by design and failed_run self-skips on a cached agent. Scoping the lifetime fix to cg:frz: alone would have been measurably inert on exactly the traffic that motivated it. So the pin, the loss signal and the depth-repair now cover both namespaces: a lost result-cache entry un-compacts an already-cached message exactly the way a lost frozen mask does, and gets the same treatment. The rewind stashes (bare content hashes — the large originals the expand loop resolves) stay fully evictable; only the small replay decisions are pinned. For extract_llm the repair costs a model call and the LLM may not reproduce the bytes exactly, which is noted at the call site; the alternative is a guaranteed full-suffix cache-write, dearer by a wide margin. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: Osher-Elhadad --- components/offload/extract_llm.go | 9 ++++++- components/offload/freeze_test.go | 34 +++++++++++++++++++++++++++ components/offload/state.go | 19 +++++++++++---- store/store.go | 39 +++++++++++++++++++++++-------- 4 files changed, 85 insertions(+), 16 deletions(-) diff --git a/components/offload/extract_llm.go b/components/offload/extract_llm.go index b1618eb..4f540a2 100644 --- a/components/offload/extract_llm.go +++ b/components/offload/extract_llm.go @@ -238,7 +238,14 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R // caching is off, any message is fair game. File reads included (largest mass); // safe because we never touch already-cached content and freeze+reapply the result. sz := schema.TextTokens(content) - if c.CacheAware && !c.TailOnly(i) { + // Exception to the tail gate: a result-cache entry this session established and the + // store then LOST. The provider already holds the compacted bytes for this message, + // so re-deriving them restores the cached representation, whereas leaving the output + // verbatim is what flips it and re-writes the suffix. (Unlike the deterministic + // offloaders this costs a model call and the LLM may not reproduce the bytes + // exactly — but the alternative is a GUARANTEED full-suffix cache-write, the more + // expensive of the two by a wide margin.) + if c.CacheAware && !c.TailOnly(i) && !repairLostResult(c, id) { dbgTail++ if sz >= floor { dbgBigTailBlocked++ // a large output we skipped ONLY because it's not in the tail diff --git a/components/offload/freeze_test.go b/components/offload/freeze_test.go index ef95e04..e3dc55f 100644 --- a/components/offload/freeze_test.go +++ b/components/offload/freeze_test.go @@ -151,3 +151,37 @@ func TestFrozenCountersMove(t *testing.T) { t.Fatalf("hits/misses must both advance: %d->%d, %d->%d", h0, h1, m0, m1) } } + +// The result cache (cg:res:) is the OTHER replay namespace — extract_llm's — and it is +// the one that carries the load in the shipped coding config (no mask; failed_run +// self-skips on a cached agent). It must get the same protection: pinned against +// eviction, and its loss reported so the compaction can be re-derived at depth. +func TestResultCachePinnedAndRepairable(t *testing.T) { + now := time.Unix(0, 0) + st := store.NewMemory(store.Options{TTLSeconds: 10, MaxEntries: 4}) + st.SetClock(func() time.Time { return now }) + c := &components.Ctx{Session: "s", Store: st} + + putResult(c, "id1", []byte("compacted")) + // Ordinary rewind stashes churn through the cache; the replay decision must survive. + for i := 0; i < 20; i++ { + st.Put("rewindhash"+string(rune('a'+i)), []byte("big original payload")) + } + if _, ok := getResult(c, "id1"); !ok { + t.Fatal("a result-cache decision must be pinned against LRU eviction") + } + if repairLostResult(c, "id1") { + t.Fatal("a live decision is not lost") + } + // Expire it: nothing reads it for longer than the TTL. + now = now.Add(11 * time.Second) + if _, ok := getResult(c, "id1"); ok { + t.Fatal("expired entry must miss") + } + if !repairLostResult(c, "id1") { + t.Fatal("a LOST result-cache decision must be distinguishable from never-cached") + } + if repairLostResult(c, "never-seen") { + t.Fatal("content that was never compacted must not authorize depth mutation") + } +} diff --git a/components/offload/state.go b/components/offload/state.go index d57169a..7738864 100644 --- a/components/offload/state.go +++ b/components/offload/state.go @@ -21,7 +21,7 @@ import ( // both costly and cache-hostile. // resultKey namespaces a per-content reduced output (extract) by session. -func resultKey(session, id string) string { return "cg:res:" + session + ":" + id } +func resultKey(session, id string) string { return store.ResultPrefix + session + ":" + id } // getResult returns a previously cached reduced output for content id, if any. func getResult(c *components.Ctx, id string) ([]byte, bool) { @@ -72,9 +72,9 @@ func freeze(c *components.Ctx, comp, original, replacement string) { // // Only the FACT of the freeze has to survive, not its payload — one key in a bounded set, // which is why the signal lives in the store instead of a second content index. -func frozenLost(c *components.Ctx, comp, content string) bool { +func frozenLost(c *components.Ctx, key string) bool { fl, ok := c.Store.(store.FrozenLoser) - return ok && fl.FrozenLost(frozenKey(c.Session, comp, contentKey(content))) + return ok && fl.FrozenLost(key) } // reapplyFrozen replays a component's frozen decision for the message at m, if one @@ -111,7 +111,16 @@ func reapplyFrozen(c *components.Ctx, comp string, m *bschemas.ChatMessage) ([]s // flips the representation and re-writes the suffix. The caller's own never-worse and // skipReduce guards still apply, so this only ever LIFTS the depth restriction. func repairLostFreeze(c *components.Ctx, comp, content string) bool { - return frozenLost(c, comp, content) + return frozenLost(c, frozenKey(c.Session, comp, contentKey(content))) +} + +// repairLostResult is repairLostFreeze for the OTHER replay namespace: extract_llm's +// per-content result cache (cg:res:), which is the same replay contract under a different +// name — and the one that actually carries the load in the shipped coding config, where +// mask is absent and failed_run self-skips on a cached agent. A lost result cache entry +// un-compacts an already-cached message exactly the same way, so it gets the same repair. +func repairLostResult(c *components.Ctx, id string) bool { + return frozenLost(c, resultKey(c.Session, id)) } // Freeze-replay counters: how often a replay landed vs found nothing. Cache-write is the @@ -195,7 +204,7 @@ func OwnsKey(st store.Store, session, key string) bool { // summaryKey namespaces the one-line SUMMARY the LLM extract emitted for a content // id, so a later turn reusing the cached reduction also re-emits the same marker // digest (byte-stable) without re-calling the model. -func summaryKey(session, id string) string { return "cg:sum1:" + session + ":" + id } +func summaryKey(session, id string) string { return store.SummaryPrefix + session + ":" + id } func getSummary(c *components.Ctx, id string) (string, bool) { b, ok := c.Store.Get(summaryKey(c.Session, id)) diff --git a/store/store.go b/store/store.go index b663db1..bc5a13a 100644 --- a/store/store.go +++ b/store/store.go @@ -46,15 +46,34 @@ type FrozenLoser interface { FrozenLost(key string) bool } -// FrozenPrefix namespaces a component's FROZEN decision — the exact replacement -// bytes it must replay on every later turn to keep an already-cached message -// byte-identical (see components/offload/state.go). Entries under this prefix are -// PINNED: exempt from LRU eviction, because losing one is not a cache miss, it is a -// cache-DESTRUCTIVE event (the message flips representation inside the provider's -// cached prefix and the whole suffix is re-written at 11.5x the read price). They are -// small (a marker line), still honor the sliding TTL, and the exemption is capped at -// half the entry cap so a pathological session can never pin the whole cache. -const FrozenPrefix = "cg:frz:" +// Key namespaces whose entries are a component's FROZEN decision — the replacement +// text it must replay on every later turn to keep an already-cached message +// byte-identical (see components/offload/state.go). Two components' worth, because the +// freeze-replay mechanism was implemented twice under different names: +// +// cg:frz: — mask / failed_run (freeze + reapplyFrozen) +// cg:res: — extract_llm's result cache, plus cg:sum1: for the summary line it +// re-emits alongside it. Functionally the same replay contract. +// +// Entries under these prefixes are PINNED: exempt from LRU eviction, because losing one +// is not a cache miss, it is a cache-DESTRUCTIVE event — the message flips representation +// inside the provider's cached prefix and the whole suffix is re-written at 11.5x the +// read price. They are small (a marker line / a compacted projection), still honor the +// sliding TTL, and the exemption is capped at half the entry cap so a pathological +// session can never pin the whole cache. The rewind stashes (bare content hashes, the +// large payloads the expand loop resolves) stay fully evictable. +const ( + FrozenPrefix = "cg:frz:" + ResultPrefix = "cg:res:" + SummaryPrefix = "cg:sum1:" +) + +// frozenNamespace reports whether key holds a replay decision (see FrozenPrefix). +func frozenNamespace(key string) bool { + return strings.HasPrefix(key, FrozenPrefix) || + strings.HasPrefix(key, ResultPrefix) || + strings.HasPrefix(key, SummaryPrefix) +} type entry struct { key string @@ -163,7 +182,7 @@ func (m *Memory) Put(key string, payload []byte) { e := &entry{key: key, payload: payload, expires: m.now().Add(m.ttl)} // Pin frozen decisions, but never more than half the cache: past that the marginal // pin protects one message while starving the rewind stashes the expand loop needs. - if strings.HasPrefix(key, FrozenPrefix) { + if frozenNamespace(key) { if m.pinnedN < m.max/2 { e.pinned = true m.pinnedN++ From 654877004712c36408f1e84b0c5ac2e58bc0559a Mon Sep 17 00:00:00 2001 From: Osher-Elhadad Date: Mon, 10 Aug 2026 03:53:26 +0000 Subject: [PATCH 3/7] fix(metrics): count extract_llm's result-cache replay in the frozen counters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A smoke run on the shipped coding config reported frozen_hits/misses = 0 while the pipeline was replaying compactions every turn: only reapplyFrozen fed the counters, and that config replays entirely through extract_llm's result cache. The counters would have been blind on exactly the traffic the freeze-lifetime fix targets, which defeats their purpose — verifying the fix on a benchmark run. getResult now feeds the same hit/miss counters as reapplyFrozen. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: Osher-Elhadad --- components/offload/freeze_test.go | 20 ++++++++++++++++++++ components/offload/state.go | 13 +++++++++++-- 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/components/offload/freeze_test.go b/components/offload/freeze_test.go index e3dc55f..08f8d3c 100644 --- a/components/offload/freeze_test.go +++ b/components/offload/freeze_test.go @@ -185,3 +185,23 @@ func TestResultCachePinnedAndRepairable(t *testing.T) { t.Fatal("content that was never compacted must not authorize depth mutation") } } + +// The counters must also move on extract_llm's replay path (getResult), not just +// reapplyFrozen — the shipped coding config does ALL of its replay through the result +// cache, so counting only reapplyFrozen would report zero freeze activity on exactly the +// traffic this fix targets. +func TestResultCacheFeedsCounters(t *testing.T) { + h0, m0 := FrozenStats() + c := &components.Ctx{Session: "sRC", Store: store.NewMemory(store.Options{})} + if _, ok := getResult(c, "idz"); ok { + t.Fatal("nothing cached yet") + } + putResult(c, "idz", []byte("compacted")) + if _, ok := getResult(c, "idz"); !ok { + t.Fatal("expected a replay hit") + } + h1, m1 := FrozenStats() + if h1 <= h0 || m1 <= m0 { + t.Fatalf("result-cache replay must feed hits/misses: %d->%d, %d->%d", h0, h1, m0, m1) + } +} diff --git a/components/offload/state.go b/components/offload/state.go index 7738864..0760f87 100644 --- a/components/offload/state.go +++ b/components/offload/state.go @@ -23,9 +23,18 @@ import ( // resultKey namespaces a per-content reduced output (extract) by session. func resultKey(session, id string) string { return store.ResultPrefix + session + ":" + id } -// getResult returns a previously cached reduced output for content id, if any. +// getResult returns a previously cached reduced output for content id, if any. This is +// extract_llm's replay lookup, so it feeds the same hit/miss counters as reapplyFrozen — +// otherwise the shipped coding config (no mask, failed_run self-skipping) would report +// zero freeze activity while doing all of its replay through here. func getResult(c *components.Ctx, id string) ([]byte, bool) { - return c.Store.Get(resultKey(c.Session, id)) + v, ok := c.Store.Get(resultKey(c.Session, id)) + if ok { + frozenHits.Add(1) + } else { + frozenMisses.Add(1) + } + return v, ok } // putResult caches a reduced output so a later turn re-sending the same content From 4baa6f9ac99447f405c23db50972313b53589773 Mon Sep 17 00:00:00 2001 From: Osher-Elhadad Date: Mon, 10 Aug 2026 04:29:15 +0000 Subject: [PATCH 4/7] fix(store): don't score an unprotected frozen decision as repaired MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found while reviewing the counters before trusting them for the benchmark write-up: at the pin cap, Put called noteLost(key) and then the repair check immediately deleted that same mark and incremented repairedN. A store at its pin cap therefore reported dropped=4 repaired=4 — frozen_flips = dropped - repaired = 0 — while four decisions were in fact unprotected. The metric failed in the flattering direction, which is the one that would have made a broken fix look like a working one. The repair check now runs once, before either branch, so it also catches the case where the entry still exists (over the cap the key stays in the map, unpinned) and cannot double-count a decision that is immediately unprotected again. repaired can no longer exceed dropped, and a test asserts that invariant. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: Osher-Elhadad --- store/store.go | 15 ++++++++++----- store/store_test.go | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 5 deletions(-) diff --git a/store/store.go b/store/store.go index bc5a13a..6da44a6 100644 --- a/store/store.go +++ b/store/store.go @@ -172,6 +172,15 @@ func (m *Memory) SetClock(now func() time.Time) { func (m *Memory) Put(key string, payload []byte) { m.mu.Lock() defer m.mu.Unlock() + // Writing a key previously recorded as lost IS the repair — the decision is present + // again. Counted here, before either branch, so it is not missed when the entry still + // exists (over the pin cap it stays in the map, unpinned) and so a decision that is + // immediately unprotected again is not ALSO scored as repaired: repaired must never + // exceed dropped, or frozen_flips reads 0 while messages are in fact flipping. + if _, wasLost := m.lostFrozen[key]; wasLost { + delete(m.lostFrozen, key) + m.repairedN++ + } if el, ok := m.items[key]; ok { e := el.Value.(*entry) e.payload = payload @@ -187,13 +196,9 @@ func (m *Memory) Put(key string, payload []byte) { e.pinned = true m.pinnedN++ } else { - m.noteLost(key) // pin cap reached: this decision is evictable, and losing it is visible + m.noteLost(key) // pin cap reached: evictable, and its loss stays visible } } - if _, wasLost := m.lostFrozen[key]; wasLost { - delete(m.lostFrozen, key) // re-frozen: the dropped decision was repaired - m.repairedN++ - } m.items[key] = m.ll.PushFront(e) for m.ll.Len() > m.max { if !m.evictOldest() { diff --git a/store/store_test.go b/store/store_test.go index 46f9feb..5f2357b 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -196,3 +196,35 @@ func TestFrozenLostIsDistinguishable(t *testing.T) { t.Fatalf("want dropped=1 repaired=1, got %d/%d", dropped, repaired) } } + +// repaired must never exceed dropped. At the pin cap a decision is recorded lost and +// stays unprotected, so it must NOT also be scored as repaired — otherwise frozen_flips +// reads 0 while messages are in fact flipping, and the metric lies in the safe direction. +func TestPinCapDoesNotFakeRepair(t *testing.T) { + m := NewMemory(Options{MaxEntries: 4}) // pin cap = 2 + for i := 0; i < 6; i++ { + m.Put(FrozenPrefix+"s:mask:"+string(rune('a'+i)), []byte("f")) + } + dropped, repaired := m.FrozenLossStats() + if dropped == 0 { + t.Fatal("over-cap frozen decisions must be counted as dropped") + } + if repaired != 0 { + t.Fatalf("nothing was re-frozen, so repaired must be 0, got %d", repaired) + } + // Re-freezing a key that WAS lost counts exactly one repair. (The first two keys are + // pinned and were never lost, so pick one the cap actually pushed out.) + var lost string + for k := range m.lostFrozen { + lost = k + break + } + m.Put(lost, []byte("f2")) + d2, r2 := m.FrozenLossStats() + if r2 != 1 { + t.Fatalf("re-freezing a lost decision must count one repair, got %d", r2) + } + if r2 > d2 { + t.Fatalf("repaired (%d) must never exceed dropped (%d)", r2, d2) + } +} From 6e009a8fea82496578021942d8aa36334f8fbdd5 Mon Sep 17 00:00:00 2001 From: Osher-Elhadad Date: Mon, 10 Aug 2026 05:06:15 +0000 Subject: [PATCH 5/7] test(offload): measure the cache-write cost of a lost freeze MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The issue's cost hypothesis (~15 reverts ≈ 2.5M cache-write tokens) had to be validated rather than accepted, and a live benchmark alone cannot isolate it — the effect only appears in sessions longer than the old TTL, and the shipped coding config leaves mask out entirely. So the mechanism is measured directly: replay a 120-turn session (26 s/request, the measured gateway latency) through mask and count the tokens a provider must re-write because a message inside the already-cached prefix changed representation. write-only TTL (old): 50 flips, 191,681 cache-write tokens, $0.44 premium sliding TTL (new): 0 flips, 0 cache-write tokens, $0.00 Same mechanism and direction as the hypothesis, for one 120-turn session; the issue's 2.5M figure covered the whole 89-task suite. The test asserts BOTH that the old behavior still reproduces flips (so the fixture cannot silently stop exercising the bug) and that the new one eliminates them. DisableSlidingTTLForTest is the test seam that makes the before/after comparison possible in a single process. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: Osher-Elhadad --- components/offload/flipcost_test.go | 88 +++++++++++++++++++++++++++++ store/store.go | 15 ++++- 2 files changed, 101 insertions(+), 2 deletions(-) create mode 100644 components/offload/flipcost_test.go diff --git a/components/offload/flipcost_test.go b/components/offload/flipcost_test.go new file mode 100644 index 0000000..e2a4505 --- /dev/null +++ b/components/offload/flipcost_test.go @@ -0,0 +1,88 @@ +package offload + +import ( + "fmt" + "strings" + "testing" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" +) + +// replaySession drives a long-horizon session through `mask` and counts the provider +// cache-write it would incur from REPRESENTATION FLIPS: a message inside the +// already-cached prefix whose forwarded bytes differ from the bytes sent last turn +// invalidates the cache from its index onward, so the whole suffix is re-written. +// +// ttl/slide select the store behavior under test. This is the issue's cost hypothesis +// (~15 reverts ≈ 2.5M cache-write tokens) reduced to something deterministic and +// measurable in CI: no gateway, no model, just the replay bookkeeping the fix changes. +func replaySession(t *testing.T, ttlSeconds int, slide bool, turns, secsPerTurn int) (flips, writeTokens int) { + t.Helper() + st := store.NewMemory(store.Options{TTLSeconds: ttlSeconds}) + now := time.Unix(0, 0) + st.SetClock(func() time.Time { return now }) + if !slide { + st.DisableSlidingTTLForTest() + } + // keep_recent: 0 so the NEWEST (uncached) tool output is itself a mask candidate — + // with a keep-recent window the tail is never masked, so no decision is ever frozen + // and there is nothing for the TTL to lose. + comp, err := newMask([]byte("keep_recent: 0\nmin_tokens: 100\n")) + if err != nil { + t.Fatal(err) + } + m := comp.(*Mask) + + var hist []bschemas.ChatMessage + prev := map[int]string{} + for turn := 0; turn < turns; turn++ { + hist = append(hist, tool(fmt.Sprintf("output %d\n", turn)+ + strings.Repeat("verbose tool output line\n", 60))) + // The agent re-sends the ORIGINAL history verbatim every turn, plus a new tail. + req := &bschemas.BifrostChatRequest{Input: append([]bschemas.ChatMessage(nil), hist...)} + c := &components.Ctx{Session: "s", Store: st, CacheAware: true, + MaxCachedIdx: len(hist) - 2} // all but the newest message is already cached + var rep components.Report + if _, err := m.Offload(req, &rep, c); err != nil { + t.Fatal(err) + } + for i := 0; i < len(req.Input)-1; i++ { + got := schema.MessageText(req.Input[i]) + if was, seen := prev[i]; seen && was != got { + flips++ + for j := i; j < len(req.Input); j++ { + writeTokens += schema.TextTokens(schema.MessageText(req.Input[j])) + } + } + prev[i] = got + } + now = now.Add(time.Duration(secsPerTurn) * time.Second) + } + return flips, writeTokens +} + +// The headline claim: over a session longer than the old TTL, the write-only TTL flips +// already-cached messages and the sliding TTL does not. Numbers are logged so the PR can +// quote them, and asserted so a regression fails the build. +func TestFlipCostOverLongSession(t *testing.T) { + const turns, secsPerTurn = 120, 26 // 120 turns x 26 s/req = 3120 s > the old 1800 s TTL + oldFlips, oldWrite := replaySession(t, 1800, false, turns, secsPerTurn) + newFlips, newWrite := replaySession(t, 1800, true, turns, secsPerTurn) + + t.Logf("write-only TTL (old): flips=%d cache-write=%d tokens premium@$2.30/M=$%.2f", + oldFlips, oldWrite, float64(oldWrite)*2.30/1e6) + t.Logf("sliding TTL (new): flips=%d cache-write=%d tokens premium@$2.30/M=$%.2f", + newFlips, newWrite, float64(newWrite)*2.30/1e6) + + if oldFlips == 0 { + t.Fatal("the old write-only TTL must reproduce the flips this issue is about") + } + if newFlips != 0 { + t.Fatalf("the sliding TTL must eliminate representation flips, got %d (%d tokens)", + newFlips, newWrite) + } +} diff --git a/store/store.go b/store/store.go index 6da44a6..e42f30d 100644 --- a/store/store.go +++ b/store/store.go @@ -103,6 +103,7 @@ type Memory struct { lostFrozen map[string]struct{} lostN int64 repairedN int64 + noSlide bool // tests only: restore the old write-only expiry (see DisableSlidingTTLForTest) } // Options configures a Memory store; the zero value yields sane defaults. @@ -169,6 +170,14 @@ func (m *Memory) SetClock(now func() time.Time) { m.now = now } +// DisableSlidingTTLForTest restores the old write-only expiry (and un-pins frozen +// entries) so a test can measure what the previous behavior cost. For TESTS only. +func (m *Memory) DisableSlidingTTLForTest() { + m.mu.Lock() + defer m.mu.Unlock() + m.noSlide = true +} + func (m *Memory) Put(key string, payload []byte) { m.mu.Lock() defer m.mu.Unlock() @@ -191,7 +200,7 @@ func (m *Memory) Put(key string, payload []byte) { e := &entry{key: key, payload: payload, expires: m.now().Add(m.ttl)} // Pin frozen decisions, but never more than half the cache: past that the marginal // pin protects one message while starving the rewind stashes the expand loop needs. - if frozenNamespace(key) { + if frozenNamespace(key) && !m.noSlide { if m.pinnedN < m.max/2 { e.pinned = true m.pinnedN++ @@ -257,7 +266,9 @@ func (m *Memory) Get(key string) ([]byte, bool) { // every turn — expiring a frozen compaction mid-task flips an already-cached // message's representation and forces the provider to re-write the whole suffix // (one cache-write costs 11.5 cache-reads). Recency and lifetime refresh together. - e.expires = m.now().Add(m.ttl) + if !m.noSlide { + e.expires = m.now().Add(m.ttl) + } m.ll.MoveToFront(el) return e.payload, true } From 617ddcc59cc825915c0da64fccb2a4e10ed13089 Mon Sep 17 00:00:00 2001 From: Osher-Elhadad Date: Mon, 10 Aug 2026 06:05:48 +0000 Subject: [PATCH 6/7] fix(store): report a frozen loss even when the entry was never pinned MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit remove() gated the loss signal on e.pinned, so a replay decision that missed the pin cap vanished silently: unreported by FrozenLost, and therefore never repaired. That is exactly backwards — an unpinned decision is the one MOST likely to be dropped, since it has no eviction protection left. Keyed on the namespace instead. Found by writing a test to pin down what the counters mean under the old semantics, which is also why DisableSlidingTTLForTest now suppresses the loss signal: the old store had none, so the before/after comparison must not hand the "before" arm a repair path it never had. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: Osher-Elhadad --- store/store.go | 8 +++++++- store/store_test.go | 18 ++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/store/store.go b/store/store.go index e42f30d..395ea9e 100644 --- a/store/store.go +++ b/store/store.go @@ -319,7 +319,13 @@ func (m *Memory) remove(el *list.Element) { e := el.Value.(*entry) if e.pinned { m.pinnedN-- - m.noteLost(e.key) // a frozen decision is disappearing — make it detectable + } + // Any replay decision disappearing must be detectable — keyed on the NAMESPACE, not on + // the pin flag. An entry that missed the pin cap is exactly the one most likely to be + // dropped, and gating this on e.pinned would let it vanish silently: unreported, and so + // never repaired. (noSlide reproduces the OLD store, which had no loss signal at all.) + if frozenNamespace(e.key) && !m.noSlide { + m.noteLost(e.key) } m.ll.Remove(el) delete(m.items, e.key) diff --git a/store/store_test.go b/store/store_test.go index 5f2357b..adce2ad 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -228,3 +228,21 @@ func TestPinCapDoesNotFakeRepair(t *testing.T) { t.Fatalf("repaired (%d) must never exceed dropped (%d)", r2, d2) } } + +// A replay decision that is dropped while UNPINNED (it missed the pin cap) must still be +// reported lost. Gating loss detection on the pin flag would let exactly the most +// at-risk entries vanish silently — unreported, and therefore never repaired. +func TestUnpinnedFrozenLossIsStillReported(t *testing.T) { + now := time.Unix(0, 0) + m := NewMemory(Options{TTLSeconds: 10, MaxEntries: 2}) // pin cap = 1 + m.SetClock(func() time.Time { return now }) + pinned := FrozenPrefix + "s:mask:pinned" + overCap := FrozenPrefix + "s:mask:overcap" + m.Put(pinned, []byte("a")) + m.Put(overCap, []byte("b")) // past the cap -> unpinned + now = now.Add(11 * time.Second) + m.Get(overCap) // expires it + if !m.FrozenLost(overCap) { + t.Fatal("an unpinned frozen decision that expired must still report as lost") + } +} From 2b27342edd29751659ecfa2693626edaba9cf537 Mon Sep 17 00:00:00 2001 From: Osher-Elhadad Date: Mon, 10 Aug 2026 08:58:13 +0000 Subject: [PATCH 7/7] fix(cache): exclude extract_llm from the depth repair, and make pins reclaimable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the freeze-lifetime work found the depth-repair path unsound for one of its two callers, and the pin/loss bookkeeping unsound in five more ways. All from review; each fix has a test that fails without it. extract_llm no longer gets the lost-decision repair. Its replacement is a SAMPLED model output (cheapmodel sends no temperature and no seed), so re-deriving could splice DIFFERENT bytes into the provider's cached prefix — the exact corruption the repair exists to prevent — while the observability scored it as a success. The trade argued in the previous commit was simply wrong: if the bytes differ, the suffix is cache-written either way, so re-deriving buys a model call for nothing. There is no upside. Two more hazards agreed: after lifting the gate the model may not run at all (throttle, timeout, floor), leaving the output verbatim at depth; and the entry is pinned anyway, so the common case is that it is never lost. mask/failed_run keep the repair — their replacement is prefix + headPeek(content) + Marker(sha256(content)), pure in (content, config) and position-independent, so re-deriving is genuinely reproducible. Also: - Expired entries are now evicted FIRST, pinned included. The TTL was only enforced in Get, and a finished session's decisions are never read again, so pinned entries were immortal: pinnedN ratcheted to max/2 and stayed, leaking half the cache and silently disabling pinning for every later session. A refresh can also reclaim a freed slot. - cg:len: (apply's prev-turn count, the MaxCachedIdx boundary) is pinned. It was competing for a pool this work halved, and losing it makes TailOnly fail open on every index — so the change was making that fail-open MORE likely, not less. - cg:res: and cg:sum1: are one JSON key. As two independently-TTL'd, independently- pinned keys, losing only the summary made the replay HIT and silently emit different bytes (the "[summary] " segment vanishing) with nothing reported lost. Deletes summaryKey/getSummary/putSummary. - The store no longer hardcodes component key prefixes; owners pass them via Options.PinPrefixes. - Over-cap entries are no longer marked lost at freeze time. They are present and readable, so counting them inflated frozen_dropped with live entries and made the next ordinary re-freeze look like a repair — flips reading 0 while nothing was wrong. - The loss-mark budget evicts oldest-first instead of an arbitrary key, so a busy session can no longer delete another session's fresh mark and leave it flipping unrepaired. - FrozenLossStats documents what it actually counts (drop EVENTS, a running balance), and routes.md documents all five /stats fields, including that frozen_misses is a lookup counter dominated by the ordinary "not compacted yet" case. Assisted-By: Claude Opus 5 (1M context) Signed-off-by: Osher-Elhadad --- components/all/freeze_repair_test.go | 87 +++++++++++++++ components/offload/extract_llm.go | 24 ++-- components/offload/freeze_test.go | 53 +++++---- components/offload/state.go | 68 ++++++------ docs/design.md | 36 ++++-- docs/reference/routes.md | 17 +++ store/store.go | 157 ++++++++++++++++++--------- store/store_test.go | 100 +++++++++++++++++ 8 files changed, 417 insertions(+), 125 deletions(-) create mode 100644 components/all/freeze_repair_test.go diff --git a/components/all/freeze_repair_test.go b/components/all/freeze_repair_test.go new file mode 100644 index 0000000..5831a93 --- /dev/null +++ b/components/all/freeze_repair_test.go @@ -0,0 +1,87 @@ +package all_test + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" +) + +// varyingModel returns a DIFFERENT valid projection on each call — exactly what a SAMPLED +// model may do. cheapmodel sends no temperature and no seed, so extract_llm's replacement +// text is not reproducible. +type varyingModel struct { + mu sync.Mutex + n int +} + +func (m *varyingModel) Complete(context.Context, string) (string, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.n++ + return "OUTPUT = \"projection variant " + strings.Repeat("x", m.n) + "\"\n", nil +} + +func (m *varyingModel) calls() int { + m.mu.Lock() + defer m.mu.Unlock() + return m.n +} + +// A LOST extract_llm decision must NOT be re-derived inside the provider's cached prefix. +// Re-deriving costs a sampled model call that can emit DIFFERENT bytes at depth — the very +// corruption the repair was meant to prevent — and it buys nothing: if the bytes differ the +// suffix is cache-written either way, so the model call is pure loss. The message is +// therefore left verbatim, like any other tail-gated miss. +// +// Only mask/failed_run get the depth repair, because their replacement is a pure function +// of (content, config) and so is genuinely reproducible. +func TestLostExtractLLMResultIsNotReDerivedAtDepth(t *testing.T) { + off := newComp(t, "extract_llm", + "strategy: code\nmin_tokens: 1\nmodel:\n source: config\ntrigger:\n min_request_tokens: 1\n") + now := time.Unix(0, 0) + st := store.NewMemory(store.Options{TTLSeconds: 10}) + st.SetClock(func() time.Time { return now }) + vm := &varyingModel{} + body := strings.Repeat("verbose tool output line\n", 60) + + // The message under test is Input[1]; MaxCachedIdx >= 1 puts it in the CACHED PREFIX. + run := func(maxCached int) string { + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{ + userMsg("find the keep records in output.txt"), toolMsg(body), toolMsg("tail"), + }} + c := &components.Ctx{Ctx: context.Background(), Session: "s", Store: st, + Model: components.ModelSpec{Static: vm}, CacheAware: true, + MaxCachedIdx: maxCached, CtxWindow: 200000} + var rep components.Report + if _, err := off.Offload(req, &rep, c); err != nil { + t.Fatal(err) + } + return schema.MessageText(req.Input[1]) + } + + first := run(-1) // tail turn: a NEW compaction is allowed and gets cached + if first == body { + t.Fatal("turn 1 must compact the tail output (fixture no longer exercises the path)") + } + callsAfterFirst := vm.calls() + + // Force the loss, then re-present the message at DEPTH. + now = now.Add(11 * time.Second) + got := run(1) + + if got != body { + t.Fatalf("a lost LLM decision must leave the cached-prefix message VERBATIM, not "+ + "re-project it with freshly sampled bytes:\n turn1=%q\n turn2=%q", first, got) + } + if vm.calls() > callsAfterFirst+1 { + t.Fatalf("no re-derivation call for the depth message (calls %d -> %d)", + callsAfterFirst, vm.calls()) + } +} diff --git a/components/offload/extract_llm.go b/components/offload/extract_llm.go index 4f540a2..3489165 100644 --- a/components/offload/extract_llm.go +++ b/components/offload/extract_llm.go @@ -227,8 +227,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R continue } if cached, hit := getResult(c, id); hit { - summary, _ := getSummary(c, id) - apply(i, content, string(cached), summary) + apply(i, content, cached.Projected, cached.Summary) dbgReapply++ continue } @@ -238,14 +237,14 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R // caching is off, any message is fair game. File reads included (largest mass); // safe because we never touch already-cached content and freeze+reapply the result. sz := schema.TextTokens(content) - // Exception to the tail gate: a result-cache entry this session established and the - // store then LOST. The provider already holds the compacted bytes for this message, - // so re-deriving them restores the cached representation, whereas leaving the output - // verbatim is what flips it and re-writes the suffix. (Unlike the deterministic - // offloaders this costs a model call and the LLM may not reproduce the bytes - // exactly — but the alternative is a GUARANTEED full-suffix cache-write, the more - // expensive of the two by a wide margin.) - if c.CacheAware && !c.TailOnly(i) && !repairLostResult(c, id) { + // No lost-decision repair here, unlike mask/failed_run: this replacement is a SAMPLED + // model output (cheapmodel sends no temperature/seed), so re-deriving at depth could + // emit different bytes inside the cached prefix — the very thing the repair exists to + // prevent. And the trade doesn't pay even setting that aside: if the bytes differ the + // suffix is cache-written either way, so re-deriving would buy a model call for + // nothing. The model may also not run at all (throttle, timeout, floor), which would + // leave the output verbatim at depth after the gate had already been lifted. + if c.CacheAware && !c.TailOnly(i) { dbgTail++ if sz >= floor { dbgBigTailBlocked++ // a large output we skipped ONLY because it's not in the tail @@ -310,10 +309,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R if out[k].projected == "" { continue } - putResult(c, cands[k].id, []byte(out[k].projected)) - if out[k].summary != "" { - putSummary(c, cands[k].id, out[k].summary) - } + putResult(c, cands[k].id, out[k].projected, out[k].summary) apply(cands[k].i, cands[k].content, out[k].projected, out[k].summary) } } diff --git a/components/offload/freeze_test.go b/components/offload/freeze_test.go index 08f8d3c..6c49816 100644 --- a/components/offload/freeze_test.go +++ b/components/offload/freeze_test.go @@ -152,37 +152,48 @@ func TestFrozenCountersMove(t *testing.T) { } } -// The result cache (cg:res:) is the OTHER replay namespace — extract_llm's — and it is -// the one that carries the load in the shipped coding config (no mask; failed_run -// self-skips on a cached agent). It must get the same protection: pinned against -// eviction, and its loss reported so the compaction can be re-derived at depth. -func TestResultCachePinnedAndRepairable(t *testing.T) { - now := time.Unix(0, 0) - st := store.NewMemory(store.Options{TTLSeconds: 10, MaxEntries: 4}) - st.SetClock(func() time.Time { return now }) +// The result cache (cg:res:) is extract_llm's replay namespace. It IS pinned against +// eviction — losing it un-compacts an already-cached message like any other replay +// decision — but it deliberately gets NO depth repair, because re-deriving it means a +// sampled model call (see repairLostFreeze). +func TestResultCachePinnedAgainstEviction(t *testing.T) { + st := store.NewMemory(store.Options{MaxEntries: 4}) c := &components.Ctx{Session: "s", Store: st} - putResult(c, "id1", []byte("compacted")) + putResult(c, "id1", "compacted", "one-line summary") // Ordinary rewind stashes churn through the cache; the replay decision must survive. for i := 0; i < 20; i++ { st.Put("rewindhash"+string(rune('a'+i)), []byte("big original payload")) } - if _, ok := getResult(c, "id1"); !ok { + got, ok := getResult(c, "id1") + if !ok { t.Fatal("a result-cache decision must be pinned against LRU eviction") } - if repairLostResult(c, "id1") { - t.Fatal("a live decision is not lost") + if got.Projected != "compacted" || got.Summary != "one-line summary" { + t.Fatalf("projection and summary must survive together, got %+v", got) } - // Expire it: nothing reads it for longer than the TTL. +} + +// The projection and its summary line must live and die as ONE key. As two independently +// TTL'd/pinned keys, losing only the summary made the replay HIT and silently emit +// different bytes (the "[summary] " segment vanishing) inside the cached prefix. +func TestResultAndSummaryShareOneKey(t *testing.T) { + now := time.Unix(0, 0) + st := store.NewMemory(store.Options{TTLSeconds: 10}) + st.SetClock(func() time.Time { return now }) + c := &components.Ctx{Session: "s", Store: st} + putResult(c, "id1", "compacted", "summary") + + // Whatever the store drops, a replay either returns BOTH parts or misses entirely — + // it can never return a projection with the summary silently missing. now = now.Add(11 * time.Second) - if _, ok := getResult(c, "id1"); ok { - t.Fatal("expired entry must miss") - } - if !repairLostResult(c, "id1") { - t.Fatal("a LOST result-cache decision must be distinguishable from never-cached") + if got, ok := getResult(c, "id1"); ok { + t.Fatalf("expired decision must miss outright, got %+v", got) } - if repairLostResult(c, "never-seen") { - t.Fatal("content that was never compacted must not authorize depth mutation") + // A half-written / unreadable payload is also treated as absent, never spliced. + st.Put(resultKey("s", "id2"), []byte("{not json")) + if got, ok := getResult(c, "id2"); ok { + t.Fatalf("unreadable decision must miss, got %+v", got) } } @@ -196,7 +207,7 @@ func TestResultCacheFeedsCounters(t *testing.T) { if _, ok := getResult(c, "idz"); ok { t.Fatal("nothing cached yet") } - putResult(c, "idz", []byte("compacted")) + putResult(c, "idz", "compacted", "") if _, ok := getResult(c, "idz"); !ok { t.Fatal("expected a replay hit") } diff --git a/components/offload/state.go b/components/offload/state.go index 0760f87..7d868d0 100644 --- a/components/offload/state.go +++ b/components/offload/state.go @@ -23,24 +23,41 @@ import ( // resultKey namespaces a per-content reduced output (extract) by session. func resultKey(session, id string) string { return store.ResultPrefix + session + ":" + id } +// cachedResult is the whole replayed decision for one content id: the compacted +// projection AND the one-line summary re-emitted beside it. They live under ONE key +// because they must live and die together — as two independently-TTL'd, independently- +// pinned keys, losing only the summary made the replay HIT and emit different bytes (the +// "[summary] " segment silently vanishing) with nothing reported lost. +type cachedResult struct { + Projected string `json:"p"` + Summary string `json:"s,omitempty"` +} + // getResult returns a previously cached reduced output for content id, if any. This is // extract_llm's replay lookup, so it feeds the same hit/miss counters as reapplyFrozen — // otherwise the shipped coding config (no mask, failed_run self-skipping) would report // zero freeze activity while doing all of its replay through here. -func getResult(c *components.Ctx, id string) ([]byte, bool) { - v, ok := c.Store.Get(resultKey(c.Session, id)) - if ok { - frozenHits.Add(1) - } else { +func getResult(c *components.Ctx, id string) (cachedResult, bool) { + b, ok := c.Store.Get(resultKey(c.Session, id)) + if !ok { frozenMisses.Add(1) + return cachedResult{}, false + } + var r cachedResult + if json.Unmarshal(b, &r) != nil || r.Projected == "" { + frozenMisses.Add(1) // unreadable => treat as absent, never splice half a decision + return cachedResult{}, false } - return v, ok + frozenHits.Add(1) + return r, true } // putResult caches a reduced output so a later turn re-sending the same content // reuses it (no LLM call, byte-identical result). -func putResult(c *components.Ctx, id string, v []byte) { - c.Store.Put(resultKey(c.Session, id), v) +func putResult(c *components.Ctx, id, projected, summary string) { + if b, err := json.Marshal(cachedResult{Projected: projected, Summary: summary}); err == nil { + c.Store.Put(resultKey(c.Session, id), b) + } } // --- Freeze + reapply (cache stability) ------------------------------------- @@ -119,19 +136,22 @@ func reapplyFrozen(c *components.Ctx, comp string, m *bschemas.ChatMessage) ([]s // Re-deriving reproduces the bytes the provider already cached; NOT re-deriving is what // flips the representation and re-writes the suffix. The caller's own never-worse and // skipReduce guards still apply, so this only ever LIFTS the depth restriction. +// +// ONLY safe for offloaders whose replacement is a pure function of (content, config): +// mask and failed_run build `prefix + headPeek(content) + Marker(sha256(content))`, which +// is position-independent — their windows (keep_recent, runs[:len-1]) gate WHETHER the +// component considers a message, never WHAT bytes it emits, and config cannot drift +// mid-session (no hot reload; a restart wipes the store with it). +// +// It is deliberately NOT offered to extract_llm. That replacement is a SAMPLED model +// output (cheapmodel sends no temperature/seed), so re-deriving may emit different bytes +// at depth — and the trade does not work even ignoring that: if the bytes differ, the +// suffix is cache-written either way, so the repair branch pays a model call for nothing. +// A lost extract_llm decision therefore just declines, like any other tail-gated miss. func repairLostFreeze(c *components.Ctx, comp, content string) bool { return frozenLost(c, frozenKey(c.Session, comp, contentKey(content))) } -// repairLostResult is repairLostFreeze for the OTHER replay namespace: extract_llm's -// per-content result cache (cg:res:), which is the same replay contract under a different -// name — and the one that actually carries the load in the shipped coding config, where -// mask is absent and failed_run self-skips on a cached agent. A lost result cache entry -// un-compacts an already-cached message exactly the same way, so it gets the same repair. -func repairLostResult(c *components.Ctx, id string) bool { - return frozenLost(c, resultKey(c.Session, id)) -} - // Freeze-replay counters: how often a replay landed vs found nothing. Cache-write is the // largest cost line on long-horizon traffic and a lost freeze is the mechanism that // produces it, so the store counts the drops and the repairs (a re-Put of a dropped @@ -210,20 +230,6 @@ func OwnsKey(st store.Store, session, key string) bool { return ok } -// summaryKey namespaces the one-line SUMMARY the LLM extract emitted for a content -// id, so a later turn reusing the cached reduction also re-emits the same marker -// digest (byte-stable) without re-calling the model. -func summaryKey(session, id string) string { return store.SummaryPrefix + session + ":" + id } - -func getSummary(c *components.Ctx, id string) (string, bool) { - b, ok := c.Store.Get(summaryKey(c.Session, id)) - return string(b), ok -} - -func putSummary(c *components.Ctx, id, s string) { - c.Store.Put(summaryKey(c.Session, id), []byte(s)) -} - // sumCheckpoint is the per-session summarize state: the exact summary message // text produced last time (re-emitted verbatim so the prefix stays byte-stable), // how many leading span messages it subsumed, a hash of that span to prove the diff --git a/docs/design.md b/docs/design.md index bf104c8..8f1645d 100644 --- a/docs/design.md +++ b/docs/design.md @@ -240,9 +240,14 @@ flip representation inside the provider's cached prefix, and the whole suffix is turn never ages out; one nobody reads still expires on its original deadline. - **Default 10000s** — Terminal-Bench tasks average ~1975s of wall clock and run to 4h, so the old 1800s default expired live decisions mid-task. Still `store.ttl_seconds`. -- **Frozen decisions are pinned** against LRU eviction (they are a marker line each), capped at - half the entry cap so one pathological session cannot pin the whole cache and starve the rewind - stashes the expand loop needs. +- **Replay decisions are pinned** against LRU eviction — `cg:frz:` (mask/failed_run), `cg:res:` + (extract_llm's projection *and* its summary line, one key so they cannot half-survive) and + `cg:len:` (apply's cache-boundary counter, whose loss makes `TailOnly` fail open). All are tiny; + the pin is capped at half the entry cap so one session cannot starve the rewind stashes the + expand loop needs. Eviction reclaims **expired** entries first, pinned included — otherwise a + finished session's decisions are never read again, never expire, and permanently occupy the pin + budget. The prefixes are supplied by their owners via `store.Options.PinPrefixes`; the store + does not know component key layouts. **The fail direction inverts for an established compaction.** Fail-open normally means "forward the original", and for a *new* compaction that is right. But once the provider has cached the compacted @@ -251,18 +256,31 @@ apart, so the store keeps the *fact* of a dropped freeze (`FrozenLoser.FrozenLos set — the payload need not survive, only the knowledge that it existed): - **never frozen** → obey the tail gate; a new compaction stays in the uncached tail. -- **frozen, then lost** → re-derive it even at depth. An offloader's replacement text is a pure - function of `(content, component config)` and the marker key is `sha256(original)`, so - re-deriving reproduces the *same* bytes the provider cached and re-establishes the freeze. The - component's own never-worse and kept-verbatim guards still apply, so this only ever lifts the +- **frozen, then lost** → re-derive it even at depth, but **only where re-derivation is + reproducible**. `mask` and `failed_run` qualify: their replacement is + `prefix + headPeek(content) + Marker(sha256(content))`, a pure function of + `(content, config)` and independent of position, so re-deriving reproduces the *same* bytes the + provider cached and re-establishes the freeze. Their windows (`keep_recent`, `runs[:len-1]`) gate + *whether* a message is considered, never *what bytes* are emitted, and config cannot drift + mid-session. The never-worse and kept-verbatim guards still apply, so the repair only lifts the depth restriction — it never authorizes new content loss. +- **`extract_llm` is deliberately excluded.** Its replacement is a *sampled* model output (the + cheap-model client sends no temperature and no seed), so re-deriving could splice **different** + bytes into the cached prefix — the exact corruption the repair exists to prevent. And the trade + does not pay even ignoring that: if the bytes differ, the suffix is cache-written either way, so + the repair branch would buy a model call for nothing. There is no upside, so a lost `extract_llm` + decision simply declines and the message is forwarded verbatim. (Its entry is still pinned, so + the common case is that it is never lost at all.) Re-enabling it would need deterministic + decoding *plus* a check that the re-derived bytes match the stored hash before splicing. `/stats` reports `frozen_hits`, `frozen_misses`, `frozen_dropped`, `frozen_repaired`, and -`frozen_flips` (= dropped − repaired, the drops that actually cost a cache-write; it should be 0). +`frozen_flips` (= dropped − repaired; should be 0). `frozen_misses` is a *lookup* counter dominated +by the ordinary "not compacted yet" case — `frozen_dropped` is the one that measures harm. See +[Routes](reference/routes.md#stats-freeze-replay-fields). The related fail-*open* on `MaxCachedIdx`: `prevLen` returning 0 on a store miss yields `MaxCachedIdx = -1`, and `Ctx.TailOnly` then permits mutating any index (measured on 11.2% of -Terminal-Bench requests). The sliding TTL shrinks that window — `cg:len:` is read every turn, so it +Terminal-Bench requests). `cg:len:` is now pinned and the sliding TTL keeps it alive, so it no longer expires mid-session — but inverting `TailOnly` to fail *closed* is a separate change. ## Session keying diff --git a/docs/reference/routes.md b/docs/reference/routes.md index 9c8730e..4f0b532 100644 --- a/docs/reference/routes.md +++ b/docs/reference/routes.md @@ -12,6 +12,23 @@ The proxy serves both provider dialects on one port (default `:4000`). | `GET /stats` | Savings rollups (token-weighted `Σ saved / Σ before`, plus `wasted_tokens`/`bounces` and per-component breakdown). | | `GET /expand?id=` | Recover an offloaded original by its `<>` id. | +### `/stats` freeze-replay fields + +Fields are only ever **added** to this payload (harnesses in `deploy/harbor` parse it), so +a consumer that reads by key keeps working. + +| Field | Meaning | +|---|---| +| `frozen_hits` | Replay lookups that found a stored decision and re-sent the same bytes. | +| `frozen_misses` | Replay lookups that found nothing. **Dominated by the ordinary "not compacted yet" case** — it is a lookup counter, not an error counter. Read `frozen_dropped` for harm. | +| `frozen_dropped` | Stored decisions the store actually **lost** (TTL expiry or eviction). Each is a chance for an already-cached message to flip representation. Counted per drop *event*. | +| `frozen_repaired` | Dropped decisions later restored, so a replay can land again. | +| `frozen_flips` | `frozen_dropped − frozen_repaired` — outstanding losses, i.e. drops that plausibly cost a suffix cache-write. **Should be 0.** | + +A healthy long-horizon run shows `frozen_hits` climbing with turn count and +`frozen_dropped` at 0; a rising `frozen_dropped` means decisions are dying mid-session +(TTL too short for the task, or the entry cap too small for the session's working set). + !!! note "`POST /compact` (compaction-service mode)" The [llm-d compaction service example](../examples/llm-d-service.md) adds a stateless `POST /compact` route: it runs the pipeline and returns the diff --git a/store/store.go b/store/store.go index 395ea9e..08c55be 100644 --- a/store/store.go +++ b/store/store.go @@ -48,31 +48,38 @@ type FrozenLoser interface { // Key namespaces whose entries are a component's FROZEN decision — the replacement // text it must replay on every later turn to keep an already-cached message -// byte-identical (see components/offload/state.go). Two components' worth, because the -// freeze-replay mechanism was implemented twice under different names: -// -// cg:frz: — mask / failed_run (freeze + reapplyFrozen) -// cg:res: — extract_llm's result cache, plus cg:sum1: for the summary line it -// re-emits alongside it. Functionally the same replay contract. +// byte-identical (see components/offload/state.go), plus the small per-session trackers +// the cache-safety machinery itself depends on. // // Entries under these prefixes are PINNED: exempt from LRU eviction, because losing one // is not a cache miss, it is a cache-DESTRUCTIVE event — the message flips representation // inside the provider's cached prefix and the whole suffix is re-written at 11.5x the -// read price. They are small (a marker line / a compacted projection), still honor the -// sliding TTL, and the exemption is capped at half the entry cap so a pathological -// session can never pin the whole cache. The rewind stashes (bare content hashes, the -// large payloads the expand loop resolves) stay fully evictable. +// read price. They are small (a marker line, a compacted projection, an integer), still +// honor the sliding TTL, and the exemption is capped at half the entry cap so a +// pathological session can never pin the whole cache. The rewind stashes (bare content +// hashes, the large payloads the expand loop resolves) stay fully evictable. +// +// The prefixes are declared by their OWNERS (components/offload, apply) and passed in via +// Options.PinPrefixes — the store must not know what a component names its keys. const ( - FrozenPrefix = "cg:frz:" - ResultPrefix = "cg:res:" - SummaryPrefix = "cg:sum1:" + FrozenPrefix = "cg:frz:" // mask / failed_run freeze decisions + ResultPrefix = "cg:res:" // extract_llm's replayed result (projection + summary, one key) + LenPrefix = "cg:len:" // apply's prev-turn message count (the MaxCachedIdx boundary) ) -// frozenNamespace reports whether key holds a replay decision (see FrozenPrefix). -func frozenNamespace(key string) bool { - return strings.HasPrefix(key, FrozenPrefix) || - strings.HasPrefix(key, ResultPrefix) || - strings.HasPrefix(key, SummaryPrefix) +// DefaultPinPrefixes is the shipped set of key namespaces whose loss is cache-destructive. +// Callers that build their own Store may pass a different set; the zero value means "none", +// so a host that opts out simply gets plain TTL+LRU. +var DefaultPinPrefixes = []string{FrozenPrefix, ResultPrefix, LenPrefix} + +// pinned reports whether key belongs to one of the configured pin namespaces. +func (m *Memory) isPinPrefix(key string) bool { + for _, p := range m.pinPrefixes { + if strings.HasPrefix(key, p) { + return true + } + } + return false } type entry struct { @@ -88,19 +95,21 @@ type entry struct { // mirroring headroom's 1800s CCR store: a frozen compaction that dies mid-task is // a cache-destructive event, not a saving. type Memory struct { - mu sync.Mutex - ttl time.Duration - max int - ll *list.List // LRU, front = most recent - items map[string]*list.Element // key -> element(*entry) - sticky map[string]map[string]struct{} - maxStick int - now func() time.Time // injectable for tests - pinnedN int // live pinned (frozen) entries, capped at max/2 + mu sync.Mutex + ttl time.Duration + max int + ll *list.List // LRU, front = most recent + items map[string]*list.Element // key -> element(*entry) + sticky map[string]map[string]struct{} + maxStick int + pinPrefixes []string + now func() time.Time // injectable for tests + pinnedN int // live pinned (frozen) entries, capped at max/2 // lostFrozen remembers keys whose FROZEN entry was dropped anyway (TTL expiry, or // the pin cap). It is the "was frozen, now LOST" signal a caller cannot otherwise // distinguish from "never frozen" — see FrozenLost. Bounded like sticky. lostFrozen map[string]struct{} + lostOrder []string // insertion order, so the OLDEST mark is evicted first lostN int64 repairedN int64 noSlide bool // tests only: restore the old write-only expiry (see DisableSlidingTTLForTest) @@ -112,10 +121,15 @@ type Options struct { // Enabled toggles the state store. nil/absent => on (backward-compatible). // false => no store: reversibility is off, so offload components must run // marker_mode: off (a full-marker offload would leave dangling markers). - Enabled *bool `yaml:"enabled"` - TTLSeconds int `yaml:"ttl_seconds"` - MaxEntries int `yaml:"max_entries"` - MaxSessions int `yaml:"max_sessions"` + Enabled *bool `yaml:"enabled"` + TTLSeconds int `yaml:"ttl_seconds"` + // PinPrefixes are key namespaces whose entries are exempt from LRU eviction because + // losing one is cache-destructive rather than merely a miss (see FrozenPrefix). nil => + // DefaultPinPrefixes. Not a yaml knob: it is a code-level property of the key layout, + // not something an operator should be tuning. + PinPrefixes []string `yaml:"-"` + MaxEntries int `yaml:"max_entries"` + MaxSessions int `yaml:"max_sessions"` } // Nop is a Store that persists nothing: Put discards, Get/Sticky always miss. @@ -150,8 +164,12 @@ func NewMemory(o Options) *Memory { if stick <= 0 { stick = 100 } + pins := o.PinPrefixes + if pins == nil { + pins = DefaultPinPrefixes + } return &Memory{ - ttl: ttl, max: max, maxStick: stick, + ttl: ttl, max: max, maxStick: stick, pinPrefixes: pins, ll: list.New(), items: map[string]*list.Element{}, sticky: map[string]map[string]struct{}{}, lostFrozen: map[string]struct{}{}, @@ -188,25 +206,38 @@ func (m *Memory) Put(key string, payload []byte) { // exceed dropped, or frozen_flips reads 0 while messages are in fact flipping. if _, wasLost := m.lostFrozen[key]; wasLost { delete(m.lostFrozen, key) + for i, k := range m.lostOrder { + if k == key { + m.lostOrder = append(m.lostOrder[:i], m.lostOrder[i+1:]...) + break + } + } m.repairedN++ } if el, ok := m.items[key]; ok { e := el.Value.(*entry) e.payload = payload e.expires = m.now().Add(m.ttl) + // Claim a pin slot if one has since freed (an earlier session's decisions expired): + // the cap is a live-entry budget, not a lifetime quota, so re-freezing every turn + // eventually protects this decision instead of leaving it permanently second-class. + if !e.pinned && m.isPinPrefix(key) && !m.noSlide && m.pinnedN < m.max/2 { + e.pinned = true + m.pinnedN++ + } m.ll.MoveToFront(el) return } e := &entry{key: key, payload: payload, expires: m.now().Add(m.ttl)} // Pin frozen decisions, but never more than half the cache: past that the marginal // pin protects one message while starving the rewind stashes the expand loop needs. - if frozenNamespace(key) && !m.noSlide { - if m.pinnedN < m.max/2 { - e.pinned = true - m.pinnedN++ - } else { - m.noteLost(key) // pin cap reached: evictable, and its loss stays visible - } + // Over the cap the entry is simply evictable — NOT recorded as lost: it is present and + // readable right now, and calling it "dropped" at write time both inflates the drop + // count with live entries and makes the very next re-freeze look like a repair. Its + // loss, if it comes, is recorded where losses actually happen (remove). + if m.isPinPrefix(key) && !m.noSlide && m.pinnedN < m.max/2 { + e.pinned = true + m.pinnedN++ } m.items[key] = m.ll.PushFront(e) for m.ll.Len() > m.max { @@ -217,15 +248,23 @@ func (m *Memory) Put(key string, payload []byte) { } // noteLost records that a frozen decision under key is gone, so a later Get miss is -// distinguishable from "never frozen". Bounded by the entry cap. +// distinguishable from "never frozen". Bounded by the entry cap, evicting the OLDEST mark +// first: dropping an arbitrary one let a busy session delete another session's fresh mark, +// so that session's next turn saw a plain miss and flipped its message unrepaired. +// ponytail: FIFO over one shared budget, not a per-session quota — the marks are +// session-scoped keys and short-lived (cleared by the next re-freeze), so age is a good +// enough proxy. Revisit if one session's churn is ever shown to starve another's. func (m *Memory) noteLost(key string) { - if len(m.lostFrozen) >= m.max { - for k := range m.lostFrozen { - delete(m.lostFrozen, k) - break - } + if _, dup := m.lostFrozen[key]; dup { + return // already marked; don't double-count or re-queue + } + for len(m.lostFrozen) >= m.max && len(m.lostOrder) > 0 { + oldest := m.lostOrder[0] + m.lostOrder = m.lostOrder[1:] + delete(m.lostFrozen, oldest) } m.lostFrozen[key] = struct{}{} + m.lostOrder = append(m.lostOrder, key) m.lostN++ } @@ -239,10 +278,14 @@ func (m *Memory) FrozenLost(key string) bool { } // FrozenLossStats returns how many frozen decisions this store has DROPPED since start -// (TTL expiry / pin cap) and how many of those were later re-Put — repaired to the same -// bytes, so no representation flip reached the provider. dropped−repaired is the count of -// flips that actually cost a suffix cache-write. Both count each key once, however many -// turns observe it. +// (TTL expiry, or eviction) and how many of those were later re-Put — restored to the +// store, so a replay can land again instead of the message flipping. +// +// Counted per DROP EVENT, not per distinct key: one key that expires, is re-frozen, and +// expires again contributes 2 drops and 1 repair. That is the intended reading — each +// event is a separate opportunity for a flip — but it means dropped−repaired is a running +// balance (marks still outstanding), not a total of distinct broken keys. A repeat drop of +// an already-marked key is not double-counted while its mark is outstanding. func (m *Memory) FrozenLossStats() (dropped, repaired int64) { m.mu.Lock() defer m.mu.Unlock() @@ -306,6 +349,20 @@ func (m *Memory) MarkSticky(session, id string) { // evictOldest drops the least-recently-used UNPINNED entry, walking back over pinned // (frozen) ones. Reports false when nothing is evictable. func (m *Memory) evictOldest() bool { + now := m.now() + // Pass 1: reclaim anything already EXPIRED, pinned included. Without this a pinned + // entry is immortal — the TTL is only enforced in Get, and a dead session is never + // read again — so pinnedN would ratchet to max/2 and stay there, leaking half the + // cache and silently disabling pinning for every later session. + for el := m.ll.Back(); el != nil; { + prev := el.Prev() + if e := el.Value.(*entry); now.After(e.expires) { + m.remove(el) + return true + } + el = prev + } + // Pass 2: nothing expired, so take the LRU entry that is not pinned. for el := m.ll.Back(); el != nil; el = el.Prev() { if !el.Value.(*entry).pinned { m.remove(el) @@ -324,7 +381,7 @@ func (m *Memory) remove(el *list.Element) { // the pin flag. An entry that missed the pin cap is exactly the one most likely to be // dropped, and gating this on e.pinned would let it vanish silently: unreported, and so // never repaired. (noSlide reproduces the OLD store, which had no loss signal at all.) - if frozenNamespace(e.key) && !m.noSlide { + if m.isPinPrefix(e.key) && !m.noSlide { m.noteLost(e.key) } m.ll.Remove(el) diff --git a/store/store_test.go b/store/store_test.go index adce2ad..fb9d6c7 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -246,3 +246,103 @@ func TestUnpinnedFrozenLossIsStillReported(t *testing.T) { t.Fatal("an unpinned frozen decision that expired must still report as lost") } } + +// Pinned entries must not be immortal. The TTL is only enforced in Get, and a dead +// session's decisions are never read again, so without expiry-aware eviction pinnedN +// ratchets to max/2 and stays there: half the cache leaks AND pinning silently stops +// working for every later session. +func TestExpiredPinnedEntriesAreReclaimed(t *testing.T) { + now := time.Unix(0, 0) + m := NewMemory(Options{TTLSeconds: 10, MaxEntries: 20}) // pin cap = 10 + m.SetClock(func() time.Time { return now }) + for i := 0; i < 10; i++ { // fill every pin slot from a "dead" session + m.Put(FrozenPrefix+"dead:mask:"+string(rune('a'+i)), []byte("f")) + } + if m.pinnedN != 10 { + t.Fatalf("expected 10 pinned, got %d", m.pinnedN) + } + now = now.Add(11 * time.Second) // the dead session's decisions are all past their TTL + for i := 0; i < 30; i++ { // a new session's traffic drives eviction + m.Put("rewind"+string(rune('a'+i)), []byte("payload")) + } + if m.pinnedN >= 10 { + t.Fatalf("expired pinned entries must be reclaimed, pinnedN=%d", m.pinnedN) + } + if m.ll.Len() > 20 { + t.Fatalf("entry cap breached, len=%d", m.ll.Len()) + } + // A fresh session can pin again, because slots actually freed. + m.Put(FrozenPrefix+"live:mask:x", []byte("f")) + if el := m.items[FrozenPrefix+"live:mask:x"]; el == nil || !el.Value.(*entry).pinned { + t.Fatal("a new session must be able to pin after old decisions expired") + } +} + +// cg:len: is apply's prev-turn message count — the MaxCachedIdx boundary. Losing it makes +// TailOnly return true for EVERY index (fail-open, mutating the cached prefix), so it must +// be pinned too. It is 2-4 bytes. +func TestLenTrackerIsPinned(t *testing.T) { + m := NewMemory(Options{MaxEntries: 20}) + m.Put(LenPrefix+"sess", []byte("42")) + for i := 0; i < 10; i++ { + m.Put(FrozenPrefix+"s:mask:"+string(rune('a'+i)), []byte("f")) + } + for i := 0; i < 40; i++ { + m.Put("rewind"+string(rune('a'+i)), []byte("payload")) + } + if got, ok := m.Get(LenPrefix + "sess"); !ok || string(got) != "42" { + t.Fatal("the cache-boundary tracker must survive eviction pressure (else TailOnly fails open)") + } +} + +// An entry that is present and readable is not "dropped". Counting it at write time (when +// it merely missed the pin cap) inflated the drop count with live entries and made the +// next ordinary re-freeze look like a repair — flips reading 0 while nothing was wrong. +func TestOverCapEntryIsNotCountedAsDropped(t *testing.T) { + m := NewMemory(Options{MaxEntries: 4}) // pin cap = 2 + for i := 0; i < 6; i++ { + m.Put(FrozenPrefix+"s:mask:"+string(rune('a'+i)), []byte("f")) + } + dropped, repaired := m.FrozenLossStats() + // Whatever was evicted is a real drop; nothing was re-frozen, so repaired must be 0. + if repaired != 0 { + t.Fatalf("no key was re-frozen, so repaired must be 0, got %d (dropped=%d)", repaired, dropped) + } + // Re-freezing the same over-cap key twice must not manufacture a drop/repair pair. + before, _ := m.FrozenLossStats() + k := FrozenPrefix + "s:mask:f" + m.Put(k, []byte("f2")) + m.Put(k, []byte("f3")) + after, rep2 := m.FrozenLossStats() + if after != before || rep2 != 0 { + t.Fatalf("re-freezing a LIVE over-cap key must not count drops/repairs: %d->%d rep=%d", + before, after, rep2) + } +} + +// The loss marks are one shared budget, so eviction must not let a busy session delete +// another session's fresh mark — that session's next turn would see a plain miss and flip +// its message unrepaired. Oldest-first keeps the newest marks. +func TestLossMarkEvictionKeepsNewest(t *testing.T) { + now := time.Unix(0, 0) + m := NewMemory(Options{TTLSeconds: 10, MaxEntries: 2}) + m.SetClock(func() time.Time { return now }) + old := FrozenPrefix + "A:mask:old" + m.Put(old, []byte("f")) + now = now.Add(11 * time.Second) + m.Get(old) // expire -> mark A's loss + if !m.FrozenLost(old) { + t.Fatal("A's loss must be marked") + } + // Session B churns enough losses to overflow the mark budget (max=2). + for i := 0; i < 5; i++ { + k := FrozenPrefix + "B:mask:" + string(rune('a'+i)) + m.Put(k, []byte("f")) + now = now.Add(11 * time.Second) + m.Get(k) + } + newest := FrozenPrefix + "B:mask:e" + if !m.FrozenLost(newest) { + t.Fatal("the NEWEST loss mark must survive the budget (oldest is evicted first)") + } +}