diff --git a/cmd/context-guru-proxy/effectiveconfig_test.go b/cmd/context-guru-proxy/effectiveconfig_test.go new file mode 100644 index 00000000..6e86eebc --- /dev/null +++ b/cmd/context-guru-proxy/effectiveconfig_test.go @@ -0,0 +1,59 @@ +package main + +import ( + "testing" + + "github.com/rossoctl/context-guru/config" + "github.com/rossoctl/context-guru/store" +) + +// /config MUST PUBLISH WHAT THE STORE WILL USE. +// +// stash_ttl_seconds is capped at ttl_seconds inside NewMemory, silently. effectiveConfig published +// the CONFIGURED field, so `stash_ttl_seconds: 20000` with `ttl_seconds: 10000` showed 20000 on +// /config and the dashboard while the store used 10000 — an operator told one number while another +// runs, which is #200's shape in the config surface instead of the metrics one. +// +// This test lives HERE rather than in store, and that is the point of it: a store-side test that +// compares store.EffectiveStashTTLSeconds against a store built from the same Options is true by +// construction and cannot fail, so main.go could regress with every store test still green. The +// drift is between the PUBLISHED map and the store, so the assertion has to span both. +func TestConfigPublishesTheEffectivePayloadHorizon(t *testing.T) { + for _, tc := range []struct { + name string + yaml string + want int + }{ + // The case that diverged. + {"a payload horizon over the decision horizon", "store: {ttl_seconds: 10000, stash_ttl_seconds: 20000}\n", 10000}, + // And the ones that did not, which must keep working. + {"a shorter payload horizon", "store: {ttl_seconds: 10000, stash_ttl_seconds: 600}\n", 600}, + {"defaults", "store: {}\n", int(store.DefaultStashTTL.Seconds())}, + } { + t.Run(tc.name, func(t *testing.T) { + cfg, err := config.LoadBytes([]byte("preset: codesafe\n" + tc.yaml)) + if err != nil { + t.Fatal(err) + } + eff := effectiveConfig(cfg, ":0", "", "", "", "", false, "") + st, ok := eff["store"].(map[string]any) + if !ok { + t.Fatalf("no store block in the published config: %#v", eff) + } + got, ok := st["stash_ttl_seconds"].(int) + if !ok { + t.Fatalf("stash_ttl_seconds is %T, not an int: %#v", st["stash_ttl_seconds"], st) + } + if got != tc.want { + t.Errorf("/config publishes stash_ttl_seconds=%d, want %d — the dashboard would "+ + "advertise a horizon the store does not use", got, tc.want) + } + // The published value and the store's must be the same number, whatever it is. This is + // the assertion that actually spans the drift: it reads the map main.go builds and the + // store the same Options produce. + if want := store.EffectiveStashTTLSeconds(cfg.Store); got != want { + t.Errorf("/config publishes %d but a store from the same Options uses %d", got, want) + } + }) + } +} diff --git a/cmd/context-guru-proxy/main.go b/cmd/context-guru-proxy/main.go index f2eeeaa0..1f00c464 100644 --- a/cmd/context-guru-proxy/main.go +++ b/cmd/context-guru-proxy/main.go @@ -37,6 +37,7 @@ import ( "github.com/rossoctl/context-guru/internal/modelinfo" "github.com/rossoctl/context-guru/metrics" "github.com/rossoctl/context-guru/proxy" + "github.com/rossoctl/context-guru/store" "github.com/rossoctl/context-guru/tenant" ) @@ -916,8 +917,14 @@ func effectiveConfig(cfg *config.Config, addr, openai, anthropic, bob, dbPath st "inject_expand": envOr("INJECT_EXPAND", "auto"), "cheap_model": os.Getenv("CHEAP_MODEL"), "cheap_model_provider": envOr("CHEAP_MODEL_PROVIDER", "anthropic"), + // stash_ttl_seconds is the EFFECTIVE value, not the configured one: it is capped at + // ttl_seconds, silently, so publishing the raw field showed 20000 on the dashboard while + // the store used 10000. A config surface that disagrees with the running store is the same + // silent divergence #200 is about. The other three need no such treatment — their defaults + // are filled in but nothing overrides an explicit value. "store": map[string]any{"ttl_seconds": cfg.Store.TTLSeconds, - "max_entries": cfg.Store.MaxEntries, "stash_max_bytes": cfg.Store.StashMaxBytes}, + "max_entries": cfg.Store.MaxEntries, "stash_max_bytes": cfg.Store.StashMaxBytes, + "stash_ttl_seconds": store.EffectiveStashTTLSeconds(cfg.Store)}, "dashboard": map[string]any{"db_path": dbPath, "capture_content": content, "trusted_cidrs": cidrs}, "build_version": buildinfo.Version, "build_commit": buildinfo.Commit, diff --git a/components/offload/payload_rederive_test.go b/components/offload/payload_rederive_test.go new file mode 100644 index 00000000..0e9da036 --- /dev/null +++ b/components/offload/payload_rederive_test.go @@ -0,0 +1,156 @@ +package offload + +import ( + "strings" + "testing" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/expand" + "github.com/rossoctl/context-guru/schema" + "github.com/rossoctl/context-guru/store" +) + +// THE PROPERTY THAT MAKES THE SHORT PAYLOAD HORIZON SAFE (#190), driven end to end rather than +// asserted in a comment. +// +// store.DefaultStashTTL gives rewind payloads a fifth of ttl_seconds so a busy period cannot hold +// the reserve saturated for ~2.8h afterwards. That is only safe because a payload, unlike a frozen +// decision, is RE-DERIVABLE: every offloader replays its decision on every turn regardless of the +// cache-tail gate, and that replay re-stashes the payload from the message text the agent just +// re-sent. This test lets a payload be reclaimed and then checks the next turn puts it back — +// on the REQUEST path, which is what matters, because an expand call can only arrive in the +// RESPONSE to that same request. +// +// If this test ever fails, the shorter horizon is no longer safe and store.DefaultStashTTL must go +// back to DefaultTTL — not the other way round. +func TestAReclaimedPayloadIsReDerivedBeforeItsMarkerGoesUpstream(t *testing.T) { + now := time.Unix(0, 0) + // A payload horizon a fixture can cross, and a decision horizon it cannot: the decision must + // survive, or this measures a lost freeze rather than a reclaimed payload. + st := store.NewMemory(store.Options{TTLSeconds: 10000, StashTTLSeconds: 100}) + st.SetClock(func() time.Time { return now }) + m := maskFor(t) + body := strings.Repeat("verbose tool output line\n", 30) + + turn := func(maxCached int) string { + t.Helper() + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{ + tool(body), tool("new tail output"), // the agent re-sends the ORIGINAL every turn + }} + 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]) + } + + // Turn 1: nothing is cached yet, so a NEW mask is allowed. It stashes the payload and stamps + // the marker — the promise this test is about. + first := turn(-1) + if first == body { + t.Fatal("turn 1 did not mask the older output; the fixture never made a promise") + } + keys := expand.ParseMarkers(first) + if len(keys) != 1 { + t.Fatalf("expected exactly one marker in the replacement, got %d", len(keys)) + } + key := keys[0] + if _, ok := expand.Resolve(st, key); !ok { + t.Fatal("the marker did not resolve on the turn that stamped it") + } + + // Let the payload cross its own horizon and force the reclamation. Reading it is what + // enforces the TTL here, and it also makes the test NON-VACUOUS: PutStash's refresh branch + // does not check expiry, so an expired-but-unswept entry would be resurrected in place and + // this test would pass without ever exercising re-derivation. + now = now.Add(101 * time.Second) + if _, ok := expand.Resolve(st, key); ok { + t.Fatal("the payload outlived stash_ttl_seconds, so nothing was reclaimed and the " + + "re-derivation below is never exercised") + } + missingBefore := StashMissing() + + // Turn 2: the output now sits inside the provider's cached prefix, so only the frozen replay + // can act — and the replay is the thing that re-stashes. + second := turn(0) + if second != first { + t.Fatalf("turn 2 flipped representation (cache-destructive):\n want %q\n got %q", + first, second) + } + orig, ok := expand.Resolve(st, key) + if !ok { + t.Fatal("the marker on the wire does NOT resolve after its payload was reclaimed: the " + + "replay did not re-derive it, so store.DefaultStashTTL is trading reversibility for " + + "reserve liveness rather than getting both") + } + if orig != body { + t.Fatalf("the re-derived payload is not the original content:\n want %q\n got %q", + body, orig) + } + // The re-stash succeeded, so no dangling marker was reported — the counter and the resolve + // have to agree, or one of them is lying about the same event. + if got := StashMissing() - missingBefore; got != 0 { + t.Fatalf("stash_missing advanced by %d on a replay that DID resolve", got) + } + // And the store booked the reclamation as absorbed, which is the counter an operator reads to + // know the shorter horizon is costing nothing. + if sst := st.StashStats(); sst.Revived != 1 || sst.Expired != 1 { + t.Fatalf("stash_expired=%d stash_revived=%d, want 1 and 1: the pair is what says a "+ + "reclamation was absorbed rather than broken", sst.Expired, sst.Revived) + } +} + +// The same path when the reserve is FULL at replay time, which is the case the re-derivation +// cannot rescue — and the one that must still be reported rather than silent. It is why +// stash_missing's remedy is the reserve first and stash_ttl_seconds second: a reclaimed payload +// only dangles if the write that would have restored it was also refused. +func TestAReclaimedPayloadThatCannotBeReStashedIsReportedMissing(t *testing.T) { + now := time.Unix(0, 0) + // A one-slot reserve (max/2), so another session's payload can hold it against the replay. + st := store.NewMemory(store.Options{MaxEntries: 2, TTLSeconds: 10000, StashTTLSeconds: 100}) + st.SetClock(func() time.Time { return now }) + m := maskFor(t) + body := strings.Repeat("verbose tool output line\n", 30) + + turn := func(maxCached int) string { + t.Helper() + req := &bschemas.BifrostChatRequest{Input: []bschemas.ChatMessage{ + tool(body), tool("new tail output"), + }} + 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]) + } + + first := turn(-1) + keys := expand.ParseMarkers(first) + if len(keys) != 1 { + t.Fatalf("expected one marker, got %d", len(keys)) + } + now = now.Add(101 * time.Second) + if _, ok := expand.Resolve(st, keys[0]); ok { + t.Fatal("the payload outlived stash_ttl_seconds") + } + // Take the freed slot before the replay can have it. + if !st.PutStash("bbbbbbbbbbbbbbb1", []byte("another session's payload")) { + t.Fatal("could not occupy the reserve") + } + missingBefore := StashMissing() + + second := turn(0) + // The replay proceeds anyway — declining would flip an already-cached message and cannot + // un-send a marker that went out a turn ago. See commitRefresh. + if second != first { + t.Fatalf("the replay was declined and the message flipped:\n want %q\n got %q", first, second) + } + if got := StashMissing() - missingBefore; got != 1 { + t.Fatalf("stash_missing advanced by %d, want 1: a dangling marker went upstream and the "+ + "counter an operator alerts on did not move", got) + } +} diff --git a/dash/redact.go b/dash/redact.go index 95ddca91..6f438ddd 100644 --- a/dash/redact.go +++ b/dash/redact.go @@ -79,8 +79,8 @@ var configAllowlist = map[string]bool{ "preset": true, "pipeline": true, "mode": true, "cache_mode": true, "inject_expand": true, "store": true, "components": true, "store_enabled": true, "store_ttl_seconds": true, "store_max_entries": true, - "store_stash_max_bytes": true, - "listen_addr": true, "openai_upstream": true, "anthropic_upstream": true, + "store_stash_max_bytes": true, "store_stash_ttl_seconds": true, + "listen_addr": true, "openai_upstream": true, "anthropic_upstream": true, "bob_upstream": true, "force_model": true, "cheap_model": true, "cheap_model_provider": true, "cheap_model_base": true, "dashboard": true, "db_path": true, "retention": true, "capture_content": true, @@ -90,6 +90,7 @@ var configAllowlist = map[string]bool{ "min_request_tokens": true, "llm_every_n_requests": true, "llm_max_per_request": true, "marker_mode": true, "min_items": true, "keep_first": true, "keep_last": true, "enabled": true, "ttl_seconds": true, "max_entries": true, "stash_max_bytes": true, + "stash_ttl_seconds": true, } // openKeys name subtrees whose immediate child keys are USER-CHOSEN and therefore diff --git a/docs/how-to/recover-context.md b/docs/how-to/recover-context.md index ce2f42fa..efadc4a4 100644 --- a/docs/how-to/recover-context.md +++ b/docs/how-to/recover-context.md @@ -58,7 +58,10 @@ sessions — and holds, per session: - **Rewind** — `cache_key → original bytes`, what the expand loop resolves. These live in a **reserve** that is never evicted to admit a new payload: once a marker has been sent the promise is outstanding, so a full reserve makes the pipeline **decline the next removal** - (counted as `stash_refused`) instead of quietly breaking an older one. + (counted as `stash_refused`) instead of quietly breaking an older one. Payloads carry a + **shorter TTL** than everything else (`stash_ttl_seconds`, 1800 s) because each turn's replay + re-derives them from the transcript — see + [why payloads expire sooner](../reference/config.md#why-payloads-expire-sooner-than-decisions). - **Sticky** — content ids already reduced on earlier turns, so output stays byte-stable across turns. - **Frozen decisions** — the exact replacement bytes an offloader replays so an @@ -78,10 +81,13 @@ deliberately (with `marker_mode: off`) so `/compact` returns a clean, marker-fre **The model called expand and got a placeholder back.** The original expired or was evicted from the store. The provider requires one `tool_result` per `tool_call_id`, so an explicit placeholder is sent rather than nothing — which turns that offload lossy. Check `stash_missing` -and `stash_expired` at [`/stats`](../reference/routes.md#get-stats): both mean a payload left the -store, so raise `store.ttl_seconds`. `stash_refused` is the *other* case and needs -`store.max_entries` or `store.stash_max_bytes` — there nothing became irreversible, because the -removals were declined. +at [`/stats`](../reference/routes.md#get-stats): that is the one that means a marker went out with +nothing behind it. `stash_expired` on its own does **not** — a reclaimed payload is normally +re-derived by the next turn's replay, counted as `stash_revived` — so the remedy is the reserve +(`store.max_entries` / `store.stash_max_bytes`), which is what refused the re-stash, and +`store.stash_ttl_seconds` only if `stash_expired` is running far ahead of `stash_revived`. +`stash_refused` is the *other* case and needs `store.max_entries` or `store.stash_max_bytes` — +there nothing became irreversible, because the removals were declined. The turn still **completes**: the placeholder continuation is sent even when *nothing* resolved, so the model reads "no longer available" and finishes with text. It used to replay diff --git a/docs/reference/config.md b/docs/reference/config.md index 682ea692..7d2a936e 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -25,6 +25,7 @@ The document has six top-level fields (from the `Config` struct in | `enabled` | `true` | Toggles the state store. `false` wires a `store.Nop`: nothing is stashed, so offloads become **one-way** and must run `marker_mode: off`. | | `ttl_seconds` | `10000` | Entry lifetime, and it **slides** — a `Get` refreshes the deadline, so an entry replayed every turn never ages out. Raised from 1800 because Terminal-Bench tasks average ~1975 s of wall clock and run to 4 h, so the old default expired live frozen decisions mid-task. | | `max_entries` | `5000` | LRU cap. Two groups of keys are **exempt** from LRU eviction, and they behave differently when full — see below. Eviction reclaims **expired** entries first, exempt ones included. Raised from 1,000: one process-wide store serves every concurrent session, and a single reversible removal writes five entries (the payload, `cg:own:`, `cg:xseen:`, and the two pinned decision records), so 1,000 was an order of magnitude under the observed volume. | +| `stash_ttl_seconds` | `1800` | The **rewind payloads'** own entry lifetime, shorter than `ttl_seconds` and capped by it (the cap is silent, so `/config` reports the **effective** value, not the configured one). A payload is re-derivable from the transcript, a frozen decision is not — see [the two horizons](#why-payloads-expire-sooner-than-decisions) below. | | `stash_max_bytes` | `268435456` (256 MiB) | What the **rewind reserve** may cost in memory. Entries are a poor proxy for it in this one namespace: every other exempt entry is a marker line or an integer, a rewind payload is a whole tool output. Whichever of `max_entries` and this binds first, binds. | | `max_sessions` | `100` | Cap on per-session sticky-id sets. | @@ -40,7 +41,8 @@ ordinary evictable entry. prefix — a payload key *is* the marker id, a bare content hash the model reads out of the request — so it is claimed explicitly and, unlike a pin, a payload that cannot be admitted is **refused**: the component declines the removal and leaves the content verbatim rather than -stamping a marker nothing can resolve. Only the TTL releases a slot. +stamping a marker nothing can resolve. Only the TTL releases a slot — which is why payloads get a +shorter one than everything else. Each exemption is capped at half `max_entries`, and **a quarter of `max_entries` is held back from both** so something is always evictable. Without that floor the two could occupy the whole @@ -53,7 +55,55 @@ no-ops. against `stash_capacity` and `stash_bytes` against `stash_max_bytes` at [`/stats`](routes.md#get-stats) to see which budget bound. Nothing became irreversible: the removals did not happen. `stash_missing` is the different, worse number — a marker replayed with -no payload behind it — and its fix is `ttl_seconds`. +no payload behind it. A replay re-stashes the payload it re-derived, so `stash_missing` only fires +when that write was **also** refused: raise the reserve first, and `stash_ttl_seconds` if +`stash_expired` is what is taking the payloads. + +#### Why payloads expire sooner than decisions + +`ttl_seconds` is sized for a **frozen decision** — the replacement bytes the provider already +cached. Nothing else holds them, so losing one flips an already-cached message and re-writes the +whole suffix at ~11.5x the read price; it has to survive a long-horizon task, idle gaps included. + +A **payload** is a copy of content the agent re-sends every turn, so it needs a much shorter +horizon. Every offloader replays its frozen decision on **every turn**, regardless of the +cache-tail gate (it must, or the message reverts full→compacted→full and churns the KV cache), and +that replay re-stashes the payload from the message text it just read. So a live marker's payload +has its deadline slid every turn, and one the TTL already took is **re-created on the request +path** — before the request goes upstream, and therefore before any `expand` call in the response +could ask for it. `stash_revived` counts exactly that. + +Giving both namespaces `ttl_seconds` meant one busy period could hold the reserve saturated for +~2.8 h after the load that filled it, refusing every removal process-wide the whole time. The +split shrinks that hangover to `stash_ttl_seconds` without making any removal irreversible. + +**Which offloaders re-derive, because it is not all of them:** + +| Offloader | Per-turn re-stash | If its payload is reclaimed | +|---|---|---| +| `mask`, `cmdfilter`, `collapse`, `failed_run`, `skeleton`, `readlifecycle`, `agentdiet` | `reapplyFrozen` → `commitRefresh`, every turn regardless of the tail gate | re-created on the request path; `stash_revived` | +| `summarize`, `extract_llm` | only past their own gates — `summarize`'s trigger and model-availability checks, `extract_llm`'s `no_goal_keywords` | a skipped turn refreshes nothing, but it splices nothing either, so no marker of theirs dangles while the skip lasts | +| `dedup`, `extract`, `linecap`, `smartcrush` | **none** — no replay path at all; they redo the transformation from the re-sent original through the *refusable* `commitMark` | once reclaimed it is a new stash, so a saturated reserve **refuses** and the message goes upstream verbatim after earlier turns sent it compacted: `stash_refused` **plus a representation flip** | + +That last row is worth reading twice, because `stash_refused`'s own description promises "nothing +became irreversible" — true about reversibility, and silent about the cache-write actually paid. It +is reachable at `ttl_seconds` too, so it is not new; a shorter payload horizon shortens the distance +to it. + +`summarize`'s trigger skip is **recurring**, not a one-off: the agent's own compaction shrinks the +incoming request and can drop it back under the trigger's `min_request_tokens` for several +consecutive turns. + +The exposure this leaves, stated plainly: a turn that runs **no pipeline** performs no refresh (an +`x-context-guru-bypass` request, or the agent-compaction bypass), so an unbroken run of bypassed +turns longer than `stash_ttl_seconds` could outlive a payload whose marker is still live. Both are +single-request events in practice, and on the first row above the outcome is the reported one — +`stash_missing` — not a silent loss. + +**`stash_expired` and `stash_revived` both at zero means the reserve never bound**, not that the +horizon is working: the sweep runs only once a budget is already binding, and an expired-but-unswept +payload is resurrected in place by the next write. What tells the two apart is `stash_refused` and +`stash_live` against `stash_capacity`. The pinned prefixes are a code-level property of the key layout, supplied by their owners via `store.Options.PinPrefixes` — not a YAML knob. diff --git a/docs/reference/routes.md b/docs/reference/routes.md index a5abd516..6f2a0097 100644 --- a/docs/reference/routes.md +++ b/docs/reference/routes.md @@ -208,8 +208,9 @@ reversibility it destroyed, silently. | `stash_live` / `stash_capacity` | Payloads held now, and the reserve's entry cap (`max_entries / 2`). `live` approaching `capacity` is the warning. | | `stash_bytes` / `stash_max_bytes` | What those payloads cost, and the byte budget (`stash_max_bytes`). Entries are a poor proxy for memory here — a payload is a whole tool output, every other exempt entry is a marker line — so read both pairs to see **which** budget bound. | | `stash_refused` | Removals **declined** because the reserve was full. The content was left verbatim and nothing became irreversible — raise `max_entries` or `stash_max_bytes`. | -| `stash_missing` | Marker replays that found **no payload** behind them: a dangling `<>` went upstream. This one *is* a broken promise — raise `ttl_seconds`. | -| `stash_expired` | Payloads reclaimed by the TTL. With `stash_missing`, the remaining way an outstanding marker stops resolving — raise `ttl_seconds`. | +| `stash_missing` | Marker replays that found **no payload** behind them: a dangling `<>` went upstream. This one *is* a broken promise. A replay re-stashes the payload it re-derived, so this fires only when that write was **also** refused — raise the reserve, and `stash_ttl_seconds` if `stash_expired` is what is taking them. | +| `stash_expired` | Payloads reclaimed by their own TTL (`stash_ttl_seconds`, shorter than `ttl_seconds`). **Not an alert on its own** — read it against `stash_revived`. | +| `stash_revived` | Reclaimed payloads written again by a later replay, which re-derives them from the transcript **before** the marker goes upstream: reclamation absorbed at no cost. Tracking `stash_expired` means the shorter payload TTL is working as designed; see [why payloads expire sooner](config.md#why-payloads-expire-sooner-than-decisions). | `stash_refused` is the **leading** indicator for `expand_unresolved_missing`: that counter cannot move until the agent happens to call `expand`, so a proxy that had stopped being able to promise diff --git a/metrics/metrics.go b/metrics/metrics.go index 7f071a5e..55626e4a 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -769,14 +769,33 @@ type Snapshot struct { // distinct broken markers — a missing payload cannot be restored, so every later turn // re-reports it for every affected message. // - // StashExpired counts payloads the TTL reclaimed, which is the one remaining way an - // outstanding marker stops resolving; the fix for that is a larger ttl_seconds. Live - // against Capacity, and Bytes against MaxBytes, say how close each budget is to binding - // before any of them fires. + // StashExpired counts payloads the TTL reclaimed. Payloads have their OWN, shorter TTL + // (stash_ttl_seconds) because a payload — unlike a frozen decision — is re-derivable from + // the transcript: every turn's replay re-writes it, so a reclaimed one is re-created on the + // request path before any expand could ask for it (see store.DefaultStashTTL, #190). + // + // StashRevived is that absorption, counted: a payload written again under a key the TTL had + // taken. It is why StashExpired is not itself an alert — Expired without Revived is a + // session that never came back, which is the reclamation working. What breaks the promise is + // StashMissing, and the remedy for a rising StashMissing is stash_ttl_seconds (the payload + // was reclaimed too eagerly) or a larger reserve (the re-stash was refused) — read Live + // against Capacity and Bytes against MaxBytes to tell which. + // + // BOTH AT ZERO IS NOT EVIDENCE THAT THE HORIZON WORKS. It means the reserve never bound. + // sweepExpired runs only from StashRoom, PutStash's pre-refusal path and evictOldest — i.e. + // only once the reserve, the shared exempt budget or the entry cap is already binding — and + // PutStash's refresh branch does not check expiry, so on an unsaturated run an + // expired-but-unswept payload is resurrected in place and NEITHER counter moves. That is the + // intended behaviour (a slot is released when a slot is wanted), but it means a run must + // actually saturate the reserve before this pair says anything, which is the same precondition + // iteration 024 failed to meet for stash_refused — and failing it is how #190 became + // undecidable from data. What distinguishes "never bound" from "working" is StashRefused and + // StashLive against StashCapacity. // Filled by the host at serve time (offload + store live below metrics). StashRefused int64 `json:"stash_refused"` StashMissing int64 `json:"stash_missing"` StashExpired int64 `json:"stash_expired"` + StashRevived int64 `json:"stash_revived"` StashLive int `json:"stash_live"` StashCapacity int `json:"stash_capacity"` StashBytes int64 `json:"stash_bytes"` diff --git a/proxy/promexport.go b/proxy/promexport.go index fd3fdcec..c52c903f 100644 --- a/proxy/promexport.go +++ b/proxy/promexport.go @@ -516,9 +516,16 @@ func (h *Handler) renderMetrics() string { "What the held rewind payloads cost, and the reserve's byte budget (stash_max_bytes). Read against cg_stash_reserve_entries: entries near capacity means raise max_entries, bytes near the budget means raise stash_max_bytes.", "gauge") promLine(&b, "cg_stash_reserve_bytes", `state="live"`, float64(st.Bytes)) promLine(&b, "cg_stash_reserve_bytes", `state="capacity"`, float64(st.MaxBytes)) + // Payloads carry their OWN, shorter TTL (stash_ttl_seconds), because a payload is + // re-derivable from the transcript and a frozen decision is not — so reclaiming one is + // ordinarily absorbed by the next turn's replay rather than being a loss. That makes + // this counter on its own ambiguous, which is why the revived series ships with it. promHeaderProc(&b, "cg_stash_expired_total", - "Rewind payloads reclaimed by the TTL. The one remaining way an outstanding marker stops resolving; raise ttl_seconds.", "counter") + "Rewind payloads reclaimed by their TTL (stash_ttl_seconds). NOT an alert on its own: read against cg_stash_revived_total, and alert on cg_stash_missing_total instead.", "counter") promLine(&b, "cg_stash_expired_total", "", float64(st.Expired)) + promHeaderProc(&b, "cg_stash_revived_total", + "Reclaimed payloads written again by a later replay, which re-derives them from the transcript before the marker goes upstream: reclamation absorbed at no cost. Tracking cg_stash_expired_total means the shorter payload TTL is working; lagging it while cg_stash_missing_total rises means raise stash_ttl_seconds.", "counter") + promLine(&b, "cg_stash_revived_total", "", float64(st.Revived)) } // Process-wide, so outside the cast for the same reason hit/miss are: a component // declines the removal, whichever store instance refused the payload. This is the @@ -532,7 +539,7 @@ func (h *Handler) renderMetrics() string { // that one. It grows with turn count rather than with distinct dangling markers, // because a payload that has gone cannot be restored and every later turn replays it. promHeaderProc(&b, "cg_stash_missing_total", - "Marker replays that found NO payload behind them: a dangling <> went upstream, so this is a broken reversibility promise rather than a declined removal. Raise ttl_seconds. Grows per turn per affected message, not per distinct marker.", "counter") + "Marker replays that found NO payload behind them: a dangling <> went upstream, so this is a broken reversibility promise rather than a declined removal. A replay re-stashes the payload it re-derived, so this only fires when that write was ALSO refused — raise max_entries/stash_max_bytes, and stash_ttl_seconds if cg_stash_expired_total is what is taking them. Grows per turn per affected message, not per distinct marker.", "counter") promLine(&b, "cg_stash_missing_total", "", float64(offload.StashMissing())) // Same rule as the two families above, and this counter would have had the same bug: diff --git a/proxy/promexport_coverage_test.go b/proxy/promexport_coverage_test.go index cf712c9a..c52c6d0b 100644 --- a/proxy/promexport_coverage_test.go +++ b/proxy/promexport_coverage_test.go @@ -96,6 +96,7 @@ var notExportedWhy = map[string]string{ "StashRefused": "cg_stash_refused_total, from offload.StashRefusals()", "StashMissing": "cg_stash_missing_total, from offload.StashMissing()", "StashExpired": "cg_stash_expired_total, from store.Memory.StashStats()", + "StashRevived": "cg_stash_revived_total, from store.Memory.StashStats()", "StashLive": `cg_stash_reserve_entries{state="live"}, from store.Memory.StashStats()`, "StashCapacity": `cg_stash_reserve_entries{state="capacity"}, from store.Memory.StashStats()`, "StashBytes": `cg_stash_reserve_bytes{state="live"}, from store.Memory.StashStats()`, diff --git a/proxy/proxy.go b/proxy/proxy.go index 354b4dd4..b3b455ea 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -1915,6 +1915,10 @@ func (h *Handler) stats(w http.ResponseWriter, r *http.Request) { snap.FrozenFlips = snap.FrozenDropped - snap.FrozenRepaired st := fl.StashStats() snap.StashLive, snap.StashCapacity, snap.StashExpired = st.Live, st.Capacity, st.Expired + // Published alongside Expired, never instead of it: the pair is what says whether the + // shorter payload TTL is being absorbed by the per-turn re-stash or is running ahead of + // it. Expired alone reported both outcomes at once. + snap.StashRevived = st.Revived snap.StashBytes, snap.StashMaxBytes = st.Bytes, st.MaxBytes } // Cached-prefix restarts after an agent compaction. Same layering as the pool counters diff --git a/proxy/stats_golden_test.go b/proxy/stats_golden_test.go index 455d07a0..e0f9065a 100644 --- a/proxy/stats_golden_test.go +++ b/proxy/stats_golden_test.go @@ -101,6 +101,7 @@ var statsGoldenTopLevel = []string{ "stash_bytes", "stash_capacity", "stash_expired", + "stash_revived", "stash_live", "stash_max_bytes", "stash_missing", diff --git a/store/payload_ttl_test.go b/store/payload_ttl_test.go new file mode 100644 index 00000000..9f9d9b6d --- /dev/null +++ b/store/payload_ttl_test.go @@ -0,0 +1,216 @@ +package store + +import ( + "strconv" + "testing" + "time" +) + +// The payload horizon (#190). +// +// #188 bounded the reserve but left a slot released ONLY by the TTL, and both namespaces shared +// one. The store is a single process-wide instance, so a busy period could hold the reserve +// saturated for the whole of ttl_seconds — ~2.8h at the default — refusing every removal for +// every session the entire time. The fix is a shorter horizon for the payloads alone, and what +// makes it safe is that a payload is re-derivable from the transcript while a frozen decision is +// not: see DefaultStashTTL, and TestAReclaimedPayloadIsReDerivedBeforeItsMarkerGoesUpstream in +// components/offload, which drives the re-derivation end to end. + +// The split itself: a payload written alongside its decisions is reclaimed at the payload +// horizon, while those decisions — which nothing else holds a copy of — are still live. +func TestPayloadsExpireSoonerThanTheDecisionsThatNameThem(t *testing.T) { + now := time.Unix(0, 0) + // Explicit, and deliberately far apart: ttl_seconds is sized for a long-horizon task's idle + // gaps, stash_ttl_seconds for one inter-turn gap. 20 entries so the reserve (max/2 = 10) is + // what binds below rather than the shared exempt budget — this test is about the horizon, and + // writeRemoval's five-entries-per-removal arithmetic is covered in stash_test.go. + m := NewMemory(Options{MaxEntries: 20, TTLSeconds: 10000, StashTTLSeconds: 100}) + m.SetClock(func() time.Time { return now }) + + // The two PINNED decision records one reversible removal writes, whose loss is the + // cache-destructive event ttl_seconds is long for. + m.Put(ResultPrefix+"sess:1", []byte("masked replacement bytes")) + m.Put(XResultPrefix+"1", []byte("masked replacement bytes")) + + k1 := "aaaaaaaaaaaaaaa0" + for i := 0; i < m.stashCap(); i++ { // saturate the reserve, as a busy period does + if !m.PutStash("aaaaaaaaaaaaaaa"+strconv.Itoa(i), []byte("payload")) { + t.Fatalf("filling an empty reserve was refused at %d", i) + } + } + if m.PutStash("bbbbbbbbbbbbbbb1", []byte("one more")) { + t.Fatal("a payload was accepted into a saturated reserve") + } + + now = now.Add(101 * time.Second) // past the PAYLOAD horizon, nowhere near ttl_seconds + + // The slot frees, so the pipeline can promise reversibility again. This is the liveness #190 + // is about: under one shared TTL it would not arrive for another 9,899s. + if !m.PutStash("bbbbbbbbbbbbbbb1", []byte("one more")) { + t.Fatalf("the reserve released no slot at the payload horizon (%v): one busy period "+ + "still saturates it for the whole of ttl_seconds", m.stashTTL) + } + if _, live := m.Get(k1); live { + t.Fatal("a payload past stash_ttl_seconds is still being served") + } + // And the half that must NOT have expired with it. A frozen decision is the replacement + // bytes the provider already cached; losing one flips an already-cached message and re-writes + // the suffix at ~11.5x the read price, which is the cost ttl_seconds is long for. + for _, k := range []string{ResultPrefix + "sess:1", XResultPrefix + "1"} { + if _, ok := m.Get(k); !ok { + t.Fatalf("%s expired with the payload: the two horizons are not actually split, so "+ + "shortening the payload TTL now costs a cache-write per message", k) + } + } +} + +// stash_expired reports two outcomes at once — a payload nobody wanted, and a payload an +// outstanding marker still needs — and those call for opposite responses. stash_revived is the +// benign half, counted at the one place that knows the entry was absent. Without it an operator +// watching a rising stash_expired cannot tell the shorter horizon working from it running ahead +// of the re-stash. Same argument as stash_refused vs stash_missing in #188. +func TestAReclaimedPayloadRewrittenByAReplayIsCountedAsRevived(t *testing.T) { + now := time.Unix(0, 0) + m := NewMemory(Options{MaxEntries: 8, TTLSeconds: 10000, StashTTLSeconds: 100}) + m.SetClock(func() time.Time { return now }) + key := "aaaaaaaaaaaaaaa1" + if !m.PutStash(key, []byte("original tool output")) { + t.Fatal("the first stash was refused") + } + if st := m.StashStats(); st.Revived != 0 { + t.Fatalf("a FIRST stash was counted as a revival (%d): the counter would report "+ + "reclamation being absorbed on a store that has never reclaimed anything", st.Revived) + } + + now = now.Add(101 * time.Second) + if _, live := m.Get(key); live { + t.Fatal("the payload outlived stash_ttl_seconds") + } + + // What a replay does: re-derive the payload from the message text the agent re-sent and write + // it back under the same key (the key IS the content hash, so the marker's bytes are + // unchanged and nothing flips). + if !m.PutStash(key, []byte("original tool output")) { + t.Fatal("re-stashing a reclaimed payload was refused") + } + st := m.StashStats() + if st.Expired == 0 { + t.Fatal("StashStats() reports 0 expired after a payload was reclaimed") + } + if st.Revived != 1 { + t.Fatalf("stash_revived = %d, want 1: the reclamation was absorbed at no cost and "+ + "nothing says so, so stash_expired reads as a possible broken promise", st.Revived) + } + if _, ok := m.Get(key); !ok { + t.Fatal("the revived payload does not resolve, so the marker really is dangling") + } + // Counted per RECLAMATION, not per write: a live payload refreshed every turn must not + // inflate the figure, or "absorbed" stops being readable as a rate against expired. + if !m.PutStash(key, []byte("original tool output")) { + t.Fatal("refreshing a live payload was refused") + } + if st := m.StashStats(); st.Revived != 1 { + t.Fatalf("stash_revived = %d after refreshing a LIVE payload, want 1", st.Revived) + } +} + +// A payload outliving the decision that names its marker is memory nothing can ever read: once +// the decision is gone, no replay stamps that marker again. So an operator who shortens +// ttl_seconds below the payload default must get the shorter of the two, not a reserve held open +// by dead payloads — which is the exact saturation #190 is about, arrived at by config. +func TestThePayloadHorizonNeverOutlivesTheDecisionHorizon(t *testing.T) { + m := NewMemory(Options{TTLSeconds: 60}) // shorter than DefaultStashTTL + if m.stashTTL > m.ttl { + t.Fatalf("stashTTL %v exceeds ttl %v: dead payloads would hold reserve slots open for "+ + "%v after the last decision naming them expired", m.stashTTL, m.ttl, m.stashTTL-m.ttl) + } + if m.stashTTL != 60*time.Second { + t.Fatalf("stashTTL %v, want the capped 60s", m.stashTTL) + } + // Still configurable in its own right, and still defaulted. + if m2 := NewMemory(Options{StashTTLSeconds: 42}); m2.stashTTL != 42*time.Second { + t.Fatalf("stash_ttl_seconds must stay configurable, got %v", m2.stashTTL) + } + if m3 := NewMemory(Options{}); m3.stashTTL != DefaultStashTTL { + t.Fatalf("zero StashTTLSeconds should yield DefaultStashTTL, got %v", m3.stashTTL) + } +} + +// The config surface must report what the store WILL USE, not what was configured. +// +// The cap is applied silently inside NewMemory, so `/config` publishing the raw field showed +// stash_ttl_seconds: 20000 on the dashboard while the store used 10000 — an operator told one thing +// while another runs, which is the same silent divergence #200 is about, in the config surface +// instead of the metrics one. EffectiveStashTTLSeconds derives it from the same code path NewMemory +// uses, so the two cannot drift apart. +func TestTheConfigSurfaceReportsTheEffectivePayloadHorizon(t *testing.T) { + // The case that diverged: a payload horizon longer than the decision horizon. + if got := EffectiveStashTTLSeconds(Options{TTLSeconds: 10000, StashTTLSeconds: 20000}); got != 10000 { + t.Errorf("reported %d, want the capped 10000: /config would advertise a horizon the store "+ + "does not use", got) + } + // And the cases that did not, which must keep working. + if got := EffectiveStashTTLSeconds(Options{TTLSeconds: 10000, StashTTLSeconds: 600}); got != 600 { + t.Errorf("reported %d, want the configured 600", got) + } + if got := EffectiveStashTTLSeconds(Options{}); got != int(DefaultStashTTL/time.Second) { + t.Errorf("reported %d, want DefaultStashTTL in seconds", got) + } + // NO CROSS-CHECK AGAINST NewMemory HERE, deliberately. EffectiveStashTTLSeconds IS + // `NewMemory(o).stashTTL`, so comparing the two is A == A: it cannot fail, and a reader + // scanning for coverage would count it as evidence. The three assertions above are the whole + // test on this side. + // + // The drift this helper exists to prevent is between the value /config PUBLISHES and the value + // the store uses, and nothing in this package can span that — main.go could go back to + // publishing the raw field with every test here still green. That assertion lives where it can + // fail: cmd/context-guru-proxy.TestConfigPublishesTheEffectivePayloadHorizon. +} + +// The constants' RELATIONSHIP is the design, so it is asserted rather than left to whoever next +// edits one of them: payloads short because a replay re-derives them, decisions long because +// nothing else holds the bytes the provider cached. +func TestDefaultPayloadHorizonIsWellInsideTheDecisionHorizon(t *testing.T) { + if DefaultStashTTL >= DefaultTTL { + t.Fatalf("DefaultStashTTL %v is not shorter than DefaultTTL %v: the reserve is still "+ + "held for a whole task horizon by one busy period (#190)", DefaultStashTTL, DefaultTTL) + } + // And not so short that an ordinary inter-turn gap — a test suite, a build, a long tool call + // — outlives a payload whose marker is still live in the transcript. + if DefaultStashTTL < 15*time.Minute { + t.Fatalf("DefaultStashTTL %v is shorter than a long tool call, so a live marker's "+ + "payload can be reclaimed between two consecutive turns", DefaultStashTTL) + } +} + +// An entry that BECOMES a payload has its deadline moved EARLIER (ttl -> stashTTL), and +// nextExpiry — the lower bound that lets sweepExpired skip a pass — has to follow it down. A +// bound left above an entry's real expiry makes the sweep return early, and for a payload that +// means a reserve slot no TTL ever reclaims: the exact permanent saturation #190 is about, +// reintroduced by the fix for it. +// +// Sized so the become-a-payload write is the ONLY thing that can lower the bound: a one-slot +// reserve, so no second PutStash gets to seed it on the way in. +func TestBecomingAPayloadLowersTheSweepBound(t *testing.T) { + now := time.Unix(0, 0) + m := NewMemory(Options{MaxEntries: 2, TTLSeconds: 10000, StashTTLSeconds: 100}) // reserve = 1 + m.SetClock(func() time.Time { return now }) + + key := "aaaaaaaaaaaaaaa1" + m.Put(key, []byte("written plain first, so nextExpiry is seeded at now+ttl")) + if !m.PutStash(key, []byte("and now claimed as a payload")) { + t.Fatal("claiming a present entry as a payload was refused") + } + if m.PutStash("bbbbbbbbbbbbbbb1", []byte("second")) { + t.Fatal("a second payload was accepted into a one-slot reserve") + } + + now = now.Add(101 * time.Second) // past the PAYLOAD horizon, far short of ttl_seconds + if !m.PutStash("bbbbbbbbbbbbbbb1", []byte("second")) { + t.Fatal("the sweep skipped an expired payload because nextExpiry still held the long " + + "horizon, so this reserve slot is never reclaimed by any TTL") + } + if _, live := m.Get(key); live { + t.Fatal("the expired payload is still being served") + } +} diff --git a/store/store.go b/store/store.go index cd6bc1f7..1eb7082e 100644 --- a/store/store.go +++ b/store/store.go @@ -174,8 +174,12 @@ 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 + mu sync.Mutex + // ttl is every other namespace's lifetime; stashTTL is the rewind payloads', and it is + // shorter because a payload is re-derivable from the transcript and a frozen decision is + // not. See DefaultStashTTL. ttl time.Duration + stashTTL time.Duration max int ll *list.List // LRU, front = most recent items map[string]*list.Element // key -> element(*entry) @@ -207,6 +211,18 @@ type Memory struct { // pressure cannot evict one. stashRefusedN int64 stashExpiredN int64 + // reclaimed remembers keys whose PAYLOAD the TTL took, so a later PutStash under the same + // key can be recognised as a REVIVAL rather than a first stash. That distinction is the + // evidence for the whole shorter-payload-TTL trade (see DefaultStashTTL): a reclamation the + // next replay re-creates cost nothing, and one it does not shows up as stash_missing. + // Without it stash_expired reports both outcomes at once and an operator cannot tell which + // they have — the ambiguous-counter shape #188 split stash_refused/stash_missing over. + // + // Bounded and FIFO exactly like lostFrozen, for the same reason: it is a diagnostic set, so + // dropping the oldest mark costs at most one uncounted revival. + reclaimed map[string]struct{} + reclaimedOrder []string + stashRevivedN int64 // 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. @@ -234,7 +250,14 @@ type Options struct { // StashMaxBytes caps the rewind reserve in BYTES. Zero => DefaultStashMaxBytes. See // Memory.stashBytes for why this namespace is budgeted in bytes and the rest in entries. StashMaxBytes int64 `yaml:"stash_max_bytes"` - MaxSessions int `yaml:"max_sessions"` + // StashTTLSeconds is the REWIND PAYLOADS' own entry lifetime, shorter than TTLSeconds. + // Zero => DefaultStashTTL. Capped at TTLSeconds, since a payload outliving the decision that + // names it is a reserve slot held for almost nothing. Report it through + // EffectiveStashTTLSeconds, never raw: the cap is applied silently, and publishing the + // configured value while the store uses another is the same silent divergence #200 is about. + // See DefaultStashTTL for why the two namespaces do not want the same horizon. + StashTTLSeconds int `yaml:"stash_ttl_seconds"` + MaxSessions int `yaml:"max_sessions"` } // Nop is a Store that persists nothing: Put discards, Get/Sticky always miss. @@ -264,6 +287,66 @@ func (Nop) Persists() bool { return false } // (test suites, training runs) with the sliding refresh doing the rest. const DefaultTTL = 10000 * time.Second +// DefaultStashTTL is how long a REWIND PAYLOAD lives: 1800s, a fifth of DefaultTTL. +// +// The two namespaces have genuinely different horizons, and giving them one TTL is what made a +// saturated reserve hold for ~2.8h after the busy period that filled it (#190). A slot was +// released only by the TTL, the store is one process-wide instance, and while the reserve is +// saturated every new removal is refused for every session at once — so savings fell to zero +// long after the load that caused it was gone. +// +// WHAT MAKES THE SHORT HORIZON SAFE is that a payload, unlike a frozen decision, is +// RE-DERIVABLE from the request in flight: +// +// - A frozen decision is the replacement bytes the provider ALREADY CACHED. Nothing else +// holds them; losing one flips an already-cached message and re-writes the suffix at ~11.5x +// the read price. It needs to survive a whole long-horizon task, idle gaps included, which +// is what DefaultTTL is sized for. +// - A payload is a copy of content the AGENT RE-SENDS every turn. The replay path calls +// components/offload.commitRefresh, which PutStashes the payload again from the message text +// it just read. So a live marker's payload has its expiry slid every turn, and one already +// reclaimed is RE-CREATED on the REQUEST path, before the request goes upstream and +// therefore before any expand call in the response could ask for it. +// +// The horizon a payload actually needs is one INTER-TURN GAP, not one session. 1800s is the +// value DefaultTTL's own comment records as too short for a frozen decision (headroom's CCR +// store) — reused here, in the one namespace whose horizon it does fit, rather than invented. +// +// WHICH OFFLOADERS ACTUALLY DO THAT, because an earlier version of this comment said "every +// offloader on every turn" and that quantifier is false in two ways with different consequences: +// +// - The reapplyFrozen family (mask, cmdfilter, collapse, failed_run, skeleton, readlifecycle, +// agentdiet) does behave as described: the replay runs on every turn regardless of the +// cache-tail gate, because it must, or the message reverts full→compacted→full. +// - summarize and extract_llm have GATES AHEAD of their replay phase — summarize returns at +// summarize.go:157 on its trigger and at :162/:165 when no model client resolves, both before +// tryReuse; extract_llm returns at extract_llm.go:659 on no_goal_keywords, before Phase 1. +// A skipped turn refreshes none of their payloads. summarize's trigger skip is RECURRING +// rather than a one-off, because the agent's own compaction shrinks the incoming request and +// can drop it back under Trigger.MinRequestTokens for several consecutive turns; "the cheap +// model is down" likewise persists for many turns by nature. A skipped component splices +// nothing, so no marker of its goes upstream on those turns and no marker dangles — the +// payload's reclamation is harmless while the skip lasts, and the exposure is only that its +// next firing may find the payload gone and the reserve full at the same moment. +// - dedup, extract, linecap and smartcrush have NO replay path at all (no reapplyFrozen and no +// commitRefresh). They redo the transformation from the re-sent original every turn, so their +// per-turn write goes through the REFUSABLE commitMark. While the payload is live that lands +// in PutStash's refresh branch and is retained unconditionally; once reclaimed it is a NEW +// stash, and a new stash into a saturated reserve is refused, the component declines, and the +// message goes upstream verbatim after earlier turns sent it compacted. So for these four the +// outcome is stash_refused PLUS a representation flip, not stash_missing — and stash_refused's +// operator-facing text promises "nothing became irreversible", which is true about +// reversibility and silent about the cache-write actually paid. Reachable at 10,000s too, so +// not introduced here; this horizon shortens the distance to it by 5.5x. +// +// The residual exposure, stated rather than waved at: a turn that runs NO pipeline performs no +// refresh (an x-context-guru-bypass request, or the agent-compaction bypass), so a long +// unbroken run of bypassed turns could outlive a payload while its marker is still live in the +// transcript. Both are single-request events in practice. On the reapplyFrozen family the outcome +// is then the already-reported one — stash_missing on the next replay — not a silent loss, and +// stash_revived is what says whether reclamation is being absorbed as designed. +const DefaultStashTTL = 1800 * time.Second + // DefaultMaxEntries is the store's default entry cap. // // It was 1,000, and that was the quantity #187 was measured against: ONE process-wide store @@ -299,6 +382,18 @@ const DefaultMaxEntries = 5000 // (stash_refused), instead of quietly making them irreversible. const DefaultStashMaxBytes = 256 << 20 +// EffectiveStashTTLSeconds is the payload horizon a store built from these Options will ACTUALLY +// use, in seconds — the default filled in and the ttl_seconds cap applied. +// +// It exists so the config surface cannot drift from the store. /config published the configured +// value, so `stash_ttl_seconds: 20000` with `ttl_seconds: 10000` displayed 20000 on the dashboard +// while the store used 10000 — a silent divergence between what an operator is told and what runs, +// which is the shape #200 is about, in the config surface instead of the metrics one. Derived from +// the same code path NewMemory uses rather than re-implemented, so the two cannot disagree. +func EffectiveStashTTLSeconds(o Options) int { + return int(NewMemory(o).stashTTL / time.Second) +} + // NewMemory builds an in-memory store. Zero/negative option fields fall back to // defaults (DefaultTTL, DefaultMaxEntries, 100 sessions of sticky sets). func NewMemory(o Options) *Memory { @@ -314,6 +409,25 @@ func NewMemory(o Options) *Memory { if stashMax <= 0 { stashMax = DefaultStashMaxBytes } + stashTTL := time.Duration(o.StashTTLSeconds) * time.Second + if o.StashTTLSeconds <= 0 { + stashTTL = DefaultStashTTL + } + // A payload outliving the frozen decision that names its marker is a reserve slot held for + // almost nothing: once the decision is gone, no replay stamps that marker again. Not "nothing + // can EVER read it" — the model can still call expand on a marker it read in an earlier turn's + // context, because the marker lives in the conversation it is reasoning over and not only in + // the request the proxy just built. That path is rare and short-lived and does not change the + // conclusion, so the cap stands; the claim is just narrower than it was. + // + // So an operator who shortens ttl_seconds below the payload default gets the shorter of the + // two rather than a reserve held open by dead payloads — which would be #190's saturation + // arrived at by configuration. Deliberately no escape hatch: a workload that wants payloads to + // outlive decisions is better served by raising ttl_seconds. EffectiveStashTTLSeconds is what + // the config surface must report, so the cap is not silent. + if stashTTL > ttl { + stashTTL = ttl + } stick := o.MaxSessions if stick <= 0 { stick = 100 @@ -323,11 +437,12 @@ func NewMemory(o Options) *Memory { pins = DefaultPinPrefixes } return &Memory{ - ttl: ttl, max: max, maxStick: stick, pinPrefixes: pins, + ttl: ttl, stashTTL: stashTTL, max: max, maxStick: stick, pinPrefixes: pins, stashMaxBytes: stashMax, ll: list.New(), items: map[string]*list.Element{}, sticky: map[string]map[string]struct{}{}, lostFrozen: map[string]struct{}{}, + reclaimed: map[string]struct{}{}, now: time.Now, } } @@ -410,6 +525,33 @@ func (m *Memory) DisableSlidingTTLForTest() { // own existing over-cap behavior: a pin degrades to an ordinary evictable entry (it is still // readable, and its loss is reported where losses happen), a stash is REFUSED (so the caller // declines the removal rather than promising what it cannot deliver). No new failure shape. +// ttlFor is the lifetime an entry gets when it is written or read. Rewind payloads take the +// shorter stashTTL, everything else the full ttl — keyed on the STASH FLAG rather than on the +// key, because a payload's key is a bare content hash the store cannot recognise (see Stasher). +func (m *Memory) ttlFor(e *entry) time.Duration { + if e.stash { + return m.stashTTL + } + return m.ttl +} + +// setExpiry stamps an entry's deadline from its own horizon and keeps nextExpiry a valid lower +// bound. EVERY write that can move a deadline goes through here. +// +// It exists because the correctness argument otherwise has to be redone per site. Three of the five +// sites can only move a deadline LATER, where noteExpiry is a no-op by construction, and two can +// move it EARLIER (a plain entry claimed as a stash, ttl -> stashTTL). Getting that wrong is not a +// visible bug: a bound left above an entry's real expiry makes sweepExpired return early, so the +// entry is never reclaimed and — for a payload — the reserve slot is held for the life of the +// process, which is the permanent version of the saturation this whole change is about. The safety +// of the later-only sites also rests on stashTTL <= ttl and a monotonic clock, so it is a +// consequence of a cap elsewhere rather than a local property. One helper, one comparison, and a +// sixth site cannot get it wrong. +func (m *Memory) setExpiry(e *entry) { + e.expires = m.now().Add(m.ttlFor(e)) + m.noteExpiry(e.expires) +} + func (m *Memory) pinCap() int { return m.max / 2 } func (m *Memory) stashCap() int { return m.max / 2 } @@ -478,7 +620,6 @@ func (m *Memory) PutStash(key string, payload []byte) bool { m.stashBytes += int64(len(payload)) - int64(len(e.payload)) } e.payload = payload - e.expires = m.now().Add(m.ttl) // Claim a reserve slot if one has freed, exactly as Put re-claims a pin slot. An // entry already present is retained whatever the reserve says: refusing a REFRESH // would make a component decline to replay a marker it has already stamped, which @@ -491,6 +632,10 @@ func (m *Memory) PutStash(key string, payload []byte) bool { m.stashN++ m.stashBytes += int64(len(payload)) } + // AFTER the slot claim above, so an entry that just became a stash gets the payload horizon + // rather than keeping the one it was written with — the one refresh that moves a deadline + // EARLIER, which is why setExpiry lowers the sweep bound unconditionally. + m.setExpiry(e) m.ll.MoveToFront(el) return true } @@ -505,10 +650,24 @@ func (m *Memory) PutStash(key string, payload []byte) bool { m.stashRefusedN++ return false } - e := &entry{key: key, payload: payload, expires: m.now().Add(m.ttl), stash: true} + e := &entry{key: key, payload: payload, stash: true} + m.setExpiry(e) + // A key the TTL took and a caller has now written again is the shorter payload horizon + // working as designed: the replay re-derived the payload from the transcript, and the marker + // it is about to send resolves. Counted here — the one place that knows the entry was absent + // — because "reclaimed" and "reclaimed and absorbed" call for opposite operator responses. + if _, wasReclaimed := m.reclaimed[key]; wasReclaimed { + delete(m.reclaimed, key) + for i, k := range m.reclaimedOrder { + if k == key { + m.reclaimedOrder = append(m.reclaimedOrder[:i], m.reclaimedOrder[i+1:]...) + break + } + } + m.stashRevivedN++ + } m.stashN++ m.stashBytes += int64(len(payload)) - m.noteExpiry(e.expires) m.items[key] = m.ll.PushFront(e) for m.ll.Len() > m.max { if !m.evictOldest() { @@ -530,6 +689,19 @@ type StashStat struct { // stash leaves now that LRU pressure cannot evict one. Refused int64 Expired int64 + // Revived counts payloads WRITTEN AGAIN under a key the TTL had reclaimed — a replay + // re-derived the payload from the transcript before its marker went upstream, so the + // reclamation cost nothing. It is the half of Expired that is benign, split out for the same + // reason stash_refused and stash_missing are: Expired alone reports both "the reserve + // released a payload nobody wanted" and "an outstanding marker just lost its payload", and + // those call for opposite responses (nothing vs. raise stash_ttl_seconds). + // + // Read it as a RATE against Expired. Revived tracking Expired means the shorter payload + // horizon is being absorbed as designed; Expired climbing while Revived stays flat means + // payloads are being reclaimed from sessions that then never came back — also fine, that is + // the reclamation this exists to do — and the outcome to alert on is neither of these but + // stash_missing, which is what a reclamation that was NOT absorbed produces. + Revived int64 } // StashStats reports the rewind reserve against both of its budgets. @@ -548,7 +720,7 @@ func (m *Memory) StashStats() StashStat { return StashStat{ Live: m.stashN, Capacity: m.stashCap(), Bytes: m.stashBytes, MaxBytes: m.stashMaxBytes, - Refused: m.stashRefusedN, Expired: m.stashExpiredN, + Refused: m.stashRefusedN, Expired: m.stashExpiredN, Revived: m.stashRevivedN, } } @@ -573,7 +745,10 @@ func (m *Memory) Put(key string, payload []byte) { if el, ok := m.items[key]; ok { e := el.Value.(*entry) e.payload = payload - e.expires = m.now().Add(m.ttl) + // setExpiry, not m.ttl: a plain Put must not hand a rewind payload the long horizon and + // undo the split. Namespaces do not collide today (a payload's key is a bare hash), so + // this is the invariant held at the write rather than a fix for an observed path. + m.setExpiry(e) // 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. @@ -584,7 +759,7 @@ func (m *Memory) Put(key string, payload []byte) { m.ll.MoveToFront(el) return } - e := &entry{key: key, payload: payload, expires: m.now().Add(m.ttl)} + e := &entry{key: key, payload: payload} // 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 @@ -595,7 +770,7 @@ func (m *Memory) Put(key string, payload []byte) { e.pinned = true m.pinnedN++ } - m.noteExpiry(e.expires) + m.setExpiry(e) m.items[key] = m.ll.PushFront(e) for m.ll.Len() > m.max { if !m.evictOldest() { @@ -625,6 +800,22 @@ func (m *Memory) noteLost(key string) { m.lostN++ } +// noteReclaimed records that the TTL took the payload under key, so a later PutStash of the +// same key is recognisable as a revival. Bounded FIFO like noteLost: oldest mark goes first, so +// a busy session cannot delete another's fresh mark and under-count its revival. +func (m *Memory) noteReclaimed(key string) { + if _, dup := m.reclaimed[key]; dup { + return + } + for len(m.reclaimed) >= m.max && len(m.reclaimedOrder) > 0 { + oldest := m.reclaimedOrder[0] + m.reclaimedOrder = m.reclaimedOrder[1:] + delete(m.reclaimed, oldest) + } + m.reclaimed[key] = struct{}{} + m.reclaimedOrder = append(m.reclaimedOrder, key) +} + // 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 { @@ -667,7 +858,7 @@ func (m *Memory) Get(key string) ([]byte, bool) { // 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.setExpiry(e) } m.ll.MoveToFront(el) return e.payload, true @@ -703,10 +894,15 @@ func (m *Memory) MarkSticky(session, id string) { s[id] = struct{}{} } -// noteExpiry keeps nextExpiry a valid LOWER BOUND on the earliest expires in the store. Every -// write sets its entry's expiry to now+ttl — the latest of any live entry — so the bound only -// ever needs seeding, never raising: the first write after a sweep supplies it, and a sweep -// recomputes it exactly. +// noteExpiry keeps nextExpiry a valid LOWER BOUND on the earliest expires in the store: it only +// ever lowers the bound, so a sweep is skipped only when nothing can possibly have expired. +// +// Being too LOW is safe — it costs one wasted sweep. Being too HIGH is not: sweepExpired returns +// early and an expired entry is never reclaimed, which for a payload means a reserve slot held +// forever. That is why every write that can move a deadline earlier must come through here. Two +// horizons make this less obvious than it was: a plain write's now+ttl is no longer "the latest +// of any live entry", because a payload written a moment later gets now+stashTTL, which is +// sooner. A sweep recomputes the bound exactly from what survived. func (m *Memory) noteExpiry(t time.Time) { if m.nextExpiry.IsZero() || t.Before(m.nextExpiry) { m.nextExpiry = t @@ -775,6 +971,7 @@ func (m *Memory) remove(el *list.Element) { if e.stash { m.stashN-- m.stashBytes -= int64(len(e.payload)) + m.noteReclaimed(e.key) // A stash only reaches here via sweepExpired (LRU pressure cannot take one), so // this counts TTL reclamation. Counted rather than silent because it is the one // remaining way an outstanding marker can stop resolving, and an operator seeing