diff --git a/components/component.go b/components/component.go index 519603c8..1d12e658 100644 --- a/components/component.go +++ b/components/component.go @@ -530,6 +530,21 @@ type Report struct { // extract_llm fans out into a pre-sized slice and assigns the result once, which is the // same shape its projected-output collection already uses. Calls []ModelCall + // Replays counts changes this component made by REPLAYING a decision it had already taken + // and frozen — the same bytes it emitted on an earlier turn, spliced again with no model + // call and no new spend. + // + // It exists because /stats could not tell a replay from an act (#176). `acted` is + // `Saved() > 0`, and a replay saves tokens, so a component whose every activation was a + // free replay of work done days ago reported `acted: 239` — indistinguishable from 239 + // paid extractions. MEASURED (iteration 023, arm B): 239 acts, 2,291 replays, and the + // component's own record showing zero surviving candidates, i.e. not one fresh call. The + // two readings have opposite consequences ("this component is expensive, turn it off" + // against "this component is amortizing, leave it on"), so they cannot share a counter. + // + // Set via Replay(), which also files the descriptive Event, so the count and the histogram + // cannot drift apart. + Replays int } // TokenRates are per-token USD rates for the model a component would call ITSELF, so a @@ -654,6 +669,22 @@ func (r *Report) Event(name string) { r.Events[name]++ } +// Replay records that one change on this request came from replaying an already-frozen +// decision, under the descriptive event name the component uses for it. +// +// It files BOTH the Event and Report.Replays deliberately, in one call: the Events histogram is +// the operator-facing vocabulary ("reapplied_same_session" vs "reapplied_cross_session") and +// Replays is the machine-readable "this cost nothing" that /stats needs to keep replays out of +// `acted`. Two separate calls at each site is how one of them gets forgotten when a third +// replay path is added — which is precisely how the free path came to be counted as paid work. +func (r *Report) Replay(name string) { + if r == nil { + return + } + r.Event(name) + r.Replays++ +} + // EventN records n at once, for an event whose subject is a COUNT rather than a single candidate — // how many were offered, how many were removed. GateN's rationale applies unchanged. func (r *Report) EventN(name string, n int) { diff --git a/components/offload/extract_llm.go b/components/offload/extract_llm.go index d7715152..8b2a013a 100644 --- a/components/offload/extract_llm.go +++ b/components/offload/extract_llm.go @@ -703,7 +703,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R if model != nil && !pressureFires { model = nil // no model call this request; frozen reapplications still run below } - metrics.RecordExtractionReason(triggerReason) + metrics.RecordExtractionReason(rep.Component, triggerReason) floor := e.outputFloor(c.CtxWindow) // Without an explicit min_tokens, derive the per-output floor from context pressure so @@ -866,18 +866,18 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R // independent TTLs and pin slots, so a replay could hit one and miss the other and // emit HALF a decision — projected text with the summary segment silently gone. if cached, hit := getResult(c, id); hit { - metrics.RecordExtractionCacheLookup(true) + metrics.RecordExtractionCacheLookup(rep.Component, true) // A REPLAY is where the amortization actually happens, so credit it — at the rate // a re-sent token would have been billed at, which on a caching backend is the // cache-read rate. This is the other half of the honest net figure: the first // application alone under-reports the value, and pricing the replays at the first // application's rate over-reports it by 12.5x. if saved := schema.TextTokens(content) - schema.TextTokens(cached.Projected); saved > 0 { - metrics.RecordExtractionValue(float64(saved) * val.repeatPerToken) + metrics.RecordExtractionValue(rep.Component, float64(saved)*val.repeatPerToken) } apply(i, content, cached.Projected, cached.Summary) dbgReapply++ - rep.Event("reapplied_same_session") + rep.Replay("reapplied_same_session") continue } // A NEW compaction, on the UNCACHED region only (cache-safe): when cache-aware that @@ -942,13 +942,13 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R // projected text with the summary segment silently gone. if !e.rewrite || effectiveMode(c, e.mode) == markerFull { if cached, hit := getResultGlobal(c, extract.ResultKey(id, e.modelName, extCfg)); hit { - metrics.RecordExtractionCacheLookup(true) + metrics.RecordExtractionCacheLookup(rep.Component, true) // Freeze into this session so later turns replay it byte-for-byte from the // same-session path above, at any depth. putResult(c, id, cached.Projected, cached.Summary) apply(i, content, cached.Projected, cached.Summary) dbgReapply++ - rep.Event("reapplied_cross_session") + rep.Replay("reapplied_cross_session") continue } } @@ -964,7 +964,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R // whose real rate over reachable candidates is 24.0% — 30 replays per model call, one // of the few parts of this component that unambiguously pays. A metric that argues for // optimizing something already working is worse than no metric. - metrics.RecordExtractionCacheLookup(false) + metrics.RecordExtractionCacheLookup(rep.Component, false) // The operator's REQUEST-level trigger, honored on every turn this component sees. // // This condition used to carry a cold-sweep carve-out, and before that `!c.CacheAware`. @@ -1018,7 +1018,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R if e.gate { // Stop exploring once calls are observed to be slow: exploration spends wall // clock as well as money, and an agent on a task deadline feels the former more. - explore := !tooSlowToExplore(metrics.ExtractionP50LatencyMs()) && + explore := !tooSlowToExplore(metrics.ExtractionP50LatencyMs(rep.Component)) && e.ratios.exploring(c.Session) // goalOverhead, not promptOverhead: the gate needs the VARIABLE part of the // prompt, because callCost adds the static preamble itself. promptOverhead is @@ -1058,7 +1058,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R // nothing was suppressed, and inflating that would make the gate look // like it was working when it has been overridden. d.reason = "advisory: " + d.reason - metrics.RecordExtractionReason(d.reason) + metrics.RecordExtractionReason(rep.Component, d.reason) rep.Gate("economic_gate_advisory") if dbg { logging.From(c.Ctx).Debug("cg.extract_llm.gate", "decision", "advisory", @@ -1068,7 +1068,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R d.allow = true } if !d.allow { - metrics.RecordExtractionSuppressed(d.reason) + metrics.RecordExtractionSuppressed(rep.Component, d.reason) // Just the gate name here: the per-reason breakdown already ships in // /stats via RecordExtractionSuppressed, and a full sentence makes a // poor histogram key. @@ -1090,7 +1090,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R } continue } - metrics.RecordExtractionReason(d.reason) + metrics.RecordExtractionReason(rep.Component, d.reason) gateReason = d.reason } // A class whose measured reduction cannot support a fixed-size window must not be @@ -1220,7 +1220,7 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R return } latency := float64(time.Since(start).Milliseconds()) - metrics.RecordExtractionCall(latency) + metrics.RecordExtractionCall(rep.Component, latency) _, inTok, outTok := callSink.Totals() cw, cr := callSink.CacheTotals() calls[k] = components.ModelCall{ @@ -1235,6 +1235,12 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R Before: cands[k].content, Rejection: why, } + // THIS component's spend, at THIS component's rates. /stats used to derive + // extraction cost from cheapmodel's process-global token totals through one rate + // card, which is neither this component's spend nor extraction's: `summarize` and + // `agentdiet` land in the same totals, and extract_llm_sweep pays the request's own + // frontier model while the card is haiku's. See metrics.RecordExtractionSpend. + metrics.RecordExtractionSpend(rep.Component, calls[k].CostUSD) // A reply that stopped exactly at the output cap was TRUNCATED, so the // Starlark program is incomplete, unparseable, and the whole call — its // money and its seconds — bought nothing. It is indistinguishable from @@ -1276,10 +1282,10 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R // Feed the observed ratio so the gate prices future calls on what this // workload actually achieves, not on an assumption. e.ratios.observe(before-schema.TextTokens(res), before) - metrics.RecordExtractionSaving(before - schema.TextTokens(res)) + metrics.RecordExtractionSaving(rep.Component, before-schema.TextTokens(res)) // What the removal was WORTH, at this turn's regime. On a cold sweep that is // the cache-write rate; the replays below are credited at the read rate. - metrics.RecordExtractionValue(float64(before-schema.TextTokens(res)) * val.perToken) + metrics.RecordExtractionValue(rep.Component, float64(before-schema.TextTokens(res))*val.perToken) } else if !timedOut { e.ratios.observe(0, before) // a miss is real evidence: ratio 0 } @@ -1304,6 +1310,38 @@ func (e *ExtractLLM) Offload(req *bschemas.BifrostChatRequest, rep *components.R // observation leaves the gate's estimate untouched; the timeouts are still // counted (above) and still brake exploration via slowCallMs, which is the // latency-aware layer that SHOULD react to a slow server. + + // ONE RECORD PER CALL (#177). Until this existed the component's only message was + // `cg.extract_llm`, one per request, carrying the DECISION and nothing about the + // CALL — so a run credited with 101 calls at 59,009 ms mean latency and a net value + // of -$1.162 had no per-request trace at all. Three things were unanswerable and + // each of them stopped an investigation: which requests made the calls, whether a + // 59-second mean was 101 slow calls or a few multi-minute outliers dragging it (the + // two have opposite fixes), and which candidates lost money. extract_llm_sweep's + // `cg.sweep.ask` already makes its economics reconstructable; this is the same for + // the tail pass. + // + // `accepted` is the never-worse outcome — the same condition that spliced the result + // above, so the log cannot say accepted while the request kept the original — and + // `rejection` is why a call produced nothing when it did not. Both, not one: an + // empty rejection on a rejected call is what made timeout, sandbox refusal and + // "nothing shrank" indistinguishable. + // + // DEBUG-gated by `dbg`, resolved once per request: the strings below are cheap but + // they are per CALL, and the repo's rule is that a payload costing anything to build + // is guarded. `content_key` rather than the content — the key is what the result + // cache, the freeze and the cross-session lookup are all keyed on, so it is the + // identity that joins this record to every other one about the same candidate. + if dbg { + logging.From(c.Ctx).Debug("cg.extract_llm.call", + "session", c.Session, "content_key", cands[k].id, + "candidate_tokens", before, "model", callModel, + "latency_ms", latency, "input_tokens", inTok, "output_tokens", outTok, + "cache_read", cr, "cache_write", cw, "cost_usd", calls[k].CostUSD, + "accepted", calls[k].Accepted, "saved_tokens", calls[k].SavedTokens, + "rejection", calls[k].Rejection, "gate", cands[k].gate, + "strategy", strategy, "timed_out", timedOut) + } } for k := 0; k < len(cands); k++ { wg.Add(1) diff --git a/components/offload/extract_llm_callrecord_test.go b/components/offload/extract_llm_callrecord_test.go new file mode 100644 index 00000000..18817097 --- /dev/null +++ b/components/offload/extract_llm_callrecord_test.go @@ -0,0 +1,189 @@ +package offload + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "strings" + "testing" + + bschemas "github.com/maximhq/bifrost/core/schemas" + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/internal/logging" + "github.com/rossoctl/context-guru/store" +) + +// shrinkingModel returns a Starlark program that drops every line but the first, so the +// extraction is ACCEPTED and the record under test carries a real saving rather than a +// rejection. Nothing here is about the program's cleverness: the test needs the accept branch. +type shrinkingModel struct{ calls int } + +func (m *shrinkingModel) Complete(_ context.Context, _ string) (string, error) { + m.calls++ + return "```python\ndef transform(text):\n return text.split(\"\\n\")[0]\n```", nil +} + +// debugCtx returns a context carrying a DEBUG-level JSON logger and the buffer it writes to. +func debugCtx(t *testing.T) (context.Context, *bytes.Buffer) { + t.Helper() + var buf bytes.Buffer + l := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + ctx := logging.With(context.Background(), l) + if !logging.Debugging(ctx) { + t.Fatal("the fixture's logger is not at DEBUG, so a guarded record would be skipped " + + "and this test would pass vacuously") + } + return ctx, &buf +} + +// records returns every logged record whose msg matches. +func records(t *testing.T, buf *bytes.Buffer, msg string) []map[string]any { + t.Helper() + var out []map[string]any + for _, ln := range strings.Split(buf.String(), "\n") { + if strings.TrimSpace(ln) == "" { + continue + } + var m map[string]any + if err := json.Unmarshal([]byte(ln), &m); err != nil { + t.Fatalf("log line is not JSON: %q (%v)", ln, err) + } + if m["msg"] == msg { + out = append(out, m) + } + } + return out +} + +// THE DEFECT (#177). extract_llm emitted exactly one message, `cg.extract_llm`, once per +// request, carrying the DECISION — tools, cands, skip_tail, skip_floor, floor — and nothing +// about the CALL. So /stats could attribute 101 calls at 59,009 ms mean latency and a net value +// of -$1.162 to this component with no per-request trace to check it against, and three +// questions were unanswerable: which requests made the calls, whether the 59-second mean was +// many slow calls or a few multi-minute outliers (opposite fixes), and which candidates lost +// money. extract_llm_sweep's `cg.sweep.ask` already made its economics reconstructable. +// +// This pins the record's EXISTENCE and its economically load-bearing fields. It asserts on the +// rendered log line, not on a struct: the fields are handed to slog as a variadic list, so one +// could be computed correctly and never reach the handler. +func TestExtractLLMLogsOneRecordPerCall(t *testing.T) { + model := &shrinkingModel{} + e := newTimeoutTestComponent(t, model) // min_tokens: 1, strategy: code, gate off + ctx, buf := debugCtx(t) + + original := strings.Repeat("2026-08-31T10:00:00Z INFO worker: processed batch\n", 400) + req := &bschemas.BifrostChatRequest{ + Input: []bschemas.ChatMessage{ + userMsg("Summarize the worker log and tell me if any batch failed."), + toolResultMsg(original), + }, + } + c := &components.Ctx{ + Session: "callrecord-test", + Store: store.NewMemory(store.Options{}), + Ctx: ctx, + Model: components.ModelSpec{Static: model, Incoming: model}, + } + rep := &components.Report{Component: "extract_llm"} + if _, err := e.Offload(req, rep, c); err != nil { + t.Fatalf("Offload: %v", err) + } + // Without a call there is nothing to record, and the assertions below would pass + // vacuously on an empty slice — the failure mode this repo has been bitten by. + if model.calls == 0 { + t.Fatal("the model was never called, so the call-record path was never reached") + } + if len(rep.Calls) == 0 { + t.Fatal("no ModelCall was reported, so there was no call to log") + } + + got := records(t, buf, "cg.extract_llm.call") + if len(got) != len(rep.Calls) { + t.Fatalf("got %d cg.extract_llm.call records for %d reported calls; #177 is that "+ + "there were 0. Log was:\n%s", len(got), len(rep.Calls), buf.String()) + } + rec := got[0] + // The economics of one call must be reconstructable from this line alone: whose session, + // which candidate, how big, on what model, how long, what it consumed, what it cost, + // whether the never-worse check took it, and what that bought. + for _, k := range []string{ + "session", "content_key", "candidate_tokens", "model", "latency_ms", + "input_tokens", "output_tokens", "cost_usd", "accepted", "saved_tokens", "rejection", + } { + if _, ok := rec[k]; !ok { + t.Errorf("cg.extract_llm.call is missing %q — the field it exists to carry", k) + } + } + if rec["session"] != "callrecord-test" { + t.Errorf("session = %v, want callrecord-test: a call that cannot be tied to a "+ + "request is the exact gap #177 reports", rec["session"]) + } + if rec["content_key"] == "" || rec["content_key"] == nil { + t.Error("content_key is empty; without the candidate's identity the record cannot " + + "be joined to the replay, freeze and cache lookups keyed on it") + } + if ct, _ := rec["candidate_tokens"].(float64); ct <= 0 { + t.Errorf("candidate_tokens = %v, want the candidate's real size", rec["candidate_tokens"]) + } + // ACCEPT/REJECT must agree with what the component actually did. A record that says + // accepted while the request kept the original is worse than no record. + if rec["accepted"] != rep.Calls[0].Accepted { + t.Errorf("accepted = %v but the reported call says %v", rec["accepted"], + rep.Calls[0].Accepted) + } + if rep.Calls[0].Accepted { + if st, _ := rec["saved_tokens"].(float64); st <= 0 { + t.Errorf("saved_tokens = %v on an accepted extraction", rec["saved_tokens"]) + } + } + // latency_ms may legitimately be 0 on a fast fake model, so the assertion is on the + // FIELD's presence (above) rather than on a value the fixture cannot guarantee. + if _, ok := rec["latency_ms"].(float64); !ok { + t.Errorf("latency_ms = %v, want a number", rec["latency_ms"]) + } +} + +// The record must be DEBUG-guarded, per this repo's stated rule: at INFO nothing may be +// emitted. This also proves the record is not being written through a path that ignores the +// level (a fmt.Println, a logger captured at construction). +// +// VACUITY, stated because it is not what it looks like: NO SINGLE-POINT MUTATION KILLS THIS +// TEST. The property is defended twice over — by the `if dbg` guard, which exists so the +// payload is not built when nobody is reading, and by the slog level on Debug(), which exists +// so it is not printed. Remove the guard and the level still suppresses it; switch Debug to +// Info and the guard still skips it. Verified: `Debug(` -> `Info(` alone leaves this test +// PASSING. Only the combined mutant (`if dbg` -> `if true` AND `Debug` -> `Info`) fails it, +// which was run and does fail. So read this test as pinning the OUTCOME, not either mechanism, +// and do not take its passing as evidence that the guard is present — TestExtractLLMLogsOneRecordPerCall +// is the one that dies when the record itself goes away. +func TestExtractLLMCallRecordIsDebugGated(t *testing.T) { + model := &shrinkingModel{} + e := newTimeoutTestComponent(t, model) + var buf bytes.Buffer + l := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + ctx := logging.With(context.Background(), l) + + original := strings.Repeat("2026-08-31T10:00:00Z INFO worker: processed batch\n", 400) + req := &bschemas.BifrostChatRequest{ + Input: []bschemas.ChatMessage{ + userMsg("Summarize the worker log and tell me if any batch failed."), + toolResultMsg(original), + }, + } + c := &components.Ctx{ + Session: "callrecord-info", + Store: store.NewMemory(store.Options{}), + Ctx: ctx, + Model: components.ModelSpec{Static: model, Incoming: model}, + } + if _, err := e.Offload(req, &components.Report{Component: "extract_llm"}, c); err != nil { + t.Fatalf("Offload: %v", err) + } + if model.calls == 0 { + t.Fatal("the model was never called, so this proves nothing about the guard") + } + if strings.Contains(buf.String(), "cg.extract_llm.call") { + t.Errorf("the per-call record was emitted at INFO: %s", buf.String()) + } +} diff --git a/components/offload/extract_sweep.go b/components/offload/extract_sweep.go index 97b62e98..5c0308c8 100644 --- a/components/offload/extract_sweep.go +++ b/components/offload/extract_sweep.go @@ -305,16 +305,16 @@ func (e *ExtractSweep) Offload(req *bschemas.BifrostChatRequest, rep *components // came from a model, but the REPLACEMENT is a pure function of (content, config), so a replay // can never emit different bytes than the turn that decided it. if cached, hit := getResult(c, id); hit { - metrics.RecordExtractionCacheLookup(true) + metrics.RecordExtractionCacheLookup(rep.Component, true) if saved := schema.TextTokens(content) - schema.TextTokens(cached.Projected); saved > 0 { - metrics.RecordExtractionValue(float64(saved) * val.repeatPerToken) + metrics.RecordExtractionValue(rep.Component, float64(saved)*val.repeatPerToken) } if k, ok := applySweepDrop(c, rep, e.mode, msg, content); ok { changed++ if k != "" { keys = append(keys, k) } - rep.Event("reapplied_same_session") + rep.Replay("reapplied_same_session") } continue } @@ -323,7 +323,7 @@ func (e *ExtractSweep) Offload(req *bschemas.BifrostChatRequest, rep *components // all such a turn has to do. continue } - metrics.RecordExtractionCacheLookup(false) + metrics.RecordExtractionCacheLookup(rep.Component, false) if schema.TextTokens(content) < e.minTokens { rep.Gate("below_output_floor") continue @@ -544,10 +544,14 @@ func (r *sweepResult) event(name string) { r.events = append(r.events, name) } // obligation, a verdict for something we did not offer. A wrong keep costs tokens on one turn; a wrong // drop is a silent permanent loss the agent does not notice and cannot ask about. The two errors are // not comparable, so this does not treat them symmetrically. +// The sweepResult is a NAMED result purely so `defer foldFallback()` below can reach it. It has to +// be: every early return here is `return nil, r` on a LOCAL, and a deferred mutation of a local +// happens after the return value has already been copied, so the fold would be silently lost on +// exactly the error paths it exists to cover. The first result stays blank-named — only this one +// needs the treatment, and saying so beats leaving a reader to wonder what `dropped` is for. func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *components.Ctx, - rep *components.Report, cands []sweepCand) ([]int, sweepResult) { + rep *components.Report, cands []sweepCand) (_ []int, r sweepResult) { - var r sweepResult // A SINGLE-CANDIDATE ASK IS THE REFUTED SHAPE WEARING A NEW NAME, so it is counted rather than // silently accepted. Shown one output, a model simply drops it: 6% live-kept on haiku and 14% on // sonnet, both inside the drop-everything null model's error bar. The ask still proceeds — a @@ -583,7 +587,26 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component for _, it := range items { before += it.SizeTokens } - start := time.Now() + // COST, which this record carried as $0.00 forever. It never set CostUSD at all, so the per-call + // ledger the dashboard shows for this component reported zero on every firing — measured live at + // $0.00 against real cache reads of 449,304 and 449,376 tokens and real completion tokens, while + // the request-level rollup (proxy/dashcapture.go, cg_llm_cost_usd) had the true $0.0940 and + // $0.1652. Two recorded totals disagreeing, one of them structurally zero, is worse than either + // alone: a component whose whole justification is cost looked free. + // + // Priced from the REQUEST's model, not a cheap-model card, because that is what this component + // calls by construction — and from the same rates the request-level figure uses, so the two agree + // rather than being two independent guesses. c.SelfRates is the model the request came in on; + // falling back to the env card keeps a figure when the host supplies no rates, the same + // convention extract_llm.pricingFor uses. + // + // Resolved BEFORE the ask because both legs below price themselves with it. The fallback goes to + // the same model the prefix ask addresses (c.Model.For("incoming")), so one rate card is right + // for both. + pricing := cheapmodel.PricingFromEnv() + if !c.SelfRates.Zero() { + pricing = ratesPricing(c.SelfRates) + } var ( reply string usage components.PrefixUsage @@ -592,7 +615,91 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component // not fire a second time for the same call. Without it a failed ask both fell back AND then // reported a zero cache read, double-counting one event as two. fellBack bool + // The ask's own totals, accumulated PER LEG. An adjudication is not one model call: the + // prefix ask and the fallback are two, on two prompts, and either can be the only one that + // happens. See recordLeg. + askMs, askCost float64 + fbUsage components.PrefixUsage + fbMs, fbCost float64 ) + // recordLeg books ONE model call this component made — its wall time and its own priced spend. + // + // PER LEG, not once per adjudication, because `fallbackAsk` is a SECOND model call on a full + // sampled transcript and it used to be accounted at $0.00 and 0 ms. The single record was built + // from the PREFIX ask's usage and assigned once, while two of the three fallback points fire + // after that assignment — so on every route without an Anthropic prefix asker (where the prefix + // ask never happens at all), on every session's first turn (ErrNoPrefix) and on every mistimed + // window (CacheRead == 0), the component's real frontier-model spend was reported as free. + // + // That was survivable before this PR only by accident: the fallback's tokens still reached + // /stats through cheapmodel's process totals, which the host passed in as the `cost` argument. + // Making the components price their own spend removed that accident, so the leg has to book + // itself. This is the same defect class as #176, in the component this PR re-scoped. + recordLeg := func(ms float64, u components.PrefixUsage) float64 { + metrics.RecordExtractionCall(rep.Component, ms) + cost := pricing.Cost(int64(u.Fresh), int64(u.Output), + int64(u.CacheWrite), int64(u.CacheRead)) + metrics.RecordExtractionSpend(rep.Component, cost) + return cost + } + // runFallback runs the expensive path and books it as its own leg, at all three call sites, so a + // fourth one cannot be added without the accounting coming with it. It records even when the + // call ERRORS: a failed completion still burned wall time, and on some failures tokens. + runFallback := func() (string, error) { + out, u, ms, err := e.fallbackAsk(ctx, req, c, &r, items, cands) + fbUsage, fbMs = u, ms + fbCost = recordLeg(ms, u) + return out, err + } + // foldFallback adds the fallback leg's tokens, dollars and wall time into the ask's ledger row. + // + // Idempotent by construction — at most one fallback runs per adjudication and this READS the + // accumulators rather than adding to them — which is what lets it be deferred and also called + // explicitly on the happy path, where the row is built after the no-asker fallback has already + // run and would otherwise overwrite the fold. + // + // DEFERRED, so it also covers the three fallback ERROR paths. Each of those is + // `if reply, err = runFallback(); err != nil { return nil, r }`, and `runFallback` has already + // booked the leg into metrics via recordLeg — so /stats counted the call and its seconds while + // the ledger row kept only the prefix ask's LatencyMs and a Strategy naming a leg that was no + // longer the only one that ran. Latency rather than dollars, because on an error the sink is + // empty (recordUsageCache is reached only after a successful decode on both backends, and + // neither returns an error after billing) — but the fallback is the SLOW leg by construction, so + // a failed one contributes tens of seconds to avg_latency_ms against a row showing milliseconds. + foldFallback := func() { + if fbMs == 0 && fbCost == 0 && fbUsage == (components.PrefixUsage{}) { + return + } + // IDENTITY FIRST, and this is the half of the fix that is easy to miss. On the no-asker + // path the fallback runs BEFORE r.rec exists, so if it errors the row is never built at + // all — Component stays "" and the caller drops the whole row on the + // `call.rec.Component != ""` guard. /stats then reports a call the ledger has no row for. + // That path never built a row before either, so the row is not a regression; the recorded + // spend and latency are new, so the DIVERGENCE is. + if r.rec.Component == "" { + r.rec.Component = rep.Component + r.rec.Model = c.ModelName + r.rec.CandidateTokens = before + r.rec.GateReason = "pre-expiry window: the cache still exists and is nearly worthless" + } + r.rec.LatencyMs = askMs + fbMs + r.rec.PromptTokens = int64(usage.Fresh + fbUsage.Fresh) + r.rec.CompletionTokens = int64(usage.Output + fbUsage.Output) + r.rec.CacheRead = int64(usage.CacheRead + fbUsage.CacheRead) + r.rec.CacheWrite = int64(usage.CacheWrite + fbUsage.CacheWrite) + r.rec.CostUSD = askCost + fbCost + // Name what actually ran. On the no-asker route the prefix ask never happens, so calling + // the row "prefix_ask+fallback" would report a leg that did not exist. + if c.PrefixAsk == nil { + r.rec.Strategy = "fallback" + } else { + r.rec.Strategy = "prefix_ask+fallback" + } + } + // One arming point for all four exits that can carry a fallback — the three error returns and + // the happy path. A per-site call was what left the error paths uncovered, and a fifth site + // added later would have been missed the same way. + defer foldFallback() if c.PrefixAsk == nil { // No asker at all: a non-Anthropic route, or no incoming client. Not a failure of the ask — // there was nothing to ask through — so it takes the same fork as a missed read. @@ -602,40 +709,35 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component r.rec.Rejection = "no prefix asker on this route and block_fallback is set" return nil, r } - if reply, err = e.fallbackAsk(ctx, req, c, &r, items, cands); err != nil { + if reply, err = runFallback(); err != nil { return nil, r } fellBack = true } else { + askStart := time.Now() reply, usage, err = c.PrefixAsk.Ask(ctx, c.Session, extract.BuildPrefixAsk(items)) - } - latency := float64(time.Since(start).Milliseconds()) - metrics.RecordExtractionCall(latency) - // COST, which this record carried as $0.00 forever. It never set CostUSD at all, so the per-call - // ledger the dashboard shows for this component reported zero on every firing — measured live at - // $0.00 against real cache reads of 449,304 and 449,376 tokens and real completion tokens, while - // the request-level rollup (proxy/dashcapture.go, cg_llm_cost_usd) had the true $0.0940 and - // $0.1652. Two recorded totals disagreeing, one of them structurally zero, is worse than either - // alone: a component whose whole justification is cost looked free. - // - // Priced from the REQUEST's model, not a cheap-model card, because that is what this component - // calls by construction — and from the same rates the request-level figure uses, so the two agree - // rather than being two independent guesses. c.SelfRates is the model the request came in on; - // falling back to the env card keeps a figure when the host supplies no rates, the same - // convention extract_llm.pricingFor uses. - pricing := cheapmodel.PricingFromEnv() - if !c.SelfRates.Zero() { - pricing = ratesPricing(c.SelfRates) + askMs = float64(time.Since(askStart).Milliseconds()) + // ErrNoPrefix is refused LOCALLY — there is no stashed body to append to, so no request + // leaves the process. Booking it as a call would put a 0 ms, $0 sample into the mean that + // the exploration brake reads, and inflate `calls` with work that by definition did not + // happen. Every other outcome, transport failure included, went to the provider. + if !errors.Is(err, components.ErrNoPrefix) { + askCost = recordLeg(askMs, usage) + } } r.rec = components.ModelCall{ Component: rep.Component, Model: c.ModelName, Strategy: "prefix_ask", - CandidateTokens: before, LatencyMs: latency, + CandidateTokens: before, LatencyMs: askMs, PromptTokens: int64(usage.Fresh), CompletionTokens: int64(usage.Output), CacheRead: int64(usage.CacheRead), CacheWrite: int64(usage.CacheWrite), - CostUSD: pricing.Cost(int64(usage.Fresh), int64(usage.Output), - int64(usage.CacheWrite), int64(usage.CacheRead)), + CostUSD: askCost, GateReason: "pre-expiry window: the cache still exists and is nearly worthless", } + // The no-asker path's fallback ran BEFORE this row existed, and the assignment above just + // overwrote the deferred fold's work-in-progress. Re-folded here rather than relying on the + // defer alone because the defer's ordering relative to this assignment is what makes that + // reliance fragile — and folding twice is free, since foldFallback reads the accumulators. + foldFallback() if ctx.Err() != nil { if errors.Is(ctx.Err(), context.DeadlineExceeded) { atomic.AddInt64(&llmTimeouts, 1) @@ -657,7 +759,7 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component r.rec.Rejection = "prefix ask failed and block_fallback is set: " + err.Error() return nil, r } - if reply, err = e.fallbackAsk(ctx, req, c, &r, items, cands); err != nil { + if reply, err = runFallback(); err != nil { return nil, r } fellBack = true @@ -687,7 +789,7 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component "declining rather than paying again for a full-price transcript read" return nil, r } - if reply, err = e.fallbackAsk(ctx, req, c, &r, items, cands); err != nil { + if reply, err = runFallback(); err != nil { return nil, r } fellBack = true @@ -814,8 +916,8 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component r.event("sweep_dropped") drop = append(drop, v.Label) removed += sz - after - metrics.RecordExtractionSaving(sz - after) - metrics.RecordExtractionValue(float64(sz-after) * savedTokenValue(c).perToken) + metrics.RecordExtractionSaving(rep.Component, sz-after) + metrics.RecordExtractionValue(rep.Component, float64(sz-after)*savedTokenValue(c).perToken) } // An output named in the inventory that no verdict mentioned is UNJUDGED, and it must not look // like a keep: 4ca1f13 found a live arm where the model silently omitted labels and the missing @@ -851,14 +953,18 @@ func (e *ExtractSweep) adjudicate(req *bschemas.BifrostChatRequest, c *component // // The reply budget is raised through components.Budgeter where the client supports it, for the same // reason the prefix ask raises it: one reply carries a verdict for every candidate. +// It returns its OWN usage and wall time so the caller can price it. `model.Complete` reports +// neither, so the usage is read from a per-call cheapmodel sink nested inside whatever scope already +// wraps ctx — the same construction extract_llm uses to attribute one call. The sink also reaches +// every ancestor, so the request's own bill does not lose these tokens. func (e *ExtractSweep) fallbackAsk(ctx context.Context, req *bschemas.BifrostChatRequest, c *components.Ctx, r *sweepResult, items []extract.AdjudicationItem, - cands []sweepCand) (string, error) { + cands []sweepCand) (string, components.PrefixUsage, float64, error) { model := c.Model.For("incoming") if model == nil { r.gate("sweep_fallback_no_model") r.rec.Rejection = "the prefix ask could not read the cache and no request model is available" - return "", errNoFallbackModel + return "", components.PrefixUsage{}, 0, errNoFallbackModel } if b, ok := model.(components.Budgeter); ok { if m := b.WithMaxTokens(cheapmodel.PrefixAskMaxTokens); m != nil { @@ -873,13 +979,22 @@ func (e *ExtractSweep) fallbackAsk(ctx context.Context, req *bschemas.BifrostCha withSamples[i] = it } r.event("sweep_fallback_used") + ctx, sink := cheapmodel.WithCallSink(ctx) + start := time.Now() reply, err := model.Complete(ctx, extract.BuildFallbackAsk(sweepIntent(req), withSamples)) + ms := float64(time.Since(start).Milliseconds()) + // Read the sink whatever happened: a completion that failed after the provider billed it still + // cost money, and returning zeros there is how spend goes missing. + _, inTok, outTok := sink.Totals() + cw, cr := sink.CacheTotals() + u := components.PrefixUsage{Fresh: int(inTok), Output: int(outTok), + CacheWrite: int(cw), CacheRead: int(cr)} if err != nil { r.gate("sweep_fallback_failed") r.rec.Rejection = "fallback completion failed: " + err.Error() - return "", err + return "", u, ms, err } - return reply, nil + return reply, u, ms, nil } // sweepIntent renders the conversation's intent for a SPENT-NESS judgement, which wants it ordered diff --git a/components/offload/extract_sweep_fallbackcost_test.go b/components/offload/extract_sweep_fallbackcost_test.go new file mode 100644 index 00000000..bc97baa8 --- /dev/null +++ b/components/offload/extract_sweep_fallbackcost_test.go @@ -0,0 +1,300 @@ +package offload + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/internal/cheapmodel" + "github.com/rossoctl/context-guru/metrics" + "github.com/rossoctl/context-guru/store" +) + +// billingModel is a real cheapmodel client pointed at a local server that reports usage. It exists +// because `recordingModel` reports none, and this test's whole subject is that the fallback's TOKENS +// are priced — a fake that bills nothing would let the assertion pass against a fix that plumbs +// nothing. Going through cheapmodel.Anthropic also exercises the actual recording path +// (recordUsageCache -> the call sink), which is what fallbackAsk now reads. +func billingModel(t *testing.T, reply string) (components.Model, func()) { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"content":[{"type":"text","text":` + jsonQuote(reply) + `}],` + + `"usage":{"input_tokens":31000,"output_tokens":420,` + + `"cache_creation_input_tokens":0,"cache_read_input_tokens":0}}`)) + })) + return cheapmodel.Anthropic{BaseURL: srv.URL, APIKey: "k", Model: "claude-sonnet-5", + MaxTokens: 2048}, srv.Close +} + +// sweepSpend reads this component's recorded extraction economics out of /stats' breakdown. +func sweepSpend(t *testing.T) (cost float64, calls int64, source string) { + t.Helper() + s := metrics.ExtractSnapshot(0, 0, 0, 0) + row := s.ByComponent["extract_llm_sweep"] + if row == nil { + return 0, 0, "" + } + return row.ExtractionCostUSD, row.Calls, row.CostSource +} + +// REVIEW FINDING (#178, MAJOR). `fallbackAsk` is a SECOND model call, on the request's own frontier +// model, carrying a sampled copy of every candidate — the expensive path by construction. Its cost +// and its wall time were both dropped from this component's accounting: +// +// - `r.rec` was built from the PREFIX ask's usage and assigned exactly once, and `fallbackAsk` +// returned only (string, error) — no usage, no cost. So RecordExtractionSpend saw $0. +// - RecordExtractionCall was called before two of the three fallback points, so the fallback's +// seconds never reached avg_latency_ms either. +// +// Before this PR that spend still reached /stats by accident, through cheapmodel's process totals +// which the host passed in as the `cost` argument. Making components price their own spend removed +// the accident: as soon as any component records spend the host fallback is discarded, so +// `extract_llm` + sweep behind a non-Anthropic upstream — where the prefix asker is nil and the +// sweep ALWAYS falls back — reported a component whose entire justification is cost as free, with a +// purely positive net_value_usd. +// +// This drives the no-asker path, which is that exact scenario. +func TestSweepFallbackAskPricesItsOwnCall(t *testing.T) { + model, done := billingModel(t, `[{"i":0,"needed_by":"none","quote":"","verdict":"keep"}]`) + defer done() + + costBefore, callsBefore, _ := sweepSpend(t) + e := newSweepSmall(t, "") + // preExpiryCtx with NO prefix asker: any non-Anthropic route, which is the reviewer's scenario. + c := preExpiryCtx("fallbackcost", nil, store.NewMemory(store.Options{})) + c.Model = components.ModelSpec{Incoming: model, Static: model} + + rep := &components.Report{Component: "extract_llm_sweep"} + if _, err := e.Offload(sweepReqStocked(), rep, c); err != nil { + t.Fatal(err) + } + // PRECONDITIONS. Without them every assertion below could pass because nothing ran. + if rep.Gates["sweep_no_asker"] != 1 { + t.Fatalf("the no-asker path was not taken, so the fallback was never reached (gates: %v)", + rep.Gates) + } + if rep.Events["sweep_fallback_used"] != 1 { + t.Fatalf("the fallback did not run (events: %v gates: %v)", rep.Events, rep.Gates) + } + if len(rep.Calls) == 0 { + t.Fatalf("no ModelCall was reported for an adjudication that made one (gates: %v)", + rep.Gates) + } + + // THE LEDGER ROW: the dashboard's per-call view must not show the expensive leg as free. + rec := rep.Calls[0] + if rec.CostUSD <= 0 { + t.Errorf("the ledger row prices the fallback at $%v — the defect this test exists for; "+ + "31,000 prompt + 420 completion tokens on a frontier model is not free", rec.CostUSD) + } + if rec.PromptTokens != 31000 || rec.CompletionTokens != 420 { + t.Errorf("the fallback's tokens did not reach the row: prompt=%d completion=%d, want "+ + "31000/420", rec.PromptTokens, rec.CompletionTokens) + } + if rec.Strategy != "fallback" { + t.Errorf("Strategy = %q, want %q: no prefix ask happened on this route, so the row must "+ + "not claim a leg that did not exist", rec.Strategy, "fallback") + } + + // AND /stats: the counters the operator and the alert rule read. + costAfter, callsAfter, source := sweepSpend(t) + if callsAfter-callsBefore != 1 { + t.Errorf("recorded %d calls for this adjudication, want 1", callsAfter-callsBefore) + } + if costAfter-costBefore <= 0 { + t.Errorf("extract_llm_sweep.extraction_cost_usd moved by $%v: the fallback's spend is "+ + "still missing from /stats", costAfter-costBefore) + } + if source != "component" { + t.Errorf("cost_source = %q, want \"component\": the component priced this call itself", + source) + } +} + +// The other two fallback paths reach it AFTER the prefix ask's record already exists, so they are +// where the fold matters: the row has to grow to cover both legs rather than keep the prefix ask's +// numbers. This drives the zero-cache-read path — a mistimed window, which the counters say is the +// common one — and pins that the adjudication books TWO calls, not one. +func TestSweepCountsBothLegsWhenItFallsBackAfterAsking(t *testing.T) { + model, done := billingModel(t, `[{"i":0,"needed_by":"none","quote":"","verdict":"keep"}]`) + defer done() + + // A prefix ask that succeeds but reads nothing from cache: usage is real (Fresh 40, Output 90) + // and CacheRead is 0, so the component asks again through the expensive path. + asker := &fakeAsker{reply: `[{"i":0,"needed_by":"none","quote":"","verdict":"keep"}]`, cacheRead: 0} + _, callsBefore, _ := sweepSpend(t) + e := newSweepSmall(t, "") + c := preExpiryCtx("bothlegs", asker, store.NewMemory(store.Options{})) + c.Model = components.ModelSpec{Incoming: model, Static: model} + + rep := &components.Report{Component: "extract_llm_sweep"} + if _, err := e.Offload(sweepReqStocked(), rep, c); err != nil { + t.Fatal(err) + } + if rep.Gates["sweep_prefix_cache_read_ZERO"] != 1 || rep.Events["sweep_fallback_used"] != 1 { + t.Fatalf("the ask-then-fall-back path was not exercised (gates: %v events: %v)", + rep.Gates, rep.Events) + } + if len(rep.Calls) == 0 { + t.Fatal("no ModelCall reported") + } + // TWO model calls went out, so two must be counted. Booking the pair as one is what let the + // mean latency and the call count describe the cheap leg alone. + _, callsAfter, _ := sweepSpend(t) + if n := callsAfter - callsBefore; n != 2 { + t.Errorf("recorded %d calls, want 2 (the prefix ask AND the fallback)", n) + } + rec := rep.Calls[0] + if rec.Strategy != "prefix_ask+fallback" { + t.Errorf("Strategy = %q, want %q", rec.Strategy, "prefix_ask+fallback") + } + // The row's tokens must be the SUM: 40 fresh + 31,000 from the fallback, 90 + 420 output. + if rec.PromptTokens != 40+31000 || rec.CompletionTokens != 90+420 { + t.Errorf("row tokens = prompt %d / completion %d, want %d/%d — the row still shows one "+ + "leg", rec.PromptTokens, rec.CompletionTokens, 40+31000, 90+420) + } +} + +// erroringModel fails the completion, after enough wall time to be measurable in whole +// milliseconds. That combination is the subject: a fallback that BURNED SECONDS AND THEN FAILED. +type erroringModel struct{ calls int64 } + +var errFallbackBroke = errors.New("fallback upstream refused the completion") + +func (m *erroringModel) Complete(_ context.Context, _ string) (string, error) { + m.calls++ + time.Sleep(8 * time.Millisecond) + return "", errFallbackBroke +} + +// REVIEW FINDING (#178, round 4). `foldFallback()` was called per site, and all three sites sit +// AFTER `if reply, err = runFallback(); err != nil { return nil, r }` — so a fallback that failed +// never reached the fold. `runFallback` has already booked the leg into metrics via `recordLeg`, so +// /stats counted the call and its seconds while the ledger row kept only the prefix ask's +// LatencyMs and a Strategy naming a leg that was no longer the only one that ran. +// +// Latency rather than dollars, because on an error the sink is empty — `recordUsageCache` is reached +// only after a successful decode on both backends and neither errors after billing. But the fallback +// is the SLOW leg by construction, so a failed one puts tens of seconds into avg_latency_ms against +// a row showing milliseconds. +// +// THE NO-ASKER PATH IS THE SEVERE ONE and it is what this test drives: there, `r.rec` does not exist +// yet when the fallback runs, so on an error the row was never built at all — Component stayed "" +// and Offload dropped the entire row on its `call.rec.Component != ""` guard. /stats was left +// reporting a call the ledger had no row for. +// +// ASSERTED AS CROSS-SURFACE AGREEMENT, deliberately. There is no marshalled surface for a ModelCall +// inside /stats — the ledger rows travel to the dash Event, not this payload — so a JSON-tag +// assertion would be testing dash's converter rather than this fix. The invariant that actually +// broke is that the two surfaces disagreed, so that is what is pinned: if /stats counts the call, +// the ledger must carry a row for it. And the assertion is on rep.Calls AFTER Offload, i.e. past +// the guard that was doing the dropping, not on the pre-guard value. +func TestAFailedFallbackStillLandsInTheLedger(t *testing.T) { + model := &erroringModel{} + _, callsBefore, _ := sweepSpend(t) + + e := newSweepSmall(t, "") + c := preExpiryCtx("foldonerror", nil, store.NewMemory(store.Options{})) // no prefix asker + c.Model = components.ModelSpec{Incoming: model, Static: model} + + rep := &components.Report{Component: "extract_llm_sweep"} + if _, err := e.Offload(sweepReqStocked(), rep, c); err != nil { + // Fail-open: a broken fallback must not surface an error to the pipeline. + t.Fatalf("Offload returned an error; the component must fail open: %v", err) + } + + // PRECONDITIONS. Each one is a way this test could otherwise pass on a fixture that simply + // declined, which is the shape that makes an assertion vacuous. + if rep.Gates["sweep_no_asker"] != 1 { + t.Fatalf("the no-asker path was not taken (gates: %v)", rep.Gates) + } + if model.calls != 1 { + t.Fatalf("the fallback model was called %d times, want 1 — nothing errored, so there is "+ + "no failed leg to account for", model.calls) + } + if rep.Gates["sweep_fallback_failed"] != 1 { + t.Fatalf("the fallback did not FAIL, so this exercises the success path instead of the "+ + "error path under test (gates: %v)", rep.Gates) + } + // And /stats booked it, which is the half that was never in doubt and is what makes the + // missing row a DISAGREEMENT rather than a symmetric omission. + _, callsAfter, _ := sweepSpend(t) + if n := callsAfter - callsBefore; n != 1 { + t.Fatalf("/stats recorded %d calls for the failed fallback, want 1", n) + } + + // THE FIX: the ledger has a row for the call /stats counted. + if len(rep.Calls) != 1 { + t.Fatalf("/stats counted a call and the ledger carries %d rows: the row is built after "+ + "the error return, so Component stayed \"\" and Offload dropped it on its "+ + "Component != \"\" guard. This is the round-4 finding.", len(rep.Calls)) + } + rec := rep.Calls[0] + if rec.Component != "extract_llm_sweep" { + t.Errorf("row Component = %q, want extract_llm_sweep — without it the caller's guard "+ + "drops the row however complete the rest of it is", rec.Component) + } + if rec.LatencyMs <= 0 { + t.Errorf("row LatencyMs = %v: the failed fallback's wall time did not reach the ledger, "+ + "so avg_latency_ms carries seconds the per-call view cannot account for", rec.LatencyMs) + } + if rec.Strategy != "fallback" { + t.Errorf("row Strategy = %q, want \"fallback\": no prefix ask happened on this route", + rec.Strategy) + } + if rec.Rejection == "" { + t.Error("row Rejection is empty on a failed fallback, so the row cannot say why it " + + "produced nothing — the distinction ModelCall.Rejection exists for") + } + // Identity fields must be real, not placeholders: a row the dashboard cannot attribute to a + // candidate size or a model is barely better than no row. + if rec.CandidateTokens <= 0 { + t.Errorf("row CandidateTokens = %d, want the inventory's real size", rec.CandidateTokens) + } +} + +// The other shape: the prefix ask succeeded, the cache read was zero, and the fallback then FAILED. +// Here r.rec already exists, so the row was never dropped — it just kept the prefix ask's latency +// while /stats carried both legs. Latency-only, and the fold has to cover it too. +func TestAFailedFallbackAfterAskingStillGrowsTheRow(t *testing.T) { + model := &erroringModel{} + asker := &fakeAsker{reply: `[{"i":0,"needed_by":"none","quote":"","verdict":"keep"}]`, cacheRead: 0} + _, callsBefore, _ := sweepSpend(t) + + e := newSweepSmall(t, "") + c := preExpiryCtx("foldonerror2", asker, store.NewMemory(store.Options{})) + c.Model = components.ModelSpec{Incoming: model, Static: model} + + rep := &components.Report{Component: "extract_llm_sweep"} + if _, err := e.Offload(sweepReqStocked(), rep, c); err != nil { + t.Fatalf("Offload must fail open: %v", err) + } + if rep.Gates["sweep_prefix_cache_read_ZERO"] != 1 || rep.Gates["sweep_fallback_failed"] != 1 { + t.Fatalf("wanted a successful ask with a zero cache read and then a FAILED fallback "+ + "(gates: %v)", rep.Gates) + } + // Two legs went out, so /stats booked two. + _, callsAfter, _ := sweepSpend(t) + if n := callsAfter - callsBefore; n != 2 { + t.Fatalf("/stats recorded %d calls, want 2 (the ask AND the failed fallback)", n) + } + if len(rep.Calls) != 1 { + t.Fatalf("want one ledger row for the adjudication, got %d", len(rep.Calls)) + } + rec := rep.Calls[0] + // The row must name both legs and carry both their seconds. The failed leg sleeps 8ms, so a row + // still showing only the ask is detectable: the fake asker returns instantly. + if rec.Strategy != "prefix_ask+fallback" { + t.Errorf("row Strategy = %q, want \"prefix_ask+fallback\" — a failed fallback still ran", + rec.Strategy) + } + if rec.LatencyMs < 8 { + t.Errorf("row LatencyMs = %v, want at least the failed fallback's ~8ms: /stats has both "+ + "legs' seconds and the per-call view must not show only the fast one", rec.LatencyMs) + } +} diff --git a/docs/components/extract_llm.md b/docs/components/extract_llm.md index 47989c22..e6ccace4 100644 --- a/docs/components/extract_llm.md +++ b/docs/components/extract_llm.md @@ -676,23 +676,49 @@ questions at a flat rate. ## Metrics -`/stats` gains an `extract` block (purely additive — every pre-existing field keeps its name, so -`deploy/harbor/*.py` keeps parsing unchanged): +`/stats` gains an `extract` block (purely additive — every pre-existing field keeps its name). + +!!! danger "These are NOT this component's figures. Read `extract.by_component`." + The block's top-level keys are the **sum across every extraction component** — this one and + [`extract_llm_sweep`](extract_llm_sweep.md), which both write the same counters. The two have + opposite economics: per-output calls on a cheap model here, one call on the request's own + frontier model there. + + This document used to present the keys below as `extract_llm`'s own, and that reading cost a + whole investigation. On a measured 45-run benchmark the block reported **101 calls at 59,009 ms + mean latency and a net value of −$1.162**, which was attributed to `extract_llm` and used to + charge it a 5,452 ms/request latency cost — while this component's own debug record showed + **zero surviving candidates on all 374 requests**, i.e. not one call. The 101 were very nearly + the sweep's 96 asks. Any per-component cost, latency or call claim must come from + `extract.by_component.extract_llm`. + + Names are frozen for `/metrics` (`cg_extract_calls_total`, `cg_extract_cost_usd`, + `cg_extract_net_value_usd`, `cg_extract_latency_ms`), which off-repo alert rules query — **not** + because the benchmark harness parses them. It does not: nothing under `deploy/` reads any key + in this block. | Field | Meaning | |---|---| -| `calls` | Extraction LLM calls made | +| `by_component` | Every field below, keyed by the component that recorded it. **The per-component-safe figures.** | +| `calls` | Extraction LLM calls made — *summed across extraction components* | | `calls_avoided` | Calls avoided by the global result cache | | `calls_suppressed` | Calls declined by the economic gate | | `cache_hit_rate` | `calls_avoided / cache_lookups` | | `prompt_cache_read_tokens` / `..._write_tokens` | Preamble caching behavior — **0 read means the breakpoint is inert** | -| `extraction_cost_usd` | What the component spent | +| `extraction_cost_usd` | What was spent. Read `cost_source` before quoting it | +| `cost_source` | Where that figure came from: `component` (each call priced itself — trust it), `host_total` (the host's process-global cheap-model spend, a superset that also carries `summarize` and `agentdiet`), `partial` (some calls unpriced, so the total is a **floor**), `unpriced` (this row made calls and priced none — nothing is known), `none` (no calls; `0` is true) | +| `unpriced_components` | On the aggregate: which components' calls priced nothing, i.e. what a `partial` or `host_total` total is **short of**. Omitted when everything priced itself | | `gross_value_usd` | What its saved tokens are worth at the rate they'd have been billed | -| **`net_value_usd`** | **The honest headline. Negative = the component is underwater.** | +| **`net_value_usd`** | **The honest headline. Negative = underwater.** `null` when the spend is not known | | `avg_latency_ms` | Mean wall time per call (latency cost on the hot path) | | `gross_saved_tokens` | Tokens removed | | `reasons` / `top_reason` | Why extraction ran or was suppressed | +Per-component, in `components.extract_llm`: **`acted` counts free replays.** A frozen decision +re-spliced on a later turn saves tokens and costs nothing, and it landed in the same counter as the +call that derived it — `acted: 239` beside `reapplied_same_session: 2,291` was read as 239 paid +extractions. Use `acted_fresh` (paid work) and `acted_replay` (free) instead. + Plus, at the top level of `/stats`: **`llm_truncated`** — replies that stopped at the model's output cap. That is the worst outcome available, full price for zero result, and it used to be invisible because a truncated program parses as nothing, exactly like a model that declined to diff --git a/docs/reference/routes.md b/docs/reference/routes.md index 0ea5485c..cead323f 100644 --- a/docs/reference/routes.md +++ b/docs/reference/routes.md @@ -57,8 +57,11 @@ either, an unknown path still 404s exactly as before. ## `GET /stats` -Fields are only ever **added** to this payload (harnesses in `deploy/harbor` parse it), so a -consumer that reads by key keeps working. The tables below group the `Snapshot` struct in +Fields are only ever **added** to this payload, so a consumer that reads by key keeps working. +Two different consumers make that a hard rule: `deploy/harbor/*.py` reads `runs`, `acted` and +`saved_tokens` off the per-component objects (`measure.py` by direct index, so a removal is a +`KeyError`), and `/metrics` re-publishes much of the rest as `cg_*` series that off-repo alert +rules and dashboards are written against — where a rename breaks monitoring *silently*. The tables below group the `Snapshot` struct in `metrics/metrics.go`; that struct is the authority, and it grows. ### Savings @@ -81,7 +84,9 @@ Per-component (`components.` and `potential_components.`): | Field | Meaning | |---|---| | `runs` | Times the component ran. | -| `acted` | Runs that actually saved tokens. | +| `acted` | Runs that actually saved tokens. **Includes free replays** — see the two fields below before reading this as work the component paid for. | +| `acted_fresh` | Runs that saved tokens by doing new work. Every deterministic component's `acted` is entirely this. | +| `acted_replay` | Runs whose whole saving came from **replaying a decision already frozen** — the same bytes re-spliced, no model call, no spend. `acted_fresh + acted_replay == acted`. A component with `acted: 239` and `acted_replay: 239` made no calls at all; one with the same `acted` and `acted_replay: 0` paid for every one. | | `mutated` | Runs that changed the request at all — may save 0 content tokens. | | `reverted` | Runs the pipeline rolled back (error, panic, or grew the request). | | `saved_tokens` | **Cumulative** — re-counted every turn the compaction re-appears. | @@ -91,6 +96,26 @@ Per-component (`components.` and `potential_components.`): | `discarded_changes` | Changes the writeback layer threw away, attributed back to this component. | | `gates` | Rejection histogram, gate name → candidates that gate declined. Omitted when empty. It is what turns `acted: 0` into a diagnosis: the component saw no candidates, or saw them and a named guard refused. | +### Extraction economics (`extract`) + +Present only when an extraction component has recorded something. See +[`extract_llm`](../components/extract_llm.md) for the field meanings. + +!!! warning "The `extract` block is a SUM across components, not one component's figures" + `extract_llm` and `extract_llm_sweep` both write these counters, and the enclosing block is + their **total**. It reads like one component's numbers and is not: a measured run attributed + 101 calls at 59,009 ms and a net value of −$1.162 to `extract_llm` when that component had + made no call at all and the figures were the sweep's. **Read `extract.by_component`** for any + per-component cost, latency or call claim. The enclosing keys are kept for `/metrics` + compatibility (`cg_extract_*`), not because they are the figure to quote. + +| Field | Meaning | +|---|---| +| `by_component` | The same fields again, keyed by the component that recorded them. The only per-component-safe figures here. Omitted when nothing recorded. | +| `cost_source` | Where `extraction_cost_usd` came from, because `$0` and *no evidence* are the same number and the opposite claim. `component` = every call priced itself at the rates of the model it called (trust it). `host_total` = the host's process-global cheap-model spend, a **superset** that also carries `summarize` and `agentdiet` and prices everything through one card. `partial` = some calls priced themselves and some did not, so the total is a **floor**. `unpriced` = this row made calls and priced none, so nothing is known about what it spent. `none` = no calls, no spend; `0` is true. | +| `net_value_usd` | `null` when the spend behind it is not known (`cost_source: unpriced`). The aggregate is never null. | +| `unpriced_components` | Which components' calls priced nothing, so `partial` and `host_total` say what the total is **short of** rather than only that it is incomplete. Aggregate only; omitted when every call priced itself. | + !!! warning "Cumulative is not unique" `saved_tokens` counts the same compaction again on every later turn that carries it. A figure like "4.8M tokens saved" is a *cumulative* total; the unique figures behind the diff --git a/metrics/extract.go b/metrics/extract.go index a663e602..cc9bff8f 100644 --- a/metrics/extract.go +++ b/metrics/extract.go @@ -17,15 +17,47 @@ import ( // activation because an operator's first question about an expensive component is always // "why did this run?". // -// These are process-global counters, matching cheapmodel.Usage's existing scope. -var ( - xCalls atomic.Int64 // extraction LLM calls actually made - xCacheHits atomic.Int64 // calls avoided by the global result cache - xSuppressed atomic.Int64 // calls suppressed by the economic gate - xGrossSaved atomic.Int64 // tokens removed (unique, first application only) - xLatencyMs atomic.Int64 // cumulative wall time in extraction calls - xLookups atomic.Int64 // result-cache lookups (hits + misses), for the hit rate - // xValueNano is the realized dollar value of what was removed, in nanodollars, recorded +// SCOPED PER COMPONENT (issue #176). These were process-global counters, and the comment +// here said so as a design statement — matching cheapmodel.Usage's scope, on the premise +// that "the LLM component in a config is extract". That premise stopped being true when +// the cold-transcript sweep became its own component: extract_llm and extract_llm_sweep +// both write these counters, so the `extract` block in /stats was the SUM of two +// components with opposite economics — one call on the request's own frontier model +// against per-output calls on a cheap one — presented under a name that reads as one of +// them. MEASURED (iteration 023, arm B): `calls: 101` and `avg_latency_ms: 59,009` were +// attributed to extract_llm while its own debug record reported `cands: 0` on all 374 +// requests, i.e. it made no call at all; the 101 were very nearly the sweep's 96 asks. +// A latency and a net value were charged to the wrong component and an experiment's +// conclusion was drawn from it. +// +// So every accessor and every recorder is keyed by component name now, and the aggregate +// is DERIVED by summing them rather than maintained alongside them — two totals kept in +// parallel is how the second one drifts from the first. +// +// The latency accessors are keyed too, and that is a behaviour change, deliberately: the +// exploration brake in offload.tooSlowToExplore read the global p50, so the sweep's +// ~59-second asks braked extract_llm's exploration on evidence from a different component +// and a different model. A brake must read the latency of the calls it is braking. + +// CostSource values for ExtractStats.CostSource. Constants because they are read off /stats by +// operators and by promexport; a typo in one of two string literals is a silently wrong label. +const ( + costSourceComponent = "component" + costSourceHost = "host_total" + costSourcePartial = "partial" + costSourceUnpriced = "unpriced" + costSourceNone = "none" +) + +// xCounters is one component's extraction accounting. +type xCounters struct { + calls atomic.Int64 // extraction LLM calls actually made + cacheHits atomic.Int64 // calls avoided by the global result cache + suppressed atomic.Int64 // calls suppressed by the economic gate + grossSaved atomic.Int64 // tokens removed (unique, first application only) + latencyMs atomic.Int64 // cumulative wall time in extraction calls + lookups atomic.Int64 // result-cache lookups (hits + misses), for the hit rate + // valueNano is the realized dollar value of what was removed, in nanodollars, recorded // BY THE COMPONENT at the rate each removal was actually worth. // // It exists because the alternative — tokens x a constant rate chosen from the cache MODE @@ -39,10 +71,21 @@ var ( // // Nanodollars so the accumulator can stay atomic; a call's value is ~1e-5 USD, so an // int64 of nanodollars holds ~9e9 USD of headroom. - xValueNano atomic.Int64 + valueNano atomic.Int64 + // spendNano is what THIS component paid, summed from each call's own ModelCall.CostUSD — + // which each component prices with the rates of the model it actually called. + // + // The alternative, and what /stats did before #176, was cheapmodel's process-global token + // totals priced through one rate card. That figure is not this component's spend and is + // not even extraction's: every cheap-model call in the process lands in it, `summarize` + // and `agentdiet` included, and the sweep's calls go to the request's own frontier model + // while the card is haiku's. So it over-attributed non-extraction spend to extraction and + // mispriced the half of extraction that does not use the cheap model. Recorded per call + // here, at the price the component itself computed. + spendNano atomic.Int64 - xReasonMu sync.Mutex - xReasons = map[string]int64{} // trigger/suppression reason -> count + reasonMu sync.Mutex + reasons map[string]int64 // trigger/suppression reason -> count // A ring of recent per-call latencies, for the MEDIAN. The mean cannot answer "are // calls slow?" on this workload: measured n=8 on one gateway, p50 3,748 ms against a @@ -50,90 +93,147 @@ var ( // queue time, so one tail sample moves the mean past a brake the typical call is // nowhere near. A ring rather than a histogram because the only consumer is one // threshold comparison and 64 samples is already more evidence than the gate needs. - xLatMu sync.Mutex - xLatRing [64]float64 - xLatN int + latMu sync.Mutex + latRing [64]float64 + latN int +} + +// xReg holds one xCounters per component name. A map under a mutex rather than a sync.Map +// because the write path is one pointer lookup per recorded event and the read path is +// /stats; neither is hot enough to want the extra indirection. +var ( + xRegMu sync.RWMutex + xReg = map[string]*xCounters{} ) -// latencyP50 returns the median of the retained samples, and how many there are. -func latencyP50() (float64, int) { - xLatMu.Lock() - defer xLatMu.Unlock() - n := xLatN - if n > len(xLatRing) { - n = len(xLatRing) +// xFor returns (creating if needed) the counters for one component. An empty name is +// accepted and bucketed under "" rather than dropped: losing an event is worse than +// showing it under an unhelpful key, and the key itself then names the caller to fix. +func xFor(component string) *xCounters { + xRegMu.RLock() + c := xReg[component] + xRegMu.RUnlock() + if c != nil { + return c } - if n == 0 { - return 0, 0 + xRegMu.Lock() + defer xRegMu.Unlock() + if c = xReg[component]; c == nil { + c = &xCounters{} + xReg[component] = c + } + return c +} + +// xComponents returns the registered component names, sorted, so snapshots are stable. +func xComponents() []string { + xRegMu.RLock() + defer xRegMu.RUnlock() + names := make([]string, 0, len(xReg)) + for k := range xReg { + names = append(names, k) + } + sort.Strings(names) + return names +} + +// latencySamples copies the retained latency ring. +func (x *xCounters) latencySamples() []float64 { + x.latMu.Lock() + defer x.latMu.Unlock() + n := x.latN + if n > len(x.latRing) { + n = len(x.latRing) } cp := make([]float64, n) - copy(cp, xLatRing[:n]) - sort.Float64s(cp) - return cp[n/2], n + copy(cp, x.latRing[:n]) + return cp +} + +// latencyP50 returns the median of the retained samples, and how many there are. +func latencyP50(samples []float64) (float64, int) { + if len(samples) == 0 { + return 0, 0 + } + sort.Float64s(samples) + return samples[len(samples)/2], len(samples) } -// RecordExtractionCall notes one extraction LLM call and its wall time. -func RecordExtractionCall(latencyMs float64) { - xCalls.Add(1) - xLatencyMs.Add(int64(latencyMs)) - xLatMu.Lock() - xLatRing[xLatN%len(xLatRing)] = latencyMs - xLatN++ - xLatMu.Unlock() +// RecordExtractionCall notes one extraction LLM call and its wall time, for `component`. +func RecordExtractionCall(component string, latencyMs float64) { + x := xFor(component) + x.calls.Add(1) + x.latencyMs.Add(int64(latencyMs)) + x.latMu.Lock() + x.latRing[x.latN%len(x.latRing)] = latencyMs + x.latN++ + x.latMu.Unlock() } -// ExtractionP50LatencyMs returns the MEDIAN observed wall time per extraction call and the -// number of samples behind it. The gate reads this rather than the mean to decide whether -// speculative calls have become too slow to be worth their wall clock — see -// offload.tooSlowToExplore for the measurement that made the mean unusable here. -func ExtractionP50LatencyMs() (float64, int64) { - p50, n := latencyP50() +// ExtractionP50LatencyMs returns the MEDIAN observed wall time per extraction call made by +// `component`, and the number of samples behind it. The gate reads this rather than the mean +// to decide whether speculative calls have become too slow to be worth their wall clock — +// see offload.tooSlowToExplore for the measurement that made the mean unusable here. +// +// PER COMPONENT (#176): reading the process-global median made one component's brake fire on +// another component's latency. extract_llm's per-output cheap-model calls and the sweep's +// single frontier-model ask differ by more than an order of magnitude, so the pooled median +// is not a fact about either. +func ExtractionP50LatencyMs(component string) (float64, int64) { + p50, n := latencyP50(xFor(component).latencySamples()) return p50, int64(n) } // RecordExtractionCacheLookup notes one global result-cache lookup and whether it hit. // A hit is a call AVOIDED — the cheapest possible outcome, and the source of ~93% of the // component's realized value in the Terminal-Bench measurement. -func RecordExtractionCacheLookup(hit bool) { - xLookups.Add(1) +func RecordExtractionCacheLookup(component string, hit bool) { + x := xFor(component) + x.lookups.Add(1) if hit { - xCacheHits.Add(1) + x.cacheHits.Add(1) } } -// ExtractionAvgLatencyMs returns the observed mean wall time per extraction call and the -// number of calls it averages. The gate reads this to stop SPECULATIVE calls once they are -// observed to be slow — exploration spends wall clock as well as money, and an agent with a -// task deadline feels the former more (PR #37: 17.8s across 2 calls that saved 0 tokens). -func ExtractionAvgLatencyMs() (float64, int64) { - calls := xCalls.Load() +// ExtractionAvgLatencyMs returns the observed mean wall time per extraction call made by +// `component` and the number of calls it averages. The gate reads this to stop SPECULATIVE +// calls once they are observed to be slow — exploration spends wall clock as well as money, +// and an agent with a task deadline feels the former more (PR #37: 17.8s across 2 calls that +// saved 0 tokens). +func ExtractionAvgLatencyMs(component string) (float64, int64) { + x := xFor(component) + calls := x.calls.Load() if calls == 0 { return 0, 0 } - return float64(xLatencyMs.Load()) / float64(calls), calls + return float64(x.latencyMs.Load()) / float64(calls), calls } // RecordExtractionSuppressed notes that the economic gate declined a call, with its reason. -func RecordExtractionSuppressed(reason string) { - xSuppressed.Add(1) - RecordExtractionReason(reason) +func RecordExtractionSuppressed(component, reason string) { + xFor(component).suppressed.Add(1) + RecordExtractionReason(component, reason) } // RecordExtractionReason counts one trigger/suppression reason. -func RecordExtractionReason(reason string) { +func RecordExtractionReason(component, reason string) { if reason == "" { return } - xReasonMu.Lock() - xReasons[reason]++ - xReasonMu.Unlock() + x := xFor(component) + x.reasonMu.Lock() + if x.reasons == nil { + x.reasons = map[string]int64{} + } + x.reasons[reason]++ + x.reasonMu.Unlock() } // RecordExtractionSaving notes tokens removed by an accepted extraction (count each // distinct compaction once — the caller dedups by content key). -func RecordExtractionSaving(tokens int) { +func RecordExtractionSaving(component string, tokens int) { if tokens > 0 { - xGrossSaved.Add(int64(tokens)) + xFor(component).grossSaved.Add(int64(tokens)) } } @@ -143,17 +243,38 @@ func RecordExtractionSaving(tokens int) { // // Called at BOTH sites deliberately: crediting only the first application under-reports by // however much the replay is worth, and crediting the replays at the first application's rate -// over-reports by 12.5x. The component is the only layer that knows which regime a request +// over-reports it by 12.5x. The component is the only layer that knows which regime a request // was in, so it is the layer that prices it. -func RecordExtractionValue(usd float64) { +func RecordExtractionValue(component string, usd float64) { if usd > 0 { - xValueNano.Add(int64(usd * 1e9)) + xFor(component).valueNano.Add(int64(usd * 1e9)) } } -// ExtractStats is the extraction economics block served inside /stats. It is ADDITIVE: -// every pre-existing /stats field keeps its name and meaning, because deploy/harbor/*.py -// parses them. +// RecordExtractionSpend notes what ONE call cost, priced by the component that made it with +// the rates of the model it actually addressed. See xCounters.spendNano for why /stats cannot +// derive this from cheapmodel's process totals. +func RecordExtractionSpend(component string, usd float64) { + if usd > 0 { + xFor(component).spendNano.Add(int64(usd * 1e9)) + } +} + +// ExtractStats is the extraction economics block served inside /stats. It is ADDITIVE: every +// pre-existing field keeps its name and meaning. +// +// NOT because the benchmark harness parses it — that reason was carried here and in +// docs/components/extract_llm.md and it is FALSE. `deploy/` reads `runs`, `acted` and +// `saved_tokens` off the per-COMPONENT objects (measure.py does so by direct index, so removing +// one is a KeyError), and reads nothing at all out of this block; grepping deploy/ for +// gross_saved_tokens, calls_avoided, extraction_cost_usd or avg_latency_ms returns nothing. +// +// The real reason these names are frozen is /metrics. proxy/promexport.go publishes them as +// cg_extract_calls_total, cg_extract_cost_usd, cg_extract_net_value_usd and cg_extract_latency_ms, +// and dash/metrics_export.go documents cg_extract_net_value_usd as the endpoint's only dollar +// figure. Those are what an operator's alert rule and dashboard query are written against, off-repo +// and unversioned, so a rename breaks monitoring silently — which is a worse failure than breaking +// the harness, because the harness would at least crash. type ExtractStats struct { Calls int64 `json:"calls"` // extraction LLM calls made CallsAvoided int64 `json:"calls_avoided"` // global result-cache hits @@ -178,20 +299,68 @@ type ExtractStats struct { // the honest headline. Negative means the component is underwater and should be off. ExtractionCostUSD float64 `json:"extraction_cost_usd"` GrossValueUSD float64 `json:"gross_value_usd"` - NetValueUSD float64 `json:"net_value_usd"` + // NetValueUSD is nil — rendered as JSON `null` — when the spend behind it is NOT KNOWN. + // + // A pointer rather than a float because 0 dollars of spend and 0 evidence of spend are the + // same number and the opposite claim, and this field is the one an operator acts on. A + // component that made calls and priced none of them would otherwise publish + // `net_value_usd: +grossValue` — "comfortably profitable" — on no cost information at all, + // while the block enclosing it reported the component underwater from the host's figure. Two + // figures for one quantity disagreeing by the whole spend is exactly the failure this change + // exists to remove, so the unknown is rendered as unknown. Read CostSource for which case a + // row is in. + // + // The AGGREGATE is never nil: it always has a determined spend, from the components, from the + // host's figure, or from there being no spend at all. + NetValueUSD *float64 `json:"net_value_usd"` + // CostSource says where ExtractionCostUSD came from. Named values, because the number alone + // cannot distinguish the cases and they call for different responses: + // + // component every call in this row priced itself, at the rates of the model it called. + // The figure is this component's own arithmetic — trust it. + // host_total the host's process-global cheap-model spend. A SUPERSET: it also carries + // `summarize` and `agentdiet`, and prices everything through one rate card. + // Aggregate only. + // partial some calls priced themselves and some did not, and the host's figure was no + // larger. The total is a FLOOR, not the bill. Aggregate only. + // unpriced this row made calls and none of them priced itself, so nothing is known about + // what it spent. ExtractionCostUSD is 0 because there is no evidence, NOT + // because the component was free, and NetValueUSD is null. + // none no calls and no spend. 0 is the true figure. + CostSource string `json:"cost_source"` + // UnpricedComponents NAMES the components whose calls priced nothing, sorted, so a + // `cost_source` of `partial` or `host_total` says WHICH component the total is missing + // rather than only that something is missing. + // + // Aggregate only, and omitted when every call priced itself. Without it the label is a + // prompt to go and scan every by_component row for `cost_source: unpriced` — which is work + // the snapshot already did, since it is the same test that set the aggregate's label. An + // operator who has to re-derive the answer from the rows will read the label and stop, and + // then the floor gets quoted as the bill. + UnpricedComponents []string `json:"unpriced_components,omitempty"` // Reasons counts why extraction ran or was suppressed, most frequent first. Reasons map[string]int64 `json:"reasons,omitempty"` // TopReason is the single most common reason — the one-line operator answer. TopReason string `json:"top_reason,omitempty"` + + // ByComponent breaks every field above down by the component that recorded it (#176). + // + // The enclosing block stays the SUM — see the type comment for why those names are frozen, + // and note it is /metrics rather than the harness that freezes them — but the sum is the + // figure that misattributed a 59-second latency and a negative net value to a component + // that made no calls, so it must never again be the only figure available. Nil inside each + // nested entry: the breakdown does not nest. + ByComponent map[string]*ExtractStats `json:"by_component,omitempty"` } // ExtractSnapshot builds the extraction stats. // -// cost is the component's own LLM spend. perSavedTokenUSD is the value of ONE saved token -// at the rate it would actually have been billed (cache-read vs fresh — the caller knows -// the traffic's cache-awareness); the value side is computed HERE, against this -// component's own GrossSavedTokens. +// cost is the host's fallback figure for extraction spend, used only for components that +// recorded none of their own (a library embedding that never fills ModelCall.CostUSD). +// perSavedTokenUSD is the value of ONE saved token at the rate it would actually have been +// billed (cache-read vs fresh — the caller knows the traffic's cache-awareness); the value +// side is computed HERE, against each component's own GrossSavedTokens. // // Taking a RATE rather than a pre-computed total is deliberate. The obvious signature // (grossValue float64) invites the caller to pass the pipeline-wide savings figure, which @@ -200,50 +369,187 @@ type ExtractStats struct { // otherwise. That is the single number this whole issue exists to get right, so the // signature makes the mistake impossible to express. func ExtractSnapshot(cost, perSavedTokenUSD float64, cacheWrite, cacheRead int64) ExtractStats { - calls := xCalls.Load() - lookups := xLookups.Load() - hits := xCacheHits.Load() - gross := xGrossSaved.Load() + names := xComponents() + total := ExtractStats{ + PromptCacheReadTokens: cacheRead, PromptCacheWriteTokens: cacheWrite, + } + var ( + perComp = make(map[string]*ExtractStats, len(names)) + grossVal float64 + spend float64 + totLatMs int64 + totReason = map[string]int64{} + anySpend bool + // unpricedCalls counts calls made by components that priced NOTHING. It is what stops the + // aggregate silently under-reporting: `anySpend` alone is one boolean across every + // component, so if extract_llm priced its calls (the cheap card is never zero) while + // extract_llm_sweep recorded none, the total became extract_llm's spend alone and the + // sweep's real dollars vanished — no fallback, no warning, a smaller number than before. + unpricedCalls int64 + // unpriced names them. `names` is sorted, so this is too, with no second sort. + unpriced []string + ) + for _, name := range names { + x := xFor(name) + s, gv, sp, recorded, touched, reasons := x.snapshot(perSavedTokenUSD) + if !touched { + continue // an entry created only by a counter LOOKUP; it recorded nothing + } + cp := s + perComp[name] = &cp + total.Calls += s.Calls + total.CallsAvoided += s.CallsAvoided + total.CallsSuppressed += s.CallsSuppressed + total.CacheLookups += s.CacheLookups + total.GrossSavedTokens += s.GrossSavedTokens + grossVal += gv + spend += sp + anySpend = anySpend || recorded + if !recorded && s.Calls > 0 { + unpricedCalls += s.Calls + unpriced = append(unpriced, name) + } + totLatMs += x.latencyMs.Load() + for k, v := range reasons { + totReason[k] += v + } + } + // WHICH SPEND FIGURE THE TOTAL PUBLISHES, and it must say which one it chose. + // + // The components' own priced spend is the figure to prefer: each knows the model it called and + // its rates. The host's `cost` is cheapmodel's process-global total priced through one card — + // a SUPERSET (it carries `summarize` and `agentdiet` too) and mispriced for any component that + // does not call the cheap model. So it is a fallback, never an equal. + switch { + case !anySpend && cost > 0: + // Nothing priced itself. A library embedding, /compact, a host that never fills + // ModelCall.CostUSD — the host's figure is all the information there is. + spend, total.CostSource = cost, costSourceHost + case unpricedCalls > 0: + // Some calls priced themselves and some did not, so the recorded sum is a FLOOR. Publish + // the larger of the floor and the host's superset figure, and name which one it is: a + // total that quietly omits real dollars is the defect, not the loud one. + total.CostSource = costSourcePartial + if cost > spend { + spend, total.CostSource = cost, costSourceHost + } + case anySpend: + total.CostSource = costSourceComponent + default: + total.CostSource = costSourceNone + } + // Published whenever anything is unpriced, whichever branch above ran: under `partial` it + // names what the floor is short of, and under `host_total` it names what forced the fallback. + total.UnpricedComponents = unpriced + total.ExtractionCostUSD = round4(spend) + total.GrossValueUSD = round4(grossVal) + net := round4(grossVal - spend) + total.NetValueUSD = &net + if total.CacheLookups > 0 { + total.CacheHitRate = float64(total.CallsAvoided) / float64(total.CacheLookups) + } + if total.Calls > 0 { + total.AvgLatencyMs = float64(totLatMs) / float64(total.Calls) + } + total.Reasons, total.TopReason = sortedReasons(totReason) + if len(perComp) > 0 { + total.ByComponent = perComp + } + return total +} + +// snapshot renders one component's counters, plus the raw pieces the aggregate needs (its +// gross value and spend in dollars, whether it priced its own spend, whether it recorded +// anything at all, and its reason histogram) — returned rather than re-read, so the total and +// the per-component row are computed from ONE read of each atomic. +func (x *xCounters) snapshot(perSavedTokenUSD float64) ( + s ExtractStats, grossValue, spend float64, spendRecorded, touched bool, reasons map[string]int64, +) { + calls := x.calls.Load() + lookups := x.lookups.Load() + hits := x.cacheHits.Load() + gross := x.grossSaved.Load() // The component's own realized valuation wins where it has one: it knows each removal's // regime and counts the replays. perSavedTokenUSD stays the fallback for a host that // records nothing (library users, /compact), where one flat rate is all there is. - grossValue := float64(gross) * perSavedTokenUSD - if v := xValueNano.Load(); v > 0 { + grossValue = float64(gross) * perSavedTokenUSD + if v := x.valueNano.Load(); v > 0 { grossValue = float64(v) / 1e9 } - - s := ExtractStats{ - Calls: calls, CallsAvoided: hits, CallsSuppressed: xSuppressed.Load(), + if sp := x.spendNano.Load(); sp > 0 { + spend, spendRecorded = float64(sp)/1e9, true + } + s = ExtractStats{ + Calls: calls, CallsAvoided: hits, CallsSuppressed: x.suppressed.Load(), CacheLookups: lookups, GrossSavedTokens: gross, - PromptCacheReadTokens: cacheRead, PromptCacheWriteTokens: cacheWrite, - ExtractionCostUSD: round4(cost), GrossValueUSD: round4(grossValue), - NetValueUSD: round4(grossValue - cost), + ExtractionCostUSD: round4(spend), GrossValueUSD: round4(grossValue), + } + // A ROW NEVER BORROWS THE HOST'S FIGURE. `cost` is one process-global number covering every + // cheap-model call including components that are not extraction at all; there is no sound way + // to split it across rows, and splitting it by call count would be an invented number + // presented at the same precision as a measured one. So a row that priced nothing says so and + // leaves the net UNKNOWN, rather than publishing +grossValue on no cost evidence. + switch { + case spendRecorded: + s.CostSource = costSourceComponent + net := round4(grossValue - spend) + s.NetValueUSD = &net + case calls > 0: + s.CostSource = costSourceUnpriced // NetValueUSD stays nil -> JSON null + default: + // No calls, so nothing was spent and 0 is the true figure — a replay-only component's + // row, which is genuinely and provably free. + s.CostSource = costSourceNone + net := round4(grossValue) + s.NetValueUSD = &net } if lookups > 0 { s.CacheHitRate = float64(hits) / float64(lookups) } if calls > 0 { - s.AvgLatencyMs = float64(xLatencyMs.Load()) / float64(calls) + s.AvgLatencyMs = float64(x.latencyMs.Load()) / float64(calls) + } + x.reasonMu.Lock() + reasons = make(map[string]int64, len(x.reasons)) + for k, v := range x.reasons { + reasons[k] = v } + x.reasonMu.Unlock() + s.Reasons, s.TopReason = sortedReasons(reasons) + touched = calls > 0 || lookups > 0 || gross > 0 || s.CallsSuppressed > 0 || + len(reasons) > 0 || x.valueNano.Load() > 0 || x.spendNano.Load() > 0 + return s, grossValue, spend, spendRecorded, touched, reasons +} - xReasonMu.Lock() - if len(xReasons) > 0 { - s.Reasons = make(map[string]int64, len(xReasons)) - keys := make([]string, 0, len(xReasons)) - for k, v := range xReasons { - s.Reasons[k] = v - keys = append(keys, k) +// sortedReasons copies a reason histogram and names its mode. Nil map in, nil map out, so the +// omitempty on Reasons still elides the field. +func sortedReasons(in map[string]int64) (map[string]int64, string) { + if len(in) == 0 { + return nil, "" + } + out := make(map[string]int64, len(in)) + keys := make([]string, 0, len(in)) + for k, v := range in { + out[k] = v + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if in[keys[i]] != in[keys[j]] { + return in[keys[i]] > in[keys[j]] } - sort.Slice(keys, func(i, j int) bool { - if xReasons[keys[i]] != xReasons[keys[j]] { - return xReasons[keys[i]] > xReasons[keys[j]] - } - return keys[i] < keys[j] // stable output for equal counts - }) - s.TopReason = keys[0] + return keys[i] < keys[j] // stable output for equal counts + }) + return out, keys[0] +} + +// Net returns the net value, and whether it is known. Consumers that must publish a number (a +// Prometheus gauge cannot say "unknown") use the bool to decide whether to publish at all rather +// than turning an unknown into a 0 that reads as break-even. +func (s ExtractStats) Net() (float64, bool) { + if s.NetValueUSD == nil { + return 0, false } - xReasonMu.Unlock() - return s + return *s.NetValueUSD, true } func round4(f float64) float64 { diff --git a/metrics/extract_scope_test.go b/metrics/extract_scope_test.go new file mode 100644 index 00000000..bcbc09c3 --- /dev/null +++ b/metrics/extract_scope_test.go @@ -0,0 +1,368 @@ +package metrics + +import ( + "encoding/json" + "testing" + + "github.com/rossoctl/context-guru/components" +) + +// REGRESSION (#176). The extraction counters were process-global, so the `extract` block in +// /stats was the SUM of extract_llm and extract_llm_sweep under a name that reads as one of +// them. MEASURED (iteration 023, arm B): 101 `calls` at 59,009 ms mean latency and a net value +// of -$1.162 were read as extract_llm's, while extract_llm's own debug record reported zero +// surviving candidates on all 374 requests and the sweep had made 96 asks. A per-component +// latency and cost claim from a pipeline containing both was unsafe, and an experiment's +// conclusion was drawn from one. +// +// The reproduction below is that shape in miniature: the sweep makes the expensive calls, the +// tail pass makes none and only replays. If the two are pooled, extract_llm shows the sweep's +// calls and the sweep's latency. +func TestExtractStatsAreScopedPerComponent(t *testing.T) { + resetExtract() + // extract_llm: no calls at all. Only frozen replays, which cost nothing and are worth + // something, so it records value without ever recording a call. + RecordExtractionValue("extract_llm", 0.004) + RecordExtractionCacheLookup("extract_llm", true) + // extract_llm_sweep: the expensive component. Two very slow asks. + RecordExtractionCall("extract_llm_sweep", 58_000) + RecordExtractionCall("extract_llm_sweep", 60_018) + RecordExtractionSpend("extract_llm_sweep", 1.166) + RecordExtractionSaving("extract_llm_sweep", 900) + + s := ExtractSnapshot(0, 0.30/1e6, 0, 0) + if s.ByComponent == nil { + t.Fatal("no per-component breakdown: the pooled figure is the only one available again") + } + tail, ok := s.ByComponent["extract_llm"] + if !ok { + t.Fatal("extract_llm missing from the breakdown") + } + sweep, ok := s.ByComponent["extract_llm_sweep"] + if !ok { + t.Fatal("extract_llm_sweep missing from the breakdown") + } + // THE DEFECT: extract_llm made no call, so nothing may credit it with one, and its mean + // latency is not 59 seconds — it has no latency at all. + if tail.Calls != 0 { + t.Errorf("extract_llm credited with %d calls; it made none (this is #176)", tail.Calls) + } + if tail.AvgLatencyMs != 0 { + t.Errorf("extract_llm avg_latency_ms = %v; it made no calls to be slow", + tail.AvgLatencyMs) + } + if sweep.Calls != 2 { + t.Errorf("extract_llm_sweep calls = %d, want 2", sweep.Calls) + } + if sweep.AvgLatencyMs != 59_009 { + t.Errorf("extract_llm_sweep avg_latency_ms = %v, want 59009", sweep.AvgLatencyMs) + } + // And the money lands on the component that spent it. extract_llm is net POSITIVE (free + // replays), the sweep net NEGATIVE — the pooled block reported one negative figure and + // attributed it to the wrong one. + if net, known := tail.Net(); !known || net <= 0 { + t.Errorf("extract_llm net = %v (known=%v); it made no call, so its free replays worth "+ + "$0.004 are provably profitable and the figure must be known", net, known) + } + if net, known := sweep.Net(); !known || net >= 0 { + t.Errorf("extract_llm_sweep net = %v (known=%v); it spent $1.166 to save 900 tokens", + net, known) + } + // And the rows must say WHERE their cost came from, so "$0" and "no evidence" are not the + // same reading. extract_llm made no calls, so its 0 is provable; the sweep priced its own. + if tail.CostSource != costSourceNone { + t.Errorf("extract_llm cost_source = %q, want %q", tail.CostSource, costSourceNone) + } + if sweep.CostSource != costSourceComponent { + t.Errorf("extract_llm_sweep cost_source = %q, want %q", + sweep.CostSource, costSourceComponent) + } + // The enclosing block stays the SUM — deploy/harbor/*.py parses it — so the fix must be + // additive, not a re-scoping of the existing keys. + if s.Calls != 2 { + t.Errorf("pooled calls = %d, want 2 (the block stays the sum for the harness)", s.Calls) + } +} + +// The latency BRAKE reads these counters to decide whether speculative calls are still worth +// their wall clock (offload.tooSlowToExplore). Pooled, the sweep's ~59-second frontier-model +// asks braked extract_llm's cheap-model exploration on evidence from a different component and +// a different model — a decision path, not just a display. +func TestLatencyAccessorsAreScopedPerComponent(t *testing.T) { + resetExtract() + RecordExtractionCall("extract_llm_sweep", 59_000) + RecordExtractionCall("extract_llm_sweep", 59_000) + RecordExtractionCall("extract_llm", 900) + + if p50, n := ExtractionP50LatencyMs("extract_llm"); p50 != 900 || n != 1 { + t.Errorf("extract_llm p50 = (%v,%d), want (900,1) — it must not see the sweep's asks", + p50, n) + } + if avg, n := ExtractionAvgLatencyMs("extract_llm"); avg != 900 || n != 1 { + t.Errorf("extract_llm mean = (%v,%d), want (900,1)", avg, n) + } + if p50, n := ExtractionP50LatencyMs("extract_llm_sweep"); p50 != 59_000 || n != 2 { + t.Errorf("extract_llm_sweep p50 = (%v,%d), want (59000,2)", p50, n) + } +} + +// Extraction spend must come from the component that priced it, not from cheapmodel's +// process-global token totals through one rate card. Those totals include `summarize` and +// `agentdiet`, and price the sweep's frontier-model asks at the cheap model's rates. +func TestRecordedSpendBeatsTheHostsGlobalFigure(t *testing.T) { + resetExtract() + RecordExtractionSpend("extract_llm", 0.02) + RecordExtractionSpend("extract_llm_sweep", 0.30) + // The host offers $9.99 — every cheap-model call in the process, extraction or not. + s := ExtractSnapshot(9.99, 0.30/1e6, 0, 0) + if s.ExtractionCostUSD != 0.32 { + t.Errorf("extraction_cost_usd = %v, want 0.32 (the components' own priced spend)", + s.ExtractionCostUSD) + } + if s.ByComponent["extract_llm"].ExtractionCostUSD != 0.02 { + t.Errorf("extract_llm cost = %v, want 0.02", + s.ByComponent["extract_llm"].ExtractionCostUSD) + } + if s.CostSource != costSourceComponent { + t.Errorf("cost_source = %q, want %q", s.CostSource, costSourceComponent) + } +} + +// REVIEW FINDING (#178). The host's figure is one process-global number covering every cheap-model +// call in the process — `summarize` and `agentdiet` included — so it cannot be split across rows. +// Applying it to the AGGREGATE ONLY produced two figures for one quantity that disagreed by the +// whole spend: the block said $9.99 while its only by_component row said $0, and an operator +// reading the row saw a comfortably positive net value on no cost evidence at all. +// +// The contract now: a row that priced nothing says so (`cost_source: unpriced`) and leaves +// net_value_usd NULL rather than publishing +grossValue. 0 dollars and 0 evidence are the same +// number and the opposite claim. +func TestAnUnpricedRowSaysSoInsteadOfClaimingZero(t *testing.T) { + resetExtract() + // A library embedding: calls are made, ModelCall.CostUSD is never filled, and the removal is + // worth something — the exact shape that read as "profitable" before. + RecordExtractionCall("extract_llm", 100) + RecordExtractionValue("extract_llm", 0.05) + s := ExtractSnapshot(9.99, 0.30/1e6, 0, 0) + + if s.ExtractionCostUSD != 9.99 || s.CostSource != costSourceHost { + t.Errorf("aggregate = $%v from %q, want 9.99 from %q: with nothing priced, the host's "+ + "figure is all the information there is", s.ExtractionCostUSD, s.CostSource, + costSourceHost) + } + row := s.ByComponent["extract_llm"] + if row == nil { + t.Fatal("extract_llm missing from the breakdown") + } + if row.CostSource != costSourceUnpriced { + t.Errorf("row cost_source = %q, want %q — $0 must not be indistinguishable from "+ + "'no spend recorded'", row.CostSource, costSourceUnpriced) + } + if net, known := row.Net(); known { + t.Errorf("row net_value_usd = %v, want null: the component made a call and priced none "+ + "of it, so nothing is known about whether it paid", net) + } + // The aggregate's net is always known — it always has a determined spend behind it. + if _, known := s.Net(); !known { + t.Error("the aggregate net must never be null") + } +} + +// REVIEW FINDING (#178), the other half. `anySpend` was one boolean across every component, so +// PARTIAL recording silently under-reported: if extract_llm priced its calls (the cheap card is +// never zero) while extract_llm_sweep priced none, the total became extract_llm's spend alone and +// the sweep's real dollars vanished — no fallback, no warning, a SMALLER number than the code +// published before this PR. +func TestPartialPricingDoesNotSilentlyUnderReportTheTotal(t *testing.T) { + resetExtract() + RecordExtractionCall("extract_llm", 100) + RecordExtractionSpend("extract_llm", 0.02) // priced + RecordExtractionCall("extract_llm_sweep", 59_000) + // ...and the sweep prices nothing. The host saw $4.00 of cheap-model spend in the process. + s := ExtractSnapshot(4.00, 0.30/1e6, 0, 0) + + if s.ExtractionCostUSD <= 0.02 { + t.Errorf("aggregate = $%v: the recorded sum alone DROPS the unpriced component's "+ + "dollars, reporting less than the host already knew", s.ExtractionCostUSD) + } + if s.ExtractionCostUSD != 4.00 || s.CostSource != costSourceHost { + t.Errorf("aggregate = $%v from %q, want 4.00 from %q (the larger of the floor and the "+ + "host's superset figure, named)", s.ExtractionCostUSD, s.CostSource, costSourceHost) + } + // And when the host's figure is NOT larger, the total is a floor and must say so rather than + // pass itself off as the bill. + resetExtract() + RecordExtractionCall("extract_llm", 100) + RecordExtractionSpend("extract_llm", 5.00) + RecordExtractionCall("extract_llm_sweep", 59_000) + s = ExtractSnapshot(1.00, 0.30/1e6, 0, 0) + if s.ExtractionCostUSD != 5.00 || s.CostSource != costSourcePartial { + t.Errorf("aggregate = $%v from %q, want 5.00 from %q", s.ExtractionCostUSD, + s.CostSource, costSourcePartial) + } +} + +// REVIEW FOLLOW-UP (#178). `cost_source: partial` said the total was a FLOOR without saying what it +// was short OF, so a reader had to go and scan every by_component row for `cost_source: unpriced` — +// re-deriving an answer the snapshot had already computed, since it is the same test that set the +// aggregate's label. An operator who has to re-derive it will read the label and stop, and then the +// floor gets quoted as the bill. +// +// Asserted on the MARSHALLED PAYLOAD, not the Go field. That is the lesson of this PR's M13: a +// `cost_source` mutant tagged `json:"-"` survived the first vacuity pass precisely because the +// assertions read the struct. +func TestTheAggregateNamesWhichComponentIsUnpriced(t *testing.T) { + resetExtract() + RecordExtractionCall("extract_llm", 100) + RecordExtractionSpend("extract_llm", 5.00) // priced + RecordExtractionCall("extract_llm_sweep", 59_000) + // ...and the sweep prices nothing, so the total is a floor and the sweep is why. + b, err := json.Marshal(ExtractSnapshot(1.00, 0.30/1e6, 0, 0)) + if err != nil { + t.Fatal(err) + } + var m map[string]any + if err := json.Unmarshal(b, &m); err != nil { + t.Fatal(err) + } + if m["cost_source"] != costSourcePartial { + t.Fatalf("precondition: rendered cost_source = %v, want %q — without a floor there is "+ + "nothing for the list to name: %s", m["cost_source"], costSourcePartial, b) + } + raw, present := m["unpriced_components"] + if !present { + t.Fatalf("the rendered block does not name the unpriced component, so `partial` still "+ + "leaves the reader to scan every row: %s", b) + } + list, ok := raw.([]any) + if !ok || len(list) == 0 { + t.Fatalf("unpriced_components = %v, want a non-empty list", raw) + } + if len(list) != 1 || list[0] != "extract_llm_sweep" { + t.Errorf("unpriced_components = %v, want [extract_llm_sweep]: extract_llm priced its "+ + "call and must not be named, the sweep did not and must be", list) + } + + // And it is OMITTED when every call priced itself — an empty list on a complete total would + // read as a warning where there is nothing to warn about. + resetExtract() + RecordExtractionCall("extract_llm", 100) + RecordExtractionSpend("extract_llm", 5.00) + b2, _ := json.Marshal(ExtractSnapshot(1.00, 0.30/1e6, 0, 0)) + var m2 map[string]any + _ = json.Unmarshal(b2, &m2) + if m2["cost_source"] != costSourceComponent { + t.Fatalf("precondition: rendered cost_source = %v, want %q", m2["cost_source"], + costSourceComponent) + } + if _, present := m2["unpriced_components"]; present { + t.Errorf("unpriced_components must be omitted when nothing is unpriced: %s", b2) + } +} + +// The breakdown has to survive JSON, because /stats is the only place anyone reads it and a +// field can be computed correctly and dropped by the encoder. +func TestByComponentSurvivesJSON(t *testing.T) { + resetExtract() + RecordExtractionCall("extract_llm_sweep", 59_000) + b, err := json.Marshal(ExtractSnapshot(0.1, 0.30/1e6, 0, 0)) + if err != nil { + t.Fatal(err) + } + var m map[string]any + if err := json.Unmarshal(b, &m); err != nil { + t.Fatal(err) + } + by, ok := m["by_component"].(map[string]any) + if !ok { + t.Fatalf("by_component missing from the encoded block: %s", b) + } + row, ok := by["extract_llm_sweep"].(map[string]any) + if !ok { + t.Fatalf("extract_llm_sweep row missing: %s", b) + } + if row["avg_latency_ms"] != 59_000.0 { + t.Errorf("avg_latency_ms = %v in the encoded row, want 59000", row["avg_latency_ms"]) + } + // cost_source must survive the ENCODER, on the row and on the block. It is the field that + // separates "$0 spent" from "no spend recorded", so a value computed correctly and dropped by + // the json tag would put the reader back where #176 found them. Verified non-vacuous: tagging + // the field `json:"-"` makes this fail. + for _, k := range []string{"cost_source", "net_value_usd", "extraction_cost_usd"} { + if _, ok := row[k]; !ok { + t.Errorf("the encoded by_component row is missing %q: %s", k, b) + } + if _, ok := m[k]; !ok { + t.Errorf("the encoded extract block is missing %q: %s", k, b) + } + } + // This row made a call and priced none of it, so the rendered net must be JSON null and the + // source must name the case. A 0 here reads as break-even on no evidence. + if row["cost_source"] != "unpriced" { + t.Errorf("rendered cost_source = %v, want \"unpriced\"", row["cost_source"]) + } + if v, present := row["net_value_usd"]; !present || v != nil { + t.Errorf("rendered net_value_usd = %v (present=%v), want null", v, present) + } + // It must not recurse: a nested row carrying its own breakdown would be an infinite + // document, and the omitempty is what prevents it. + if _, nested := row["by_component"]; nested { + t.Error("the breakdown nests inside itself") + } + // And it must be OMITTED when nothing recorded, so a parser that does not know the key + // sees no change at all. + resetExtract() + b2, _ := json.Marshal(ExtractSnapshot(0, 0.30/1e6, 0, 0)) + var m2 map[string]any + _ = json.Unmarshal(b2, &m2) + if _, present := m2["by_component"]; present { + t.Errorf("by_component must be omitted when empty: %s", b2) + } +} + +// REGRESSION (#176, second half). `acted` is `Saved() > 0`, and a frozen decision replayed on a +// later turn saves tokens for free — so 2,291 replays and a handful of paid extractions landed +// in one counter. The measured snapshot showed `acted: 239` on a component whose own record +// proved it made no call, and that number was read as 239 paid extractions. +func TestActedSeparatesFreeReplaysFromPaidWork(t *testing.T) { + a := NewAggregator() + // Three requests that only REPLAYED a decision frozen on an earlier turn: tokens saved, + // nothing spent, no model call. + for i := 0; i < 3; i++ { + r := components.Report{Component: "extract_llm", Kind: "offload", + TokensBefore: 10_000, TokensAfter: 6_000} + r.Replay("reapplied_same_session") + a.Component(r) + } + // One request that actually paid for a call. + a.Component(components.Report{Component: "extract_llm", Kind: "offload", + TokensBefore: 10_000, TokensAfter: 6_000, + Calls: []components.ModelCall{{Component: "extract_llm", Model: "haiku", CostUSD: 0.01}}}) + + snap := a.Snapshot() + cs, ok := snap.Components["extract_llm"] + if !ok { + t.Fatal("extract_llm missing from the rollup") + } + if cs.Acted != 4 { + t.Fatalf("acted = %d, want 4 (the pre-existing key keeps its meaning)", cs.Acted) + } + if cs.ActedReplay != 3 { + t.Errorf("acted_replay = %d, want 3 — free replays counted as paid work (this is #176)", + cs.ActedReplay) + } + if cs.ActedFresh != 1 { + t.Errorf("acted_fresh = %d, want 1", cs.ActedFresh) + } + if cs.ActedFresh+cs.ActedReplay != cs.Acted { + t.Errorf("the split must partition acted: %d + %d != %d", + cs.ActedFresh, cs.ActedReplay, cs.Acted) + } + // A deterministic component does fresh work every run and pays no model for it; it must + // not read as replaying. + a.Component(components.Report{Component: "dedup", Kind: "offload", + TokensBefore: 100, TokensAfter: 40}) + if d := a.Snapshot().Components["dedup"]; d.ActedFresh != 1 || d.ActedReplay != 0 { + t.Errorf("dedup: fresh=%d replay=%d, want 1/0", d.ActedFresh, d.ActedReplay) + } +} diff --git a/metrics/extract_test.go b/metrics/extract_test.go index 6985f980..48ec2296 100644 --- a/metrics/extract_test.go +++ b/metrics/extract_test.go @@ -11,11 +11,15 @@ func TestNetValueGoesNegativeWhenUnderwater(t *testing.T) { resetExtract() // The measured Terminal-Bench shape: ~197,548 unique tokens saved at the cache-read // rate ($0.30/MTok) against $3.26 of extraction spend. - RecordExtractionSaving(197548) + RecordExtractionSaving("extract_llm", 197548) s := ExtractSnapshot(3.26, 0.30/1e6, 0, 0) - if s.NetValueUSD >= 0 { + net, known := s.Net() + if !known { + t.Fatalf("the aggregate net must always be known, got null: %+v", s) + } + if net >= 0 { t.Fatalf("net must be negative when spend exceeds value: net=%v gross=%v cost=%v", - s.NetValueUSD, s.GrossValueUSD, s.ExtractionCostUSD) + net, s.GrossValueUSD, s.ExtractionCostUSD) } // And the ratio must reproduce the issue's ~8x-underwater claim to the right order. if ratio := s.ExtractionCostUSD / s.GrossValueUSD; ratio < 40 { @@ -27,14 +31,14 @@ func TestNetValueGoesNegativeWhenUnderwater(t *testing.T) { // calls avoided by cache and calls suppressed by the gate. func TestExtractSnapshotExposesAllCounters(t *testing.T) { resetExtract() - RecordExtractionCall(450) - RecordExtractionCall(550) - RecordExtractionCacheLookup(true) - RecordExtractionCacheLookup(true) - RecordExtractionCacheLookup(false) - RecordExtractionSuppressed("suppressed: cache-aware, saving below call cost") - RecordExtractionSaving(1200) - RecordExtractionReason("high context pressure") + RecordExtractionCall("extract_llm", 450) + RecordExtractionCall("extract_llm", 550) + RecordExtractionCacheLookup("extract_llm", true) + RecordExtractionCacheLookup("extract_llm", true) + RecordExtractionCacheLookup("extract_llm", false) + RecordExtractionSuppressed("extract_llm", "suppressed: cache-aware, saving below call cost") + RecordExtractionSaving("extract_llm", 1200) + RecordExtractionReason("extract_llm", "high context pressure") // 1,200 own saved tokens at a rate chosen to give gross value exactly $0.50. s := ExtractSnapshot(0.024, 0.5/1200, 800, 0) @@ -57,8 +61,8 @@ func TestExtractSnapshotExposesAllCounters(t *testing.T) { if d := s.CacheHitRate - want; d > 1e-9 || d < -1e-9 { t.Errorf("CacheHitRate = %v, want %v", s.CacheHitRate, want) } - if s.NetValueUSD != 0.476 { - t.Errorf("NetValueUSD = %v, want 0.476", s.NetValueUSD) + if net, known := s.Net(); !known || net != 0.476 { + t.Errorf("NetValueUSD = %v (known=%v), want 0.476", net, known) } // The trigger reason must be recoverable — an operator's first question. if s.TopReason == "" || len(s.Reasons) != 2 { @@ -71,7 +75,7 @@ func TestExtractSnapshotExposesAllCounters(t *testing.T) { func TestPromptCacheReadZeroIsReported(t *testing.T) { resetExtract() for i := 0; i < 5; i++ { - RecordExtractionCall(400) + RecordExtractionCall("extract_llm", 400) } s := ExtractSnapshot(0.06, 0.30/1e6, 0, 0) if s.Calls != 5 || s.PromptCacheReadTokens != 0 { @@ -129,17 +133,13 @@ func TestSnapshotJSONKeysAreBackwardCompatible(t *testing.T) { } } -// resetExtract clears the process counters so assertions are independent. +// resetExtract clears the per-component counters so assertions are independent. Dropping the +// whole registry rather than zeroing each field: a counter added later would otherwise leak +// across tests silently, which is the failure mode a reset helper exists to prevent. func resetExtract() { - xCalls.Store(0) - xCacheHits.Store(0) - xSuppressed.Store(0) - xGrossSaved.Store(0) - xLatencyMs.Store(0) - xLookups.Store(0) - xReasonMu.Lock() - xReasons = map[string]int64{} - xReasonMu.Unlock() + xRegMu.Lock() + xReg = map[string]*xCounters{} + xRegMu.Unlock() } // REGRESSION (H3, reviewer-verified): /stats must value extract_llm's OWN savings, never @@ -154,7 +154,7 @@ func resetExtract() { func TestNetValueUsesComponentOwnSavingsNotPipelineTotal(t *testing.T) { resetExtract() // The component itself saved 1,000 tokens and spent $0.05. - RecordExtractionSaving(1000) + RecordExtractionSaving("extract_llm", 1000) const rate = 0.30 / 1e6 // cache-read rate per token s := ExtractSnapshot(0.05, rate, 0, 0) @@ -162,8 +162,9 @@ func TestNetValueUsesComponentOwnSavingsNotPipelineTotal(t *testing.T) { if d := s.GrossValueUSD - round4(wantGross); d > 1e-9 || d < -1e-9 { t.Fatalf("GrossValueUSD = %v, want %v (1,000 own tokens x rate)", s.GrossValueUSD, round4(wantGross)) } - if s.NetValueUSD >= 0 { - t.Fatalf("net must be negative: spent $0.05 to save $%.6f", wantGross) + if net, known := s.Net(); !known || net >= 0 { + t.Fatalf("net must be negative (got %v, known=%v): spent $0.05 to save $%.6f", + net, known, wantGross) } // The pipeline-wide figure in a real run is orders of magnitude larger. Prove the // snapshot is NOT reading anything like it: had a 2,000,000-token pipeline total leaked @@ -181,12 +182,12 @@ func TestNetValueUsesComponentOwnSavingsNotPipelineTotal(t *testing.T) { // count (0 calls => no signal, must not read as "fast"). func TestExtractionAvgLatencyMs(t *testing.T) { resetExtract() - if avg, calls := ExtractionAvgLatencyMs(); avg != 0 || calls != 0 { + if avg, calls := ExtractionAvgLatencyMs("extract_llm"); avg != 0 || calls != 0 { t.Fatalf("with no calls expected (0,0), got (%v,%d)", avg, calls) } - RecordExtractionCall(4000) - RecordExtractionCall(8000) - avg, calls := ExtractionAvgLatencyMs() + RecordExtractionCall("extract_llm", 4000) + RecordExtractionCall("extract_llm", 8000) + avg, calls := ExtractionAvgLatencyMs("extract_llm") if calls != 2 || avg != 6000 { t.Fatalf("got (%v,%d), want (6000,2)", avg, calls) } diff --git a/metrics/metrics.go b/metrics/metrics.go index 6a1eaf2e..9f26d70d 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -213,8 +213,24 @@ func (a *Aggregator) SetMode(m components.Mode) { } type compStat struct { - Runs int64 `json:"runs"` - Acted int64 `json:"acted"` // runs that actually saved tokens + Runs int64 `json:"runs"` + Acted int64 `json:"acted"` // runs that actually saved tokens + // ActedFresh and ActedReplay partition Acted by whether the saving cost anything (#176). + // + // `acted` alone cannot answer the only question an operator asks about an expensive + // component — "is it still spending?" — because a frozen decision replayed on every + // subsequent turn saves tokens for free and lands in the same counter as the call that + // derived it. MEASURED (iteration 023, arm B): `acted: 239` beside `reapplied_same_session: + // 2,291` on a component whose own debug record showed no candidate surviving its gates on + // any of 374 requests. The reading taken from `acted` was "this component made 239 paid + // extractions"; the truth was zero. + // + // A run is ActedReplay when it saved tokens, made NO model call, and replayed at least one + // frozen decision — free work. Everything else that saved tokens is ActedFresh, which is + // therefore also what every deterministic component reports (it does fresh work every time + // it runs, it just does not pay a model for it). + ActedFresh int64 `json:"acted_fresh"` + ActedReplay int64 `json:"acted_replay"` Mutated int64 `json:"mutated"` // runs that changed the request at all (may save 0 content tokens, e.g. cacheinject) Reverted int64 `json:"reverted"` Saved int64 `json:"saved_tokens"` // CUMULATIVE: summed every turn the compaction re-appears @@ -296,6 +312,22 @@ func (cs compStat) forSnapshot() compStat { return cs } +// act credits one run that saved tokens, and classifies it as FRESH work or a free REPLAY. +// +// One method called from both the enforced and the observe loop, rather than the same three +// lines written twice: the two loops in this file already drifted once over copying Gates, and +// the whole point of this counter is that a wrong reading of it is not visible in the output. +// A replay is a run that saved tokens without making a model call, off a decision the +// component had already frozen — see components.Report.Replays. +func (cs *compStat) act(r components.Report) { + cs.Acted++ + if len(r.Calls) == 0 && r.Replays > 0 { + cs.ActedReplay++ + return + } + cs.ActedFresh++ +} + // verdict reads the counters. See compStat.Verdict for why this exists rather than leaving // a consumer to compare acted against mutated. func (cs compStat) verdict() string { @@ -406,7 +438,7 @@ func (a *Aggregator) Component(r components.Report) { cs.Mutated++ // did something, even if it saved no content tokens } if r.Saved() > 0 && !r.Reverted && !r.Skipped { - cs.Acted++ + cs.act(r) } } @@ -486,7 +518,7 @@ func (a *Aggregator) observeComp(r components.Report) { cs.Mutated++ } if r.Saved() > 0 && !r.Reverted && !r.Skipped { - cs.Acted++ + cs.act(r) } } diff --git a/proxy/extract_attribution_test.go b/proxy/extract_attribution_test.go new file mode 100644 index 00000000..804d5315 --- /dev/null +++ b/proxy/extract_attribution_test.go @@ -0,0 +1,179 @@ +package proxy + +import ( + "encoding/json" + "net/http/httptest" + "testing" + + "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/metrics" +) + +// The attribution fix has to reach the SURFACE, not just the counter. /stats is where the +// misattribution was read (#176) — 101 calls and a 59-second mean charged to a component whose +// own log said it made none — and the statsHandler builds the extract block itself, so a +// correctly scoped counter can still arrive pooled or be dropped entirely by the handler. +// +// Asserted on the rendered body for that reason. +func TestStatsRendersPerComponentExtractionAttribution(t *testing.T) { + // These counters are process-global by construction (one proxy, one process), so the + // assertions below are stated against a BASELINE read rather than against zero: another + // test in this package that exercised extract_llm for real would otherwise make this one + // fail for a reason that has nothing to do with #176. + base := metrics.ExtractSnapshot(0, 0, 0, 0) + baseTailCalls, baseTailCost := int64(0), 0.0 + if r := base.ByComponent["extract_llm"]; r != nil { + baseTailCalls, baseTailCost = r.Calls, r.ExtractionCostUSD + } + + // The measured shape: the sweep pays for the calls, extract_llm only replays. + metrics.RecordExtractionCall("extract_llm_sweep", 59_009) + metrics.RecordExtractionSpend("extract_llm_sweep", 1.20) + // Value but no call and no lookup: that IS extract_llm's measured behaviour here (frozen + // replays are worth something and cost nothing), and it deliberately touches no counter + // another test in this package asserts an absolute value on. + metrics.RecordExtractionValue("extract_llm", 0.0038) + + agg := metrics.NewAggregator() + // And the acted split: three free replays against one paid call, which is the pair that + // `acted: 239` beside `reapplied_same_session: 2,291` could not distinguish. + for i := 0; i < 3; i++ { + r := components.Report{Component: "extract_llm", Kind: "offload", + TokensBefore: 10_000, TokensAfter: 6_000} + r.Replay("reapplied_same_session") + agg.Component(r) + } + agg.Component(components.Report{Component: "extract_llm", Kind: "offload", + TokensBefore: 10_000, TokensAfter: 6_000, + Calls: []components.ModelCall{{Component: "extract_llm", Model: "haiku", CostUSD: 0.01}}}) + + h := New(nil, nil, agg, Options{}) + w := httptest.NewRecorder() + h.stats(w, httptest.NewRequest("GET", "/stats", nil)) + + var got struct { + Extract struct { + Calls int64 `json:"calls"` + CostSource string `json:"cost_source"` + ByComponent map[string]struct { + Calls int64 `json:"calls"` + AvgLatencyMs float64 `json:"avg_latency_ms"` + ExtractionCostUSD float64 `json:"extraction_cost_usd"` + NetValueUSD *float64 `json:"net_value_usd"` + CostSource string `json:"cost_source"` + } `json:"by_component"` + } `json:"extract"` + Components map[string]struct { + Acted int64 `json:"acted"` + ActedFresh int64 `json:"acted_fresh"` + ActedReplay int64 `json:"acted_replay"` + } `json:"components"` + } + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { + t.Fatalf("/stats is not the expected shape: %v\n%s", err, w.Body.String()) + } + + if len(got.Extract.ByComponent) == 0 { + t.Fatalf("/stats serves no extract.by_component, so the pooled figure is again the "+ + "only one an operator can read: %s", w.Body.String()) + } + sweep, ok := got.Extract.ByComponent["extract_llm_sweep"] + if !ok { + t.Fatalf("extract_llm_sweep absent from extract.by_component: %s", w.Body.String()) + } + tail := got.Extract.ByComponent["extract_llm"] + // THE DEFECT, at the surface: the calls and the latency belong to the sweep alone. + if sweep.Calls == 0 || sweep.AvgLatencyMs == 0 { + t.Errorf("the sweep's own calls/latency did not reach /stats: %+v", sweep) + } + if tail.Calls != baseTailCalls { + t.Errorf("/stats credits extract_llm with %d calls (baseline %d); it made none here, "+ + "and the sweep's must not leak into its row (#176)", tail.Calls, baseTailCalls) + } + if sweep.ExtractionCostUSD == 0 { + t.Error("the sweep's spend did not reach /stats, so its net value is unchecked") + } + if tail.ExtractionCostUSD != baseTailCost { + t.Errorf("extract_llm charged $%v (baseline $%v) it did not spend", + tail.ExtractionCostUSD, baseTailCost) + } + // cost_source has to reach the WIRE, on the block and on every row: it is what separates a + // provable $0 from an unknown, and the statsHandler builds this block itself so the field can + // be computed and dropped. The sweep priced its own call; extract_llm made none, so its 0 is + // provable rather than unknown. + if got.Extract.CostSource == "" { + t.Errorf("/stats serves no extract.cost_source: %s", w.Body.String()) + } + if sweep.CostSource != "component" { + t.Errorf("sweep cost_source = %q on the wire, want \"component\"", sweep.CostSource) + } + if tail.CostSource != "none" { + t.Errorf("extract_llm cost_source = %q on the wire, want \"none\" (no calls, so $0 is "+ + "the true figure and not a missing one)", tail.CostSource) + } + if sweep.NetValueUSD == nil { + t.Error("the sweep priced its own call, so its net_value_usd must not be null on the wire") + } + + cs, ok := got.Components["extract_llm"] + if !ok { + t.Fatalf("extract_llm absent from components: %s", w.Body.String()) + } + if cs.Acted != 4 { + t.Errorf("acted = %d, want 4 (the pre-existing key must keep its meaning)", cs.Acted) + } + if cs.ActedReplay != 3 || cs.ActedFresh != 1 { + t.Errorf("/stats acted split = fresh %d / replay %d, want 1/3 — without it a component "+ + "amortizing frozen work reads identically to one making paid calls", + cs.ActedFresh, cs.ActedReplay) + } +} + +// REVIEW FOLLOW-UP (#178), at the surface. metrics computes `unpriced_components`; the statsHandler +// builds the extract block itself, so the list can be correct and never reach the wire. Same reason +// the rest of this file asserts on the rendered body. +// +// A component name of its own rather than one of the two real ones: these counters are +// process-global for the whole package run, and TestStatsRendersPerComponentExtractionAttribution +// asserts ABSOLUTE values on `extract_llm` and `extract_llm_sweep` (zero calls, zero cost). Making +// either of them unpriced here would break that test through shared state, for a reason unrelated to +// what either test is about. What is under test is the aggregate's list, not the name in it. +func TestStatsNamesTheUnpricedComponent(t *testing.T) { + const unpriced = "extract_llm_unpriced_fixture" + metrics.RecordExtractionCall(unpriced, 1200) // calls, and no RecordExtractionSpend + + h := New(nil, nil, metrics.NewAggregator(), Options{}) + w := httptest.NewRecorder() + h.stats(w, httptest.NewRequest("GET", "/stats", nil)) + + var got struct { + Extract struct { + CostSource string `json:"cost_source"` + UnpricedComponents []string `json:"unpriced_components"` + } `json:"extract"` + } + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { + t.Fatalf("/stats is not the expected shape: %v\n%s", err, w.Body.String()) + } + // PRECONDITION: the total is incomplete, so there is something for the list to name. Without + // this the assertion below could pass vacuously on a snapshot that had nothing to report. + if got.Extract.CostSource != "partial" && got.Extract.CostSource != "host_total" { + t.Fatalf("cost_source = %q, want partial or host_total: an unpriced call must make the "+ + "total incomplete, or this test proves nothing", got.Extract.CostSource) + } + if len(got.Extract.UnpricedComponents) == 0 { + t.Fatalf("/stats serves an incomplete total with no unpriced_components, so an operator "+ + "cannot tell WHAT it is short of: %s", w.Body.String()) + } + var found bool + for _, n := range got.Extract.UnpricedComponents { + if n == unpriced { + found = true + } + } + if !found { + t.Errorf("unpriced_components = %v on the wire, missing %q — the component that made an "+ + "unpriced call is the one the reader needs named", + got.Extract.UnpricedComponents, unpriced) + } +} diff --git a/proxy/promexport.go b/proxy/promexport.go index 4215a801..192d8f8b 100644 --- a/proxy/promexport.go +++ b/proxy/promexport.go @@ -611,7 +611,13 @@ func (h *Handler) renderMetrics() string { promLine(&b, "cg_extract_cost_usd", "", xs.ExtractionCostUSD) promHeader(&b, "cg_extract_net_value_usd", "What extraction's own saved tokens are worth at the rate they would have been billed, MINUS what it spent. NEGATIVE means the component is underwater and should be turned off — alert on it.", "gauge") - promLine(&b, "cg_extract_net_value_usd", "", xs.NetValueUSD) + // The aggregate always has a determined spend behind it (the components', the host's, or none + // at all), so this is never the unknown case — but read it through Net() rather than + // dereferencing, so a future snapshot that cannot price the total omits the series instead of + // publishing an unknown as 0, which on this gauge reads as exactly break-even. + if net, known := xs.Net(); known { + promLine(&b, "cg_extract_net_value_usd", "", net) + } promHeader(&b, "cg_extract_latency_ms", "Mean wall time per extraction call. The gate stops speculative calls once this is observed to be slow, so it is an input as well as a symptom.", "gauge") promLine(&b, "cg_extract_latency_ms", "", xs.AvgLatencyMs) diff --git a/proxy/promexport_components_test.go b/proxy/promexport_components_test.go index ca455eaf..a9d518a2 100644 --- a/proxy/promexport_components_test.go +++ b/proxy/promexport_components_test.go @@ -61,8 +61,8 @@ func TestMutatedAndGateDeclinesAreExported(t *testing.T) { // TestExtractEconomicsAreExported. NetValueUSD was -$0.7085 live and appeared only in // /stats, which nothing scrapes and nothing alerts on. func TestExtractEconomicsAreExported(t *testing.T) { - metrics.RecordExtractionSuppressed("gate:not_worth_it") - metrics.RecordExtractionCacheLookup(true) + metrics.RecordExtractionSuppressed("extract_llm", "gate:not_worth_it") + metrics.RecordExtractionCacheLookup("extract_llm", true) h := New(nil, nil, metrics.NewAggregator(), Options{}) body := h.renderMetrics() for _, want := range []string{ diff --git a/proxy/stats_golden_test.go b/proxy/stats_golden_test.go index 4062b662..13a45f62 100644 --- a/proxy/stats_golden_test.go +++ b/proxy/stats_golden_test.go @@ -109,6 +109,13 @@ var statsGoldenTopLevel = []string{ // by name. var statsGoldenComponent = []string{ "acted", + // acted_fresh / acted_replay partition acted by whether the saving cost anything (#176). + // `acted` alone counted a frozen decision replayed on a later turn — free, no model call — + // in the same figure as the call that derived it, so a measured `acted: 239` beside + // `reapplied_same_session: 2,291` was read as 239 paid extractions on a component that + // made none. Added to the reviewed contract rather than loosening the assertion. + "acted_fresh", + "acted_replay", "discarded_changes", "duration_ms", "mutated",