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 b1618eb..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,6 +237,13 @@ 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) + // 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 { @@ -303,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/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/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/components/offload/freeze_test.go b/components/offload/freeze_test.go new file mode 100644 index 0000000..6c49816 --- /dev/null +++ b/components/offload/freeze_test.go @@ -0,0 +1,218 @@ +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) + } +} + +// 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", "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")) + } + got, ok := getResult(c, "id1") + if !ok { + t.Fatal("a result-cache decision must be pinned against LRU eviction") + } + if got.Projected != "compacted" || got.Summary != "one-line summary" { + t.Fatalf("projection and summary must survive together, got %+v", got) + } +} + +// 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 got, ok := getResult(c, "id1"); ok { + t.Fatalf("expired decision must miss outright, got %+v", got) + } + // 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) + } +} + +// 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", "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/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..7d868d0 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" @@ -20,17 +21,43 @@ 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) { - return c.Store.Get(resultKey(c.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) (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 + } + 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) ------------------------------------- @@ -44,7 +71,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 +81,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, key string) bool { + fl, ok := c.Store.(store.FrozenLoser) + return ok && fl.FrozenLost(key) +} + // 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 +114,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 +131,41 @@ 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. +// +// 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))) +} + +// 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) } @@ -142,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 "cg:sum1:" + 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 3d5c3c0..8f1645d 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,61 @@ 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`. +- **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 +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, 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; 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). `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 `session.Resolve(explicit, system, firstUser)`: an explicit host id wins; otherwise a stable @@ -265,7 +321,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/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/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..08c55be 100644 --- a/store/store.go +++ b/store/store.go @@ -13,6 +13,7 @@ package store import ( "container/list" + "strings" "sync" "time" ) @@ -34,24 +35,84 @@ 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 +} + +// 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), 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, 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:" // 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) +) + +// 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 { 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 - 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 + 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) } // Options configures a Memory store; the zero value yields sane defaults. @@ -60,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. @@ -77,12 +143,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 { @@ -92,31 +164,132 @@ 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{}{}, - 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 +} + +// 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() + // 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) + 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. + // 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 { - 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, 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 _, 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++ +} + +// 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, 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() + return m.lostN, m.repairedN } func (m *Memory) Get(key string) ([]byte, bool) { @@ -131,6 +304,14 @@ 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. + if !m.noSlide { + e.expires = m.now().Add(m.ttl) + } m.ll.MoveToFront(el) return e.payload, true } @@ -165,13 +346,44 @@ 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 { + 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) + return true + } } + return false } func (m *Memory) remove(el *list.Element) { + e := el.Value.(*entry) + if e.pinned { + m.pinnedN-- + } + // 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 m.isPinPrefix(e.key) && !m.noSlide { + m.noteLost(e.key) + } 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..fb9d6c7 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -71,3 +71,278 @@ 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) + } +} + +// 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) + } +} + +// 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") + } +} + +// 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)") + } +}