From fd354b76811611169d242d538383a63bfb4585b8 Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Tue, 3 Nov 2026 01:50:28 +0200 Subject: [PATCH 1/4] feat(dash): replay the per-wait ping plan in the keep-alive calculator (#868) The calculator and the recommendation modelled a fixed K per idle span. Since #867 the proxy picks the count from what the thread waits for. - keepalive: Policy.SpanPings is the keeper's rule for a stored row (the wait cause, the no-tool rule, done, retry, OpenAI, 1-hour tier). It calls WaitPings. The tool-stop rule moves into one shared function. - dash: pingSpans and the expiry read carry the stored facts of the request before each gap, with the thread's tool history, in one statement each. - The calculator has a live-plan rung marked current, and the fixed-K rungs beside it. Each rung and the recommendation state their model (plan: wait_plan or fixed_k). keepalive_wait_plan: false keeps fixed K. - The proxy tells the dashboard the account's plan settings. Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- cmd/context-guru-proxy/main.go | 2 + dash/api.go | 4 + dash/cachelifetime_test.go | 3 +- dash/keepalive.go | 361 +++++++++++++++++++------ dash/keepaliveapi.go | 19 +- dash/keepaliveplan_test.go | 302 +++++++++++++++++++++ dash/keepalivetab_test.go | 34 ++- dash/testdata/golden/ka_calc.json | 16 +- dash/testdata/golden/ka_recommend.json | 5 + dash/ui/app.js | 50 +++- docs/data-dependence.md | 2 +- docs/how-to/dashboard-guide.md | 2 +- keepalive/waitplan.go | 70 ++++- proxy/keepalive.go | 24 ++ 14 files changed, 778 insertions(+), 116 deletions(-) create mode 100644 dash/keepaliveplan_test.go diff --git a/cmd/context-guru-proxy/main.go b/cmd/context-guru-proxy/main.go index 78be2b781..498ca6d1b 100644 --- a/cmd/context-guru-proxy/main.go +++ b/cmd/context-guru-proxy/main.go @@ -742,6 +742,8 @@ func main() { h.API().SetWindows(windows) // The keeper's ping plan, so /api/cachestate can say how long a cache is held warm. h.API().SetKeepAlivePlans(h.KeepAlivePlans) + // The account's per-wait plan settings, so the calculator replays the policy that runs (#868). + h.API().SetWaitPolicy(h.DashWaitPolicy()) h.API().SetPingStates(h.KeepAlivePingStates) // The configuration the adapt reports compare against: each account's in hosted mode, this // proxy's own (with any applied override) otherwise. diff --git a/dash/api.go b/dash/api.go index c05888f56..3ff691f82 100644 --- a/dash/api.go +++ b/dash/api.go @@ -18,6 +18,7 @@ import ( "github.com/rossoctl/context-guru/apply" "github.com/rossoctl/context-guru/internal/modelinfo" + "github.com/rossoctl/context-guru/keepalive" "golang.org/x/sync/errgroup" ) @@ -52,6 +53,9 @@ type API struct { // pingable says per thread whether a manual ping can be sent. nil = never stated. pingable func(tenantID string, allTenants bool, session string) map[string]PingState kaPlans func(tenantID string, allTenants bool, session string) map[string]KeepAlivePlan + // waitPolicyFn reads a tenant's per-wait ping plan settings for the keep-alive calculator and + // recommendation (#868). nil = the shipped plan, on. + waitPolicyFn func(tenantID string) keepalive.Policy // pricer values the pre-instrumentation split figure on read. nil = that figure is omitted. pricer modelinfo.Pricer // windows resolves a model's context window, and is separate from pricer because the two diff --git a/dash/cachelifetime_test.go b/dash/cachelifetime_test.go index 4b5d2eb80..5143cd48f 100644 --- a/dash/cachelifetime_test.go +++ b/dash/cachelifetime_test.go @@ -1,6 +1,7 @@ package dash import ( + "github.com/rossoctl/context-guru/keepalive" "math" "testing" "time" @@ -247,7 +248,7 @@ func TestKeepAliveCalcJudgesAnExpiryAgainstItsOwnLifetime(t *testing.T) { {"anthropic", "aws/claude-sonnet-5", 0}, } { fx := newKAFixture(t, exp("s", c.provider, c.model)...) - out, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 280, 100_000, c.model, price, 2) + out, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 280, 100_000, c.model, price, 2, keepalive.Policy{}) if err != nil { t.Fatal(err) } diff --git a/dash/keepalive.go b/dash/keepalive.go index a681b2f31..c3584cfcd 100644 --- a/dash/keepalive.go +++ b/dash/keepalive.go @@ -13,6 +13,7 @@ import ( "github.com/rossoctl/context-guru/apply" "github.com/rossoctl/context-guru/internal/modelinfo" + "github.com/rossoctl/context-guru/keepalive" "github.com/rossoctl/context-guru/kvcache" ) @@ -1272,6 +1273,9 @@ type pingSpan struct { // gap is the seconds until the next request in the session. Open says there was none. gap float64 open bool + // facts are what the request that opens the span says about the wait: its stop reason, wait + // cause, thread kind and tool history. Filled only when the replay models the per-wait plan. + facts keepalive.SpanFacts } // pings is how many pings this span attracts under (X, K). @@ -1287,6 +1291,50 @@ func (s pingSpan) pings(idleSeconds float64, maxPings int) int { return PingsPerSpan(s.gap, idleSeconds, maxPings) } +// planPings is how many pings this span attracts under the per-wait plan (#868). The count comes +// from keepalive.Policy.SpanPings, the keeper's own rule; the span's length caps it as in pings. +func (s pingSpan) planPings(idleSeconds float64, pol keepalive.Policy) int { + return s.pings(idleSeconds, pol.SpanPings(s.facts)) +} + +// toolStopList is keepalive.ToolStopReasons as a SQL list of string literals. The names are fixed +// constants of the keeper, never user input. +func toolStopList() string { + q := make([]string, len(keepalive.ToolStopReasons)) + for i, r := range keepalive.ToolStopReasons { + q[i] = "'" + r + "'" + } + return strings.Join(q, ",") +} + +// factsCTE reads each request with the facts the plan needs. toolSeen is true when this request or an +// earlier one of the same THREAD called a tool: a tool stop reason, or a stored wait on its own tool. +// Pings are excluded by the filter, as in every replay. One extra window, partitioned by thread. +func factsCTE(cond string) string { + return `t AS ( + SELECT r.id AS id, r.tenant_id AS tenant_id, r.session_id AS session_id, r.thread_id AS thread_id, + r.ts AS ts, r.cache_read + r.cache_write AS prefix, r.cache_write AS cache_write, + r.cost_usd AS cost_usd, r.model AS model, r.provider AS provider, + r.cache_miss_reason AS cache_miss_reason, r.cache_ttl AS cache_ttl, + COALESCE(r.stop_reason, '') AS stop, COALESCE(r.waiting_for, '') AS wf, + MAX(CASE WHEN r.stop_reason IN (` + toolStopList() + `) OR r.waiting_for = 'own_tool' + THEN 1 ELSE 0 END) + OVER (PARTITION BY r.tenant_id, r.session_id, r.thread_id ORDER BY r.ts, r.id) AS seen + FROM requests r WHERE ` + cond + `)` +} + +// scanFacts builds the SpanFacts of one stored row. +func scanFacts(thread, stop, wf, provider, cacheTTL string, seen int64) keepalive.SpanFacts { + return keepalive.SpanFacts{Cause: wf, StopReason: stop, Subagent: keepalive.IsSubagentThread(thread), + ToolSeen: seen > 0, OpenAI: provider == "openai", OneHour: cacheTTL == apply.TTLEphemeral1h} +} + +// DefaultWaitPolicy is the shipped keep-alive policy for the replay: the per-wait plan on, with +// `base` as the base ping count. A host that knows the account's own policy supplies that instead. +func DefaultWaitPolicy(base int) keepalive.Policy { + return keepalive.Policy{WaitPlan: true, MaxPings: base} +} + // pingSpans is the ONE definition of "the idle spans a live keep-alive would ping in", shared by // the calculator and by the recommendation's bootstrap so the two cannot disagree about what a // policy costs. @@ -1310,13 +1358,20 @@ func (s pingSpan) pings(idleSeconds float64, maxPings int) int { // package does not read tenant configuration — an account that tuned its floor sees a replay of // the default policy, and the panel's copy names 20k so the page stays self-consistent. // -// THE COUNT IS A FIXED K (#858). The live policy now picks the count per wait (keepalive.Policy.WaitPings: -// a tool wait up to 8, background K+3, a person K+1, no tool call yet 0). This replay does not model -// that: it charges one K per span. So the ladder and the recommendation describe the PLAN-OFF policy -// (keepalive_wait_plan: false), and no rung is marked "current". Modelling the plan needs the wait cause -// and tool history of each span and of each expiry's previous request; that is a follow-up. -func (d *DB) pingSpans(f Filter, minPrefix int64) ([]pingSpan, error) { +// THE COUNT (#858, #868). The live policy picks the count per wait (keepalive.Policy.WaitPings: a tool +// wait up to 8, background K+3, a person K+1, no tool call yet 0). With withFacts, every span carries the +// stored facts of the request that opens it (stop reason, wait cause, thread kind, tool history), and +// pingSpan.planPings asks the keeper's own rule (Policy.SpanPings) for the count. Without it, the spans +// carry no facts and only the fixed-K form (pings) is meaningful. +// +// The span is the gap to the next request of the SESSION, and the wait is read from the request that +// opens it, on its own thread. When the main agent and a subagent interleave, the last request before the +// idle gap is the one that decides, as it would for the keeper's per-thread entry that is still warm last. +func (d *DB) pingSpans(f Filter, minPrefix int64, withFacts bool) ([]pingSpan, error) { cond, args := f.where() + if withFacts { + return d.pingSpansWithFacts(cond, args, minPrefix) + } // LEAD, not LAG: a span belongs to the request that OPENS it, which is the request the gate // is evaluated on, and LEAD is NULL exactly on the session-final row. rows, err := d.sql.QueryContext(d.readCtx(), `WITH s AS ( @@ -1346,6 +1401,111 @@ func (d *DB) pingSpans(f Filter, minPrefix int64) ([]pingSpan, error) { return out, rows.Err() } +// pingSpansWithFacts is pingSpans with the plan's facts. One statement: the facts window and the span +// windows run over the same filtered rows. +func (d *DB) pingSpansWithFacts(cond string, args []any, minPrefix int64) ([]pingSpan, error) { + rows, err := d.sql.QueryContext(d.readCtx(), `WITH `+factsCTE(cond)+`, s AS ( + SELECT session_id, thread_id, provider, cache_ttl, stop, wf, seen, prefix, + ROW_NUMBER() OVER (PARTITION BY tenant_id, session_id ORDER BY ts, id) - 1 AS turn, + (LEAD(ts) OVER (PARTITION BY tenant_id, session_id ORDER BY ts, id) - ts) / 1000.0 AS gap_s + FROM t) + SELECT session_id, gap_s, thread_id, provider, cache_ttl, stop, wf, seen FROM s + WHERE turn >= 1 AND prefix >= ? AND (gap_s IS NULL OR gap_s > 0)`, + append(args, minPrefix)...) + if err != nil { + return nil, err + } + defer rows.Close() + var out []pingSpan + for rows.Next() { + var sess, thread, prov, ttl, stop, wf string + var gap sql.NullFloat64 + var seen int64 + if err := rows.Scan(&sess, &gap, &thread, &prov, &ttl, &stop, &wf, &seen); err != nil { + return nil, err + } + out = append(out, pingSpan{session: sess, gap: gap.Float64, open: !gap.Valid, + facts: scanFacts(thread, stop, wf, prov, ttl, seen)}) + } + return out, rows.Err() +} + +// expiry is one addressable cache expiry of the account: the idle gap before it, what it cost, the +// lifetime of its cache entry and, when the replay models the plan, the facts of the request that +// opened the gap. +type expiry struct { + session string + gap float64 + usd float64 + ttl time.Duration + facts keepalive.SpanFacts +} + +// expiries reads the addressable expiries (the addressable CTE), with the plan's facts of the previous +// request when withFacts. Same rows as the addressable CTE, so the fixed-K rungs and the plan rung count +// the same expiries. +func (d *DB) expiries(f Filter, withFacts bool) ([]expiry, error) { + var out []expiry + if !withFacts { + aCond, aArgs := addressable(f) + rows, err := d.sql.QueryContext(d.readCtx(), aCond+` SELECT session_id, gap_s, cost_usd, provider, model FROM addressable`, aArgs...) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var e expiry + var prov, mod string + if err := rows.Scan(&e.session, &e.gap, &e.usd, &prov, &mod); err != nil { + return nil, err + } + e.ttl = missLifetime(prov, mod) + out = append(out, e) + } + return out, rows.Err() + } + cond, args := f.where() + rows, err := d.sql.QueryContext(d.readCtx(), `WITH `+factsCTE(cond)+`, s AS ( + SELECT session_id, ts, cost_usd, model, provider, cache_miss_reason, cache_write, + LAG(ts) OVER w AS prev_ts, LAG(thread_id) OVER w AS p_thread, LAG(stop) OVER w AS p_stop, + LAG(wf) OVER w AS p_wf, LAG(provider) OVER w AS p_prov, LAG(cache_ttl) OVER w AS p_ttl, + LAG(seen) OVER w AS p_seen + FROM t WINDOW w AS (PARTITION BY tenant_id, session_id ORDER BY ts, id)) + SELECT session_id, (ts - prev_ts) / 1000.0, cost_usd, provider, model, + p_thread, p_stop, p_wf, p_prov, p_ttl, p_seen FROM s + WHERE cache_miss_reason = 'ttl_expiry' AND cache_write > 0 AND prev_ts IS NOT NULL`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var e expiry + var prov, mod, pThread, pStop, pWf, pProv, pTTL string + var pSeen int64 + if err := rows.Scan(&e.session, &e.gap, &e.usd, &prov, &mod, &pThread, &pStop, &pWf, &pProv, &pTTL, &pSeen); err != nil { + return nil, err + } + e.ttl = missLifetime(prov, mod) + e.facts = scanFacts(pThread, pStop, pWf, pProv, pTTL, pSeen) + out = append(out, e) + } + return out, rows.Err() +} + +// covers reports whether n pings at idle X reach an expiry that came after `gap` seconds of idle, on a +// cache entry of lifetime ttl. No ping, no reach. +func covers(x float64, n int, gap float64, ttl time.Duration) bool { + return n > 0 && gap > x && gap <= CoverageSecondsFor(x, n, ttl) +} + +// The two models a ladder or a recommendation can use for the ping count (#868). +const ( + // PlanWait is the live per-wait plan (keepalive.Policy.WaitPings and the no-tool rule). + PlanWait = "wait_plan" + // PlanFixed is one K for every wait: the policy with keepalive_wait_plan off. + PlanFixed = "fixed_k" +) + // CalcRow is one rung of the K ladder. type CalcRow struct { MaxPings int `json:"max_pings"` @@ -1364,6 +1524,9 @@ type CalcRow struct { SavedUSD float64 `json:"saved_usd,omitempty"` NetUSD float64 `json:"net_usd,omitempty"` Current bool `json:"current,omitempty"` + // Plan says which model this rung uses: "wait_plan" (the live per-wait plan, with MaxPings as + // its base) or "fixed_k" (MaxPings pings for every wait). + Plan string `json:"plan"` } // KeepAliveCalc is the calculator's whole answer. @@ -1383,7 +1546,9 @@ type KeepAliveCalc struct { Addressable int64 `json:"addressable_misses"` AddressableUSD float64 `json:"addressable_usd"` Rows []CalcRow `json:"rows"` - // Plan says which policy the ladder models: "fixed_k", the policy with keepalive_wait_plan off (#858). + // Plan says which policy the account runs: "wait_plan", or "fixed_k" when its keepalive_wait_plan is + // off (#858). With "wait_plan" the ladder carries a rung of the live plan, marked current, and the + // fixed-K rungs are the comparison. Each rung also states its own model. Plan string `json:"plan"` // PrefixSource says where the prefill came from, so nobody reads a number as measured when // it was typed: "session", "account_median" or "given". @@ -1400,12 +1565,24 @@ type KeepAliveCalc struct { // per-ping cost is bimodal (p50 $0.0004, p99 $0.2275) and there is no honest average of it. // The rates themselves are the operator's price list; a model missing from it leaves every // dollar field omitted (Priced above). +// +// wait is the account's policy for the per-wait plan (#868): WaitPlan on gives the ladder a "wait_plan" +// rung, the live plan with base k, marked current, above the fixed-K rungs it is compared with. WaitPlan +// off (the account's keepalive_wait_plan: false) gives fixed-K rungs only, the one with K = k marked +// current. Only the tuning fields of wait are read; its MaxPings is replaced by k. func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model string, - price func(string) (modelinfo.Price, bool), _ int) (*KeepAliveCalc, error) { + price func(string) (modelinfo.Price, bool), k int, wait keepalive.Policy) (*KeepAliveCalc, error) { if idleSeconds <= 0 { idleSeconds = recIdleSeconds } - out := &KeepAliveCalc{IdleSeconds: idleSeconds, Prefix: prefix, Model: model, Rows: []CalcRow{}, Plan: "fixed_k"} + if k <= 0 { + k = recMaxPings + } + wait.MaxPings = k + out := &KeepAliveCalc{IdleSeconds: idleSeconds, Prefix: prefix, Model: model, Rows: []CalcRow{}, Plan: PlanFixed} + if wait.WaitPlan { + out.Plan = PlanWait + } var p modelinfo.Price if price != nil && model != "" { if pr, ok := price(model); ok && !pr.Zero() { @@ -1414,30 +1591,14 @@ func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model st } // The account's own spans and its own addressable gaps, one read each, and the coverage count: // three independent reads, run together (the first two are each a pass over every request). - aCond, aArgs := addressable(f) - var gaps []float64 - var usd []float64 - var lifetimes []time.Duration // per expiry, parallel to gaps + var exps []expiry var spans []pingSpan var cov *KeepAliveCoverage var g errgroup.Group g.Go(func() error { - rows, err := d.sql.QueryContext(d.readCtx(), aCond+` SELECT gap_s, cost_usd, provider, model FROM addressable`, aArgs...) - if err != nil { - return err - } - defer rows.Close() - for rows.Next() { - var g, u float64 - var prov, mod string - if err := rows.Scan(&g, &u, &prov, &mod); err != nil { - return err - } - gaps = append(gaps, g) - usd = append(usd, u) - lifetimes = append(lifetimes, missLifetime(prov, mod)) - } - return rows.Err() + var err error + exps, err = d.expiries(f, wait.WaitPlan) + return err }) // EVERY idle span the gate admits, not only the ones that expired: the ping cost is paid on // all of them, and counting only the spans that paid off is how a calculator flatters its own @@ -1445,7 +1606,7 @@ func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model st // on the panel, which says which pings are counted. g.Go(func() error { var err error - spans, err = d.pingSpans(f, kaGateMinPrefix) + spans, err = d.pingSpans(f, kaGateMinPrefix, wait.WaitPlan) return err }) g.Go(func() error { @@ -1456,37 +1617,58 @@ func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model st if err := g.Wait(); err != nil { return nil, err } - for _, u := range usd { + for _, e := range exps { out.Addressable++ - out.AddressableUSD += u + out.AddressableUSD += e.usd } if out.Priced { out.PingUSDEach = float64(prefix)*p.CacheRead + p.Output out.AvoidedUSDEach = float64(prefix) * (p.CacheWrite - p.CacheRead) } - for k := 1; k <= 4; k++ { - // Coverage is the account-wide policy figure (the 5-minute default, no row). The test - // below uses each expiry's OWN lifetime, so a GPT-5.6 gap is judged against its 30 min. - row := CalcRow{MaxPings: k, Coverage: CoverageSeconds(idleSeconds, k), - // Never marked: the live policy picks its count per wait (#858), so no fixed-K rung is it. - Current: false} - for i, g := range gaps { - if g > idleSeconds && g <= CoverageSecondsFor(idleSeconds, k, lifetimes[i]) { - row.Convertible++ - row.ConvertibleUSD += usd[i] - } - } + // finish fills the money columns of a rung from its counts. + finish := func(row *CalcRow) { if out.AddressableUSD > 0 { row.SharePct = 100 * row.ConvertibleUSD / out.AddressableUSD } - for _, sp := range spans { - row.Pings += int64(sp.pings(idleSeconds, k)) - } if out.Priced { row.PingUSD = float64(row.Pings) * out.PingUSDEach row.SavedUSD = float64(row.Convertible) * out.AvoidedUSDEach row.NetUSD = row.SavedUSD - row.PingUSD } + } + if wait.WaitPlan { + // The live plan with base k (#868): each expiry and each span gets the count the keeper would + // have planned for the wait before it. The coverage is the base figure; the reach of a wait + // is its own count's (a tool wait up to 8, a thread with no tool call yet 0). + row := CalcRow{MaxPings: k, Coverage: CoverageSeconds(idleSeconds, k), Current: true, Plan: PlanWait} + for _, e := range exps { + if covers(idleSeconds, wait.SpanPings(e.facts), e.gap, e.ttl) { + row.Convertible++ + row.ConvertibleUSD += e.usd + } + } + for _, sp := range spans { + row.Pings += int64(sp.planPings(idleSeconds, wait)) + } + finish(&row) + out.Rows = append(out.Rows, row) + } + for kk := 1; kk <= 4; kk++ { + // Coverage is the account-wide policy figure (the 5-minute default, no row). The test + // below uses each expiry's OWN lifetime, so a GPT-5.6 gap is judged against its 30 min. + // The fixed rung with K = k is the current one only when the plan is off. + row := CalcRow{MaxPings: kk, Coverage: CoverageSeconds(idleSeconds, kk), Plan: PlanFixed, + Current: !wait.WaitPlan && kk == k} + for _, e := range exps { + if e.gap > idleSeconds && e.gap <= CoverageSecondsFor(idleSeconds, kk, e.ttl) { + row.Convertible++ + row.ConvertibleUSD += e.usd + } + } + for _, sp := range spans { + row.Pings += int64(sp.pings(idleSeconds, kk)) + } + finish(&row) out.Rows = append(out.Rows, row) } out.KeepAliveCoverage = *cov @@ -1587,6 +1769,12 @@ const ( type KeepAliveRecommendation struct { // Refused is the reason, when there is one. Its presence is the branch. Refused string `json:"refused,omitempty"` + // Plan says which model the interval is for: "wait_plan" (the live per-wait plan, MaxPings is its + // base) or "fixed_k" (the account's keepalive_wait_plan is off, or the old fixed policy). + Plan string `json:"plan"` + // FixedK is the same window replayed under a fixed K = MaxPings, for comparison with the plan. + // Present only for a "wait_plan" recommendation. A range, like the main one. + FixedK *RecInterval `json:"fixed_k,omitempty"` // IdleSeconds and MaxPings are the recommendation. K = 1 is NEVER returned, and the reason // is REACH, not a loss: coverage is K*X + TTL, so one ping reaches 580 s against two pings' // 860 s. The "4.7 minutes / -$71 service-wide" this comment used to give is the K*X coverage @@ -1613,8 +1801,15 @@ type KeepAliveRecommendation struct { KeepAliveCoverage } -// The fixed-count policy the recommendation models (240 s, K=2): the policy with keepalive_wait_plan off. -// The live default picks its count per wait (#858) and is not modelled here. +// RecInterval is a 90% range in dollars. Never a point. +type RecInterval struct { + LoUSD float64 `json:"lo_usd"` + HiUSD float64 `json:"hi_usd"` +} + +// The base policy the recommendation models (240 s, K=2). With the per-wait plan on (#868) K=2 is the +// plan's base and the interval is the plan's; recMaxPings alone is the fixed-count policy, the one with +// keepalive_wait_plan off. const ( recIdleSeconds = 240 recMaxPings = 2 @@ -1629,22 +1824,25 @@ const ( ) // KeepAliveRecommend answers "what should I set?" — or refuses. -func (d *DB) KeepAliveRecommend(f Filter) (*KeepAliveRecommendation, error) { - out := &KeepAliveRecommendation{ServiceLoUSD: serviceLoUSD, ServiceHiUSD: serviceHiUSD} +// +// wait is the account's policy (#868). With WaitPlan on the interval is the live plan's (base +// recMaxPings) and the fixed-K interval comes beside it; with WaitPlan off the interval is the +// fixed-K one, as before #858. +func (d *DB) KeepAliveRecommend(f Filter, wait keepalive.Policy) (*KeepAliveRecommendation, error) { + wait.MaxPings = recMaxPings + out := &KeepAliveRecommendation{ServiceLoUSD: serviceLoUSD, ServiceHiUSD: serviceHiUSD, Plan: PlanFixed} + if wait.WaitPlan { + out.Plan = PlanWait + } // The four reads below are independent and three of them are a pass over every request, so they // run together; the refusals further down still decide what is reported, just after the reads // rather than between them. - type miss struct { - gap, usd float64 - ttl time.Duration // the expiry's own cache lifetime (missLifetime) - } var ( cov *KeepAliveCoverage pingUSD float64 spans []pingSpan - bySession = map[string][]miss{} + bySession = map[string][]expiry{} ) - aCond, aArgs := addressable(f) var g errgroup.Group g.Go(func() error { var err error @@ -1653,23 +1851,11 @@ func (d *DB) KeepAliveRecommend(f Filter) (*KeepAliveRecommendation, error) { }) // The account's own addressable expiries, grouped by session — the resampling unit. g.Go(func() error { - rows, err := d.sql.QueryContext(d.readCtx(), aCond+` - SELECT session_id, gap_s, cost_usd, provider, model FROM addressable`, aArgs...) - if err != nil { - return err - } - defer rows.Close() - for rows.Next() { - var s string - var m miss - var prov, mod string - if err := rows.Scan(&s, &m.gap, &m.usd, &prov, &mod); err != nil { - return err - } - m.ttl = missLifetime(prov, mod) - bySession[s] = append(bySession[s], m) + exps, err := d.expiries(f, wait.WaitPlan) + for _, e := range exps { + bySession[e.session] = append(bySession[e.session], e) } - return rows.Err() + return err }) // The per-session ping cost of the SAME policy, so the resample scores a net and not a // saving. Ping cost is derived from the account's own median back-derived input rate, @@ -1683,7 +1869,7 @@ func (d *DB) KeepAliveRecommend(f Filter) (*KeepAliveRecommendation, error) { }) g.Go(func() error { var err error - spans, err = d.pingSpans(f, kaGateMinPrefix) + spans, err = d.pingSpans(f, kaGateMinPrefix, wait.WaitPlan) return err }) if err := g.Wait(); err != nil { @@ -1731,19 +1917,32 @@ func (d *DB) KeepAliveRecommend(f Filter) (*KeepAliveRecommendation, error) { keys = append(keys, s) } sort.Strings(keys) // deterministic draw order for a given seed - net := func(s string, k int) float64 { + // net is one session's value under a model: planned says "the per-wait plan with base k", else + // "k pings for every wait". + net := func(s string, k int, planned bool) float64 { + pol := wait + pol.MaxPings = k var v float64 for _, m := range bySession[s] { - if m.gap > recIdleSeconds && m.gap <= CoverageSecondsFor(recIdleSeconds, k, m.ttl) { + if planned { + if covers(recIdleSeconds, pol.SpanPings(m.facts), m.gap, m.ttl) { + v += m.usd * recRecoverableShare + } + } else if m.gap > recIdleSeconds && m.gap <= CoverageSecondsFor(recIdleSeconds, k, m.ttl) { v += m.usd * recRecoverableShare } } for _, sp := range bySpan[s] { - v -= float64(sp.pings(recIdleSeconds, k)) * pingUSD + if planned { + v -= float64(sp.planPings(recIdleSeconds, pol)) * pingUSD + } else { + v -= float64(sp.pings(recIdleSeconds, k)) * pingUSD + } } return v } - lo, hi := bootstrapCI(keys, func(s string) float64 { return net(s, recMaxPings) }) + planned := wait.WaitPlan + lo, hi := bootstrapCI(keys, func(s string) float64 { return net(s, recMaxPings, planned) }) // Condition 3, and it is a ONE-SIDED test. An interval that spans zero cannot tell the // effect from nothing; an interval entirely BELOW zero is worse than inconclusive — this // account's own gaps say the mechanism would cost it money — and both are refusals. The @@ -1766,11 +1965,15 @@ func (d *DB) KeepAliveRecommend(f Filter) (*KeepAliveRecommendation, error) { } out.IdleSeconds, out.MaxPings = recIdleSeconds, recMaxPings out.LoUSD, out.HiUSD = lo, hi + if planned { + flo, fhi := bootstrapCI(keys, func(s string) float64 { return net(s, recMaxPings, false) }) + out.FixedK = &RecInterval{LoUSD: flo, HiUSD: fhi} + } // K=3 is offered only where it beats K=2 by more than this account's OWN noise — its // interval has to clear K=2's entirely. Anything weaker than that is the comparison this // project has got wrong five times: two numbers whose intervals overlap are a tie, and on // the production corpus K=2 and K=3 ARE a tie for every account. The panel says so. - if lo3, _ := bootstrapCI(keys, func(s string) float64 { return net(s, 3) }); lo3 > hi { + if lo3, _ := bootstrapCI(keys, func(s string) float64 { return net(s, 3, planned) }); lo3 > hi { out.AltMaxPings = 3 } return out, nil diff --git a/dash/keepaliveapi.go b/dash/keepaliveapi.go index 756690974..bda965e4c 100644 --- a/dash/keepaliveapi.go +++ b/dash/keepaliveapi.go @@ -4,8 +4,23 @@ import ( "net/http" "strconv" "time" + + "github.com/rossoctl/context-guru/keepalive" ) +// SetWaitPolicy supplies an account's per-wait ping plan settings (keepalive_wait_plan and its three +// counts) for the calculator and the recommendation. dash holds no tenant configuration of its own, +// so the host answers. nil: the shipped plan, on. +func (a *API) SetWaitPolicy(fn func(tenantID string) keepalive.Policy) { a.waitPolicyFn = fn } + +// waitPolicy is the plan settings of the caller's account. The base count is set by the caller. +func (a *API) waitPolicy(tenantID string) keepalive.Policy { + if a.waitPolicyFn == nil { + return DefaultWaitPolicy(0) + } + return a.waitPolicyFn(tenantID) +} + // The keep-alive tab's six read routes. // // All in API.routes(), which is the table Mount walks AND the table both scoping tests walk — @@ -137,7 +152,7 @@ func (a *API) keepAliveCalc(w http.ResponseWriter, r *http.Request) { model = m } } - out, err := db.KeepAliveCalc(f, x, prefix, model, a.priceFn(r), k) + out, err := db.KeepAliveCalc(f, x, prefix, model, a.priceFn(r), k, a.waitPolicy(f.Tenant)) if err != nil { httpErr(w, http.StatusInternalServerError, err.Error()) return @@ -177,7 +192,7 @@ func (a *API) keepAliveRecommend(w http.ResponseWriter, r *http.Request) { a.unauthorized(w) return } - rec, err := a.db(r).KeepAliveRecommend(f) + rec, err := a.db(r).KeepAliveRecommend(f, a.waitPolicy(f.Tenant)) if err != nil { httpErr(w, http.StatusInternalServerError, err.Error()) return diff --git a/dash/keepaliveplan_test.go b/dash/keepaliveplan_test.go new file mode 100644 index 000000000..37466676a --- /dev/null +++ b/dash/keepaliveplan_test.go @@ -0,0 +1,302 @@ +package dash + +import ( + "context" + "fmt" + "net/http" + "testing" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + + "github.com/rossoctl/context-guru/dispatch" + "github.com/rossoctl/context-guru/internal/modelinfo" + "github.com/rossoctl/context-guru/internal/waitcause" + "github.com/rossoctl/context-guru/keepalive" +) + +// The dashboard replays the per-wait ping plan (#868). These tests keep that replay equal to the keeper. + +// nopDispatcher is a keeper host that never sends: the drift test only reads the plan. +type nopDispatcher struct{ next int } + +func (d *nopDispatcher) Do(context.Context, dispatch.Request) (dispatch.Response, error) { + return dispatch.Response{}, dispatch.ErrNotHeld +} +func (d *nopDispatcher) Release(dispatch.Ref) {} +func (d *nopDispatcher) Price(context.Context, string, string, int64) (dispatch.Price, bool) { + return dispatch.Price{}, false +} + +const planBody = `{"model":"aws/claude-sonnet-5","max_tokens":32000,"stream":true,` + + `"tools":[{"name":"Bash","description":"run a command","input_schema":{"type":"object"}}],` + + `"system":[{"type":"text","text":"you are claude code","cache_control":{"type":"ephemeral"}}],` + + `"messages":[{"role":"user","content":[{"type":"text","text":"hi","cache_control":{"type":"ephemeral"}}]}]}` + +// turn is one finished response of a thread. +type turn struct { + names []string + stop string +} + +var ( + tCall = turn{[]string{"Bash"}, "tool_use"} + tEnd = turn{nil, "end_turn"} +) + +// keeperPings is the ping count the KEEPER plans for the wait after the last of seq, on thread. +// 0 when the keeper plans no ping (no tool yet, done). +func keeperPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) int { + t.Helper() + now := time.Date(2026, 8, 22, 12, 0, 0, 0, time.UTC) + d := &nopDispatcher{} + k := keepalive.New(keepalive.Options{Dispatcher: d, Now: func() time.Time { return now }, + Go: func(f func()) { f() }, + OpenTasks: func(string, string, string) (int, int) { return bg, tm }}) + pol := keepalive.Policy{KeepAlive: true, Idle: 280 * time.Second, MaxPings: base, + MaxUSDPerPing: 0.25, MinPrefixTokens: 20000, WaitPlan: true} + key := keepalive.Key{Tenant: "t1", Session: "s1", Thread: thread} + for i, r := range seq { + hdr := http.Header{} + hdr.Set("Authorization", "Bearer sk-x") + hdr.Set("Anthropic-Version", "2023-06-01") + k.Record(keepalive.Record{Key: key, StartedAt: now.Add(-time.Duration(len(seq)-1-i) * time.Second), + Body: []byte(planBody), Provider: bschemas.Anthropic, Route: "/v1/messages", Status: 200, + Usage: keepalive.Usage{CacheRead: 48576, CacheWrite: 1200, StopReason: r.stop}, + ToolUseNames: r.names, UsageOK: true, Policy: pol, Header: hdr, + Hold: func() dispatch.Ref { d.next++; return dispatch.Ref(fmt.Sprint(d.next)) }}) + } + return k.Plans("t1", false, "s1")[thread].PingsLeft +} + +// dashPings is the count the DASHBOARD replay gives the same thread: it stores the rows the proxy +// would write (the wait cause comes from the same classifier), reads them back through pingSpans, +// and asks the span for its count. The last span is open, so it gets its full count. +func dashPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) int { + t.Helper() + db := openTestDB(t) + var evs []*Event + for i, r := range seq { + e := kaAgent(1_000_000+int64(i)*1000, "s1", 0.01) + e.CacheRead, e.CacheWrite, e.ThreadID, e.StopReason = 48576, 1200, thread, r.stop + // What the proxy stores: the classifier's answer for this response. The last row's open-task + // counts are the ones the keeper's OpenTasks hook reports. + b, m := 0, 0 + if i == len(seq)-1 { + b, m = bg, tm + } + e.WaitingFor, e.WaitTool = waitcause.ClassifyWaitingFor(r.names, r.stop, false, true, b, m) + evs = append(evs, e) + } + if err := db.insertBatch(evs); err != nil { + t.Fatal(err) + } + spans, err := db.pingSpans(Filter{TenantAll: true}, kaGateMinPrefix, true) + if err != nil { + t.Fatal(err) + } + pol := keepalive.Policy{WaitPlan: true, MaxPings: base} + n := 0 + for _, sp := range spans { + if sp.open { + n = sp.planPings(280, pol) + } + } + return n +} + +// TestDashboardReplayMatchesTheKeeper pins the dashboard's plan to the keeper's: the same responses +// give the same ping count through both paths. It fails when either side changes a rule alone. +func TestDashboardReplayMatchesTheKeeper(t *testing.T) { + sub := "a:abc" + ask := turn{[]string{"AskUserQuestion"}, "tool_use"} + for _, base := range []int{2, 10} { + for _, tc := range []struct { + name string + thread string + seq []turn + bg, tm int + }{ + {"main, own tool", "", []turn{tCall, tCall}, 0, 0}, + {"subagent, own tool", sub, []turn{tCall, tCall}, 0, 0}, + {"main, end_turn, background", "", []turn{tCall, tEnd}, 1, 0}, + {"main, end_turn, a person", "", []turn{tCall, tEnd}, 0, 0}, + {"main, ask-a-person tool", "", []turn{tCall, ask}, 0, 0}, + {"main, end_turn, teammate", "", []turn{tCall, tEnd}, 0, 1}, + {"subagent, end_turn", sub, []turn{tCall, tEnd}, 0, 0}, + {"subagent, end_turn, background", sub, []turn{tCall, tEnd}, 1, 0}, + {"subagent, handback", sub, []turn{tCall, {[]string{"SubagentHandback"}, "tool_use"}}, 0, 0}, + {"done", "", []turn{tCall, {[]string{"StructuredOutput"}, "tool_use"}}, 0, 0}, + {"no tool call yet", "", []turn{tEnd, tEnd}, 0, 0}, + {"no tool call yet, background open", "", []turn{tEnd, tEnd}, 1, 0}, + {"earlier tool call counts", "", []turn{tCall, tEnd, tEnd}, 0, 0}, + {"tool stop, no names, background open", "", []turn{tCall, {nil, "tool_use"}}, 1, 0}, + } { + t.Run(fmt.Sprintf("%s/base%d", tc.name, base), func(t *testing.T) { + want := keeperPings(t, base, tc.thread, tc.bg, tc.tm, tc.seq) + got := dashPings(t, base, tc.thread, tc.bg, tc.tm, tc.seq) + if got != want { + t.Errorf("dashboard replay = %d pings, keeper = %d", got, want) + } + }) + } + } +} + +// One known gap, stated and pinned (not hidden): a tool stop with NO tool name and nothing open is stored +// as a wait for a person, as an ask-a-person tool is. The row cannot tell the two apart. The keeper reads +// the first as a tool wait (8 pings), the replay as a person (base + 1). The case needs a spliced +// sniff window, which is rare. +func TestDashboardReplayKnownGapSplicedToolStop(t *testing.T) { + seq := []turn{tCall, {nil, "tool_use"}} + if got := keeperPings(t, 2, "", 0, 0, seq); got != 8 { + t.Fatalf("keeper = %d, want 8", got) + } + if got := dashPings(t, 2, "", 0, 0, seq); got != 3 { + t.Errorf("dashboard replay = %d, want 3 (base + 1)", got) + } +} + +func planPrice(string) (modelinfo.Price, bool) { return ibmSonnet, true } + +// planFixture: session "a" has a tool wait that a fixed K=2 cuts short; session "b" has a thread that +// never called a tool, which the plan does not ping. Two expiries show the same split on reach. +func planFixture(t *testing.T) *kaFixture { + t.Helper() + row := func(ts int64, sess, stop, wf string) *Event { + e := kaAgent(ts, sess, 0.05) + e.CacheRead, e.CacheWrite, e.StopReason, e.WaitingFor = 100_000, 0, stop, wf + return e + } + sec := func(s int64) int64 { return s * 1000 } + t0 := int64(10_000_000) + evs := []*Event{ + row(t0, "a", "tool_use", "own_tool"), + row(t0+sec(60), "a", "tool_use", "own_tool"), + row(t0+sec(1560), "a", "end_turn", "human"), + row(t0, "b", "end_turn", "human"), + row(t0+sec(60), "b", "end_turn", "human"), + } + // E1: a tool wait of 1000 s (reached by 8 pings, not by 2). E2: a thread with no tool call yet, 700 s. + e1 := kaExpiry(t0+sec(20_000), "e1", 1000, 1.0, 100_000) + e1[0].StopReason, e1[0].WaitingFor = "tool_use", "own_tool" + e2 := kaExpiry(t0+sec(40_000), "e2", 700, 1.0, 100_000) + e2[0].StopReason, e2[0].WaitingFor = "end_turn", "human" + evs = append(evs, e1...) + evs = append(evs, e2...) + return newKAFixture(t, evs...) +} + +func rungOf(t *testing.T, c *KeepAliveCalc, plan string, k int) CalcRow { + t.Helper() + for _, r := range c.Rows { + if r.Plan == plan && r.MaxPings == k { + return r + } + } + t.Fatalf("no %s rung with K=%d in %+v", plan, k, c.Rows) + return CalcRow{} +} + +func TestCalculatorHasALivePlanRungMarkedCurrent(t *testing.T) { + fx := planFixture(t) + on := keepalive.Policy{WaitPlan: true} + c, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 280, 100_000, "aws/claude-sonnet-5", planPrice, 2, on) + if err != nil { + t.Fatal(err) + } + if c.Plan != PlanWait || len(c.Rows) != 5 { + t.Fatalf("plan = %q, rows = %d, want wait_plan and 5 (the plan, then K=1..4)", c.Plan, len(c.Rows)) + } + plan, fixed2 := rungOf(t, c, PlanWait, 2), rungOf(t, c, PlanFixed, 2) + if !plan.Current || fixed2.Current { + t.Errorf("current: plan %v, fixed K=2 %v; want the plan only", plan.Current, fixed2.Current) + } + // Session a: a 1500 s tool wait gets floor(1220/280)+1 = 5 pings (fixed: 2), and the last request is a + // person (base+1 = 3; fixed: 2). Session b never called a tool: 0 (fixed: 2). The e1/e2 sessions add + // their own spans, the same on both sides of this comparison only where noted below. + if plan.Pings == fixed2.Pings { + t.Errorf("plan pings = fixed pings = %d; the plan should differ from fixed K here", plan.Pings) + } + // Reach: E1 (tool wait, 1000 s) only the plan reaches; E2 (no tool yet, 700 s) only the fixed K=2 does. + if plan.Convertible != 1 || fixed2.Convertible != 1 { + t.Errorf("convertible: plan %d, fixed K=2 %d, want 1 and 1", plan.Convertible, fixed2.Convertible) + } + if rungOf(t, c, PlanFixed, 3).Convertible != 2 { + t.Errorf("fixed K=3 should reach both expiries") + } +} + +// The off switch: an account with keepalive_wait_plan false gets fixed-K figures, as before. +func TestCalculatorWithThePlanOffModelsFixedK(t *testing.T) { + fx := planFixture(t) + c, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 280, 100_000, "aws/claude-sonnet-5", planPrice, 2, keepalive.Policy{}) + if err != nil { + t.Fatal(err) + } + if c.Plan != PlanFixed || len(c.Rows) != 4 { + t.Fatalf("plan = %q, rows = %d, want fixed_k and 4", c.Plan, len(c.Rows)) + } + for _, r := range c.Rows { + if r.Plan != PlanFixed || r.Current != (r.MaxPings == 2) { + t.Errorf("rung %+v: want fixed_k, current only at K=2", r) + } + } +} + +func TestRecommendationStatesItsModelAndTheFixedKComparison(t *testing.T) { + fx := newKAFixture(t) + rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}, keepalive.Policy{WaitPlan: true}) + if err != nil { + t.Fatal(err) + } + if rec.Plan != PlanWait { + t.Errorf("plan = %q, want wait_plan", rec.Plan) + } + off, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}, keepalive.Policy{}) + if err != nil { + t.Fatal(err) + } + if off.Plan != PlanFixed || off.FixedK != nil { + t.Errorf("plan off: plan = %q, fixed_k = %v; want fixed_k and no comparison", off.Plan, off.FixedK) + } +} + +// The plan replay adds the facts window to two reads. It must stay near the cost of the fixed-K reads. +// Relative bound, as in zz_perf_test.go: an absolute bound is a machine away from meaningless. +func TestPlanReplayStaysNearTheFixedKCost(t *testing.T) { + db := openTestDB(t) + const n = 6_000 + batch := make([]*Event, 0, 1000) + for i := 0; i < n; i++ { + e := kaAgent(int64(1_700_000_000_000+i*100_000), fmt.Sprintf("s%d", i/40), 0.01) + e.ThreadID = map[bool]string{true: "a:x", false: ""}[i%3 == 0] + e.CacheRead, e.CacheWrite, e.StopReason, e.WaitingFor = 90_000, 100, "tool_use", "own_tool" + if i%7 == 0 { + e.CacheMissReason, e.CacheRead, e.CacheWrite = CacheTTLExpiry, 0, 90_000 + } + batch = append(batch, e) + if len(batch) == 1000 { + if err := db.insertBatch(batch); err != nil { + t.Fatal(err) + } + batch = batch[:0] + } + } + f := Filter{TenantAll: true} + run := func(plan bool) time.Duration { + t0 := time.Now() + if _, err := db.pingSpans(f, kaGateMinPrefix, plan); err != nil { + t.Fatal(err) + } + if _, err := db.expiries(f, plan); err != nil { + t.Fatal(err) + } + return time.Since(t0) + } + fixed, plan := run(false), run(true) + t.Logf("%d rows: fixed-K reads %v, plan reads %v", n, fixed, plan) + if plan > 6*fixed+2*time.Second { + t.Errorf("plan reads %v against fixed-K reads %v", plan, fixed) + } +} diff --git a/dash/keepalivetab_test.go b/dash/keepalivetab_test.go index 743ff9cfe..9cd29f8f6 100644 --- a/dash/keepalivetab_test.go +++ b/dash/keepalivetab_test.go @@ -9,6 +9,7 @@ import ( "testing" "github.com/rossoctl/context-guru/internal/modelinfo" + "github.com/rossoctl/context-guru/keepalive" ) // The keep-alive tab's arithmetic and its attribution, and the tests are mostly about the @@ -424,11 +425,11 @@ func TestCalculatorPricesFromTheRowsModelNotABlend(t *testing.T) { } return modelinfo.Price{}, false } - a, err := fx.db.KeepAliveCalc(f, 280, 300_000, "cheap", price, 2) + a, err := fx.db.KeepAliveCalc(f, 280, 300_000, "cheap", price, 2, keepalive.Policy{}) if err != nil { t.Fatal(err) } - b, err := fx.db.KeepAliveCalc(f, 280, 300_000, "dear", price, 2) + b, err := fx.db.KeepAliveCalc(f, 280, 300_000, "dear", price, 2, keepalive.Policy{}) if err != nil { t.Fatal(err) } @@ -446,7 +447,7 @@ func TestCalculatorPricesFromTheRowsModelNotABlend(t *testing.T) { "miss", a.AvoidedUSDEach) } // The unpriced model: no dollar figure at all, and it says so. - un, err := fx.db.KeepAliveCalc(f, 280, 300_000, "nobody-prices-me", price, 2) + un, err := fx.db.KeepAliveCalc(f, 280, 300_000, "nobody-prices-me", price, 2, keepalive.Policy{}) if err != nil { t.Fatal(err) } @@ -481,7 +482,7 @@ func TestCalculatorGoldenAgainstTheProductionBands(t *testing.T) { } fx := newKAFixture(t, evs...) price := func(string) (modelinfo.Price, bool) { return ibmSonnet, true } - out, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 280, 300_000, "aws/claude-sonnet-5", price, 2) + out, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 280, 300_000, "aws/claude-sonnet-5", price, 2, keepalive.Policy{}) if err != nil { t.Fatal(err) } @@ -507,14 +508,17 @@ func TestCalculatorGoldenAgainstTheProductionBands(t *testing.T) { if g3 > g1 { t.Errorf("the ladder does not flatten: K1->K2 gains %d, K3->K4 gains %d", g1, g3) } - // No rung is marked current (#858): the live policy picks its count per wait, and the ladder - // models the fixed-K (plan off) policy, which it says on the wire. + // With the plan off (the zero Policy), the model is fixed K, every rung says so, and the rung with + // K = the account's base is the current one (#868). for _, r := range out.Rows { - if r.Current { - t.Errorf("K=%d is marked current; no fixed-K rung is the live policy", r.MaxPings) + if r.Plan != PlanFixed { + t.Errorf("K=%d plan = %q, want fixed_k", r.MaxPings, r.Plan) + } + if r.Current != (r.MaxPings == 2) { + t.Errorf("K=%d current = %v", r.MaxPings, r.Current) } } - if out.Plan != "fixed_k" { + if out.Plan != PlanFixed { t.Errorf("plan = %q, want fixed_k", out.Plan) } } @@ -563,7 +567,7 @@ func TestKeepAlivePanelsStateTheirCoverage(t *testing.T) { // below it. func TestRecommendationRefusesThinHistory(t *testing.T) { fx := newKAFixture(t, kaSpread(t, 19, 700, 1.0)...) - rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}) + rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}, keepalive.Policy{}) if err != nil { t.Fatal(err) } @@ -591,7 +595,7 @@ func TestRecommendationRefusesWhenTheIntervalCrossesZero(t *testing.T) { // any coverage at X=280 — so nothing converts, the pings cost money, and the interval sits // on or below zero rather than excluding it above. fx := newKAFixture(t, kaSpread(t, 30, 40_000, 1.0)...) - rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}) + rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}, keepalive.Policy{}) if err != nil { t.Fatal(err) } @@ -614,7 +618,7 @@ func TestRecommendationRefusesAnIntervalEntirelyBelowZero(t *testing.T) { // 30 expiries whose gaps are far past any coverage at X=280 (2.8 hours), on sessions that // nevertheless have plenty of idle spans to be pinged through. Cost with no conversion. fx := newKAFixture(t, kaSpread(t, 30, 10_000, 2.0)...) - rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}) + rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}, keepalive.Policy{}) if err != nil { t.Fatal(err) } @@ -641,7 +645,7 @@ func TestRecommendationRefusesAnIntervalEntirelyBelowZero(t *testing.T) { // does not exist, because a field that exists gets rendered. func TestRecommendationPayloadCarriesNoPointEstimate(t *testing.T) { fx := newKAFixture(t, kaSpread(t, 25, 700, 1.0)...) - rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}) + rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}, keepalive.Policy{}) if err != nil { t.Fatal(err) } @@ -678,7 +682,7 @@ func TestRecommendationNeverSuggestsOnePing(t *testing.T) { gap float64 }{{25, 400}, {40, 700}, {25, 1100}, {30, 5000}} { fx := newKAFixture(t, kaSpread(t, tc.n, tc.gap, 1.0)...) - rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}) + rec, err := fx.db.KeepAliveRecommend(Filter{TenantAll: true}, keepalive.Policy{}) if err != nil { t.Fatal(err) } @@ -841,7 +845,7 @@ func TestTheCalculatorChargesThePingsAPolicyWouldActuallySend(t *testing.T) { calc, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 280, 100_000, "aws/claude-sonnet-5", func(string) (modelinfo.Price, bool) { return modelinfo.Price{Input: 3e-6, Output: 15e-6, CacheRead: 3e-7, CacheWrite: 3.75e-6}, true - }, 2) + }, 2, keepalive.Policy{}) if err != nil { t.Fatal(err) } diff --git a/dash/testdata/golden/ka_calc.json b/dash/testdata/golden/ka_calc.json index 7505ddf8c..af9e75303 100644 --- a/dash/testdata/golden/ka_calc.json +++ b/dash/testdata/golden/ka_calc.json @@ -5,18 +5,29 @@ "keepalive_recorded_from": 1767606380000, "keepalive_recorded_rows": 585, "model": "m-unpriced", - "plan": "fixed_k", + "plan": "wait_plan", "prefix_source": "account_median", "prefix_tokens": 30998, "priced": false, "requests": 591, "rows": [ + { + "convertible_misses": 112, + "convertible_usd": 0.6000000000000004, + "coverage_seconds": 860, + "current": true, + "max_pings": 2, + "pings": 433, + "plan": "wait_plan", + "share_of_addressable_pct": 70.75471698113208 + }, { "convertible_misses": 112, "convertible_usd": 0.6000000000000004, "coverage_seconds": 580, "max_pings": 1, "pings": 281, + "plan": "fixed_k", "share_of_addressable_pct": 70.75471698113208 }, { @@ -25,6 +36,7 @@ "coverage_seconds": 860, "max_pings": 2, "pings": 433, + "plan": "fixed_k", "share_of_addressable_pct": 70.75471698113208 }, { @@ -33,6 +45,7 @@ "coverage_seconds": 1140, "max_pings": 3, "pings": 585, + "plan": "fixed_k", "share_of_addressable_pct": 70.75471698113208 }, { @@ -41,6 +54,7 @@ "coverage_seconds": 1420, "max_pings": 4, "pings": 737, + "plan": "fixed_k", "share_of_addressable_pct": 70.75471698113208 } ] diff --git a/dash/testdata/golden/ka_recommend.json b/dash/testdata/golden/ka_recommend.json index 91f3e7a92..249020d1c 100644 --- a/dash/testdata/golden/ka_recommend.json +++ b/dash/testdata/golden/ka_recommend.json @@ -1,4 +1,8 @@ { + "fixed_k": { + "hi_usd": 0.33678400000000003, + "lo_usd": 0.22627199999999986 + }, "hi_usd": 0.33678400000000003, "idle_seconds": 240, "keepalive_recorded_from": 1767606380000, @@ -6,6 +10,7 @@ "lo_usd": 0.22627199999999986, "max_pings": 2, "n": 204, + "plan": "wait_plan", "requests": 591, "service_hi_usd": 237, "service_lo_usd": 95, diff --git a/dash/ui/app.js b/dash/ui/app.js index cdb6ca74f..a9f73d150 100644 --- a/dash/ui/app.js +++ b/dash/ui/app.js @@ -968,7 +968,7 @@ const TILE_INFO = { }, 'ka-rec-policy': { what: 'The idle interval and ping count suggested for this account.', - how: 'A fixed policy (240 s, 2 pings, the setting with the per-wait plan off), offered only when your own history can tell its effect apart ' + how: 'The live policy (240 s, base of 2 pings, with the per-wait plan on unless your account switched it off), offered only when your own history can tell its effect apart ' + 'from zero: at least 20 addressable cache expiries, at least 200 requests, and a 90% ' + 'interval that excludes no-change.', catch: 'One ping is never suggested. Not because it loses money \u2014 it does not \u2014 ' @@ -9516,6 +9516,19 @@ function renderKACalcControls() { el('label', { for: 'ka-k' }, 'Max pings (K)'), k, apply)); } +/** kaRungName names a ladder rung by its model: the live plan, or a fixed count. */ +function kaRungName(r) { + const name = r.plan === 'wait_plan' ? 'Live plan, base ' + r.max_pings : 'Fixed ' + r.max_pings; + return name + (r.current ? ' (current)' : ''); +} + +/** kaRungCoverage says how far a rung reaches. The plan's reach depends on the wait, so it has no single figure. */ +function kaRungCoverage(r) { + return r.plan === 'wait_plan' + ? 'by wait: a tool up to 8, a person base+1' + : r.coverage_seconds.toFixed(0) + 's = ' + r.max_pings + '×' + kaState.x + ' + 300'; +} + /** loadKACalc renders the K ladder. */ async function loadKACalc() { renderKACalcControls(); @@ -9551,7 +9564,7 @@ async function loadKACalc() { 'the dollar columns are blank. They are not zero.')); const tbl = el('table', { class: 'grid', 'data-testid': 'ka-ladder' }, el('thead', {}, el('tr', {}, - el('th', {}, 'Max pings (K)'), el('th', {}, 'Cache stays alive'), + el('th', {}, 'Policy (K)'), el('th', {}, 'Cache stays alive'), el('th', { class: 'num' }, 'Your expiries inside it'), el('th', { class: 'num' }, 'Their cost'), // The value is convertible_usd / addressable_usd — a share of the DOLLARS, not of the @@ -9565,8 +9578,8 @@ async function loadKACalc() { // Emphasis by weight, not by a new hue: the current row is the accent, the rest recede. tb.appendChild(el('tr', { class: r.current ? 'is-current' : 'muted-row', 'data-testid': 'ka-ladder-k' + r.max_pings }, - el('td', {}, String(r.max_pings) + (r.current ? ' (current)' : '')), - el('td', {}, r.coverage_seconds.toFixed(0) + 's = ' + r.max_pings + '×' + kaState.x + ' + 300'), + el('td', {}, kaRungName(r)), + el('td', {}, kaRungCoverage(r)), el('td', { class: 'num' }, num(r.convertible_misses)), el('td', { class: 'num' }, usd(r.convertible_usd)), el('td', { class: 'num' }, pct(r.share_of_addressable_pct, 1)), @@ -9588,25 +9601,32 @@ async function loadKACalc() { const share = el('div'); host.appendChild(share); barRows(share, c.rows.map((r) => ({ - label: 'K=' + r.max_pings + (r.current ? ' (current)' : ''), + label: kaRungName(r), value: r.share_of_addressable_pct, max: 100, display: pct(r.share_of_addressable_pct, 1) + ' · ' + num(r.pings) + ' pings', color: r.current ? 'var(--accent)' : KA_MUTED, title: `${r.convertible_misses} of ${c.addressable_misses} expiries reachable`, })), {}); host.appendChild(el('p', { class: 'hint' }, - 'Each row is one cap on pings. Reading across: how many of your expiries it would have caught, ' + + 'Each row is one policy. Reading across: how many of your expiries it would have caught, ' + 'what they cost, and what the pings cost. A replay of your past gaps, not a forecast. ' + - 'Catching grows sharply from one ping to two, then flattens while ping cost keeps climbing. ' + - 'Every rung is a fixed count. The live policy picks its count from what each thread waits for ' + - '(up to 8 for a tool wait), so no rung is marked as the current one.')); + (c.plan === 'wait_plan' + ? 'The first row is the live plan: it picks the count from what each thread waits for ' + + '(up to 8 for a tool wait, the base plus 3 for background work, the base plus 1 for a person, ' + + 'none before the thread has called a tool). The fixed rows send the same count for every wait, ' + + 'for comparison. ' + : 'Your account has the per-wait plan switched off, so every row is a fixed count and the row ' + + 'with your count is the current one. ') + + 'Catching grows sharply from one ping to two, then flattens while ping cost keeps climbing.')); host.appendChild(el('details', { class: 'why' }, el('summary', {}, 'Technical details'), el('p', {}, - 'PINGS is every ping a fixed-count policy would have sent over this window, gated as the ' + + 'PINGS is every ping the policy would have sent over this window, gated as the ' + 'request path gates (past the session\u2019s first request, cached prompt over 20k tokens) and ' + 'including the pings a session\u2019s last request attracts, because a live policy cannot know a ' + 'session has ended and sends its full K anyway. It counts only expiries that were ' + - 'addressable (the provider wrote a prefix that a ping could have refreshed).'))); + 'addressable (the provider wrote a prefix that a ping could have refreshed). The live-plan row ' + + 'reads the stop reason and wait cause the proxy stored for the request before each idle gap, and ' + + 'whether the thread had called a tool by then. It uses the same rule the proxy uses.'))); } /** @@ -9657,9 +9677,15 @@ async function loadKARecommend() { usd(rec.lo_usd) + ' – ' + usd(rec.hi_usd), `90% interval from ${num(rec.n)} expiries in ${num(rec.sessions)} sessions`, rec.lo_usd > 0 ? 'good' : 'bad'), - tile('ka-rec-policy', 'Suggested', rec.idle_seconds + 's / ' + rec.max_pings + ' pings', + tile('ka-rec-policy', 'Suggested', rec.idle_seconds + 's / ' + rec.max_pings + ' pings' + + (rec.plan === 'wait_plan' ? ' (base)' : ''), 'on sessions past their first request with a cached prompt over 20k tokens', 'accent'), ])); + if (rec.plan === 'wait_plan' && rec.fixed_k) { + host.appendChild(el('p', { class: 'note', 'data-testid': 'ka-rec-fixed' }, + `The live plan picks the count from each wait. For comparison, ${rec.max_pings} pings for every wait ` + + `would have given ${usd(rec.fixed_k.lo_usd)} – ${usd(rec.fixed_k.hi_usd)} over the same window.`)); + } host.appendChild(el('p', { class: 'note' }, 'The direction is resolvable; the size is not — a window like this one cannot pin it ', el('strong', {}, 'closer than about a factor of two'), diff --git a/docs/data-dependence.md b/docs/data-dependence.md index 50a69868c..209366da7 100644 --- a/docs/data-dependence.md +++ b/docs/data-dependence.md @@ -219,7 +219,7 @@ One row for each registered component. The test `components/all/datadep_test.go` | **Keep-alive pings** | content. The host keeps the credential. | ❌ **Should be off. Today it is still on**, and keeps the last body. The proxy binary also keeps the credential. | ❌ Off | ✅ Works | To keep the prompt cache warm, the proxy sends the exact same prompt again with the user's credential. Without the text, it cannot ping. The Ping button (`POST /api/me/cache/ping`) uses the same body and key. With keep-alive off, a request that carries the header `x-context-guru-ping: 1` is still held, for the button only: no ping is sent for it, and the body and key are released 300 seconds after the last request on a 5-minute cache, or 59 minutes after it on a 1-hour cache (a press does not extend it), and one account holds at most 16 such threads. Without the header nothing is held. The install adds the header and tells the user. An older install (from before `--no-ping-header`) gets it at its next session start, with the same notice. Any other install without the header is never changed. It gets a one-time note instead. The cache-state answer also carries the time of the next planned ping (a number, metadata), so the mod can hide the button while a ping fires before the cache expires. | | **Body stash** | content | ❌ **Should be off. Today it is still on**, with 2 full bodies per thread. | ❌ Off. The sweep's question then cannot be asked, and `cache_aware_summarizer` stops. | ✅ Works | To ask the model a cheap question, Context Guru adds the question to the exact text it sent last turn. The provider already cached that text. The sweep keeps **2 more full copies of the conversation per thread**, outside the stash, for a thread the keep-alive can ping (see `extract_llm_sweep` above). | | Observe-mode store | content | ❌ **Should follow the level. Today it is still a real Store.** | It should follow the level | ✅ Works | Observe mode runs the pipeline later on a copy, to estimate savings. The copy should store only what the real run would store. | -| Dashboard database | content and metadata | ❌ Off | 🟡 Works, metadata: numbers, fixed codes, ids, and tool, MCP-server and skill names. **No** before/after text, summaries, declaration text, error text or rejection text. | ✅ Works. It keeps text only when content capture is on. | The dashboard shows what Context Guru removed and what your setup carries. Every number on it needs only counts and prices. Every request row comes from traffic the proxy saw itself. On a local install the dashboard also reads two things from your disk, and stores neither: the working folder recorded in Claude Code's session files, to group folders into projects for the project picker, and each installed skill's description from its `SKILL.md`. An earlier history import copied whole request rows from those session files; it is removed, and its rows are deleted when the database opens. | +| Dashboard database | content and metadata | ❌ Off | 🟡 Works, metadata: numbers, fixed codes, ids, and tool, MCP-server and skill names. **No** before/after text, summaries, declaration text, error text or rejection text. | ✅ Works. It keeps text only when content capture is on. | The dashboard shows what Context Guru removed and what your setup carries. Every number on it needs only counts and prices. Every request row comes from traffic the proxy saw itself. On a local install the dashboard also reads two things from your disk, and stores neither: the working folder recorded in Claude Code's session files, to group folders into projects for the project picker, and each installed skill's description from its `SKILL.md`. The keep-alive calculator and recommendation (#868) replay the per-wait ping plan from columns the proxy already writes for each request: stop reason, wait cause, thread id, cache tier and provider. They read no text. An earlier history import copied whole request rows from those session files; it is removed, and its rows are deleted when the database opens. | | Dashboard memos | content | ❌ Off, with the dashboard | 🟡 Works, metadata: a hash and a token count. That is all they need. | ✅ Works. They keep declaration and system-prompt text. | They remember a tool list so they do not count its tokens again on every request. A hash and a count are enough. | | Archive | content and metadata | ❌ Off | 🟡 Works, metadata rows only | ✅ Works | It moves old conversations off the local disk. The database stays small, and you can still open old history. | | Insights | metadata | ❌ Off (it needs the dashboard) | 🟡 Works, metadata | 🟡 Works, metadata | A cost report needs only counts, tool names and prices, not the text. | diff --git a/docs/how-to/dashboard-guide.md b/docs/how-to/dashboard-guide.md index 26fe7c169..d55ec0ff2 100644 --- a/docs/how-to/dashboard-guide.md +++ b/docs/how-to/dashboard-guide.md @@ -82,7 +82,7 @@ Detailed diagnostics, opened: **How often your cache ran out.** Compact charts of the expiries: how many per day, what they cost, and how long you had been idle when they happened. -**What would other settings have earned?** A calculator and a recommendation. Change how long to wait before a ping and how many pings to send, and see what your past traffic would have cost. +**What would other settings have earned?** A calculator and a recommendation. Change how long to wait before a ping and how many pings to send, and see what your past traffic would have cost. The first row of the calculator is the live plan: the proxy picks the ping count from what each thread waits for. The fixed rows send the same count for every wait, for comparison. If your account sets `keepalive_wait_plan: false`, every row is a fixed count. ## Inventory diff --git a/keepalive/waitplan.go b/keepalive/waitplan.go index 7c7d5f1f0..96051cc5e 100644 --- a/keepalive/waitplan.go +++ b/keepalive/waitplan.go @@ -95,20 +95,82 @@ func (k *Keeper) WaitingFor(key Key, names []string, incomplete bool, stopReason // A response that stopped for a tool call waits for that tool, even when the name list missed // the call (a spliced sniff window): the classifier reads such a response as a wait for a person. // Whatever the classifier said (open background tasks read "background"), a tool stop gets the tool budget. - if len(names) == 0 && toolStop(stopReason) && w != waitcause.WaitingDone && w != waitcause.WaitingRetry { + return toolStopWait(w, len(names) == 0, stopReason) +} + +// toolStopWait applies the tool-stop rule: a response that stopped for a tool call and has no tool +// name waits for that tool, whatever the classifier said, unless the thread is done or the call failed. +func toolStopWait(w string, noNames bool, stopReason string) string { + if noNames && toolStop(stopReason) && w != waitcause.WaitingDone && w != waitcause.WaitingRetry { return waitcause.WaitingOwnTool } return w } +// ToolStopReasons are the stop reasons that mean "the response called a tool". The dashboard reads +// the same list when it looks for the first tool call of a thread in its stored rows. +var ToolStopReasons = []string{"tool_use", "tool_calls", "function_call"} + func toolStop(stopReason string) bool { - switch stopReason { - case "tool_use", "tool_calls", "function_call": - return true + for _, s := range ToolStopReasons { + if s == stopReason { + return true + } } return false } +// SpanFacts is what a stored request row says about the idle span that follows it: the facts the +// plan needs. The dashboard replays the plan from them (Policy.SpanPings). +type SpanFacts struct { + // Cause is the row's stored waiting_for ("" on an old row). + Cause string + // StopReason is the row's stored stop reason ("" on an old row). + StopReason string + // Subagent is true for a subagent thread (IsSubagentThread). + Subagent bool + // ToolSeen is true when this response or an earlier one of the thread called a tool. + ToolSeen bool + // OpenAI is true for an OpenAI request. The plan does not cover OpenAI: it keeps the base count. + OpenAI bool + // OneHour is true when the request asked for the 1-hour cache tier. That tier keeps its own + // schedule: the base count. + OneHour bool +} + +// storedWait is the wait cause the keeper would plan for, from a stored row. A stored row has no +// tool names, so the keeper's "tool stop with no name" rule is read from the cause: a tool stop that +// the classifier read as background work or a teammate waits for its tool. A tool stop stored as +// "human" stays human, because a row cannot tell an ask-a-person tool from a name list that missed +// the call (the keeper reads the second as a tool wait). That case is rare. +func storedWait(cause, stopReason string) string { + return toolStopWait(cause, cause != waitcause.WaitingHuman && cause != waitcause.WaitingOwnTool, stopReason) +} + +// SpanPings is the ping count the plan gives the idle span after a response, from the facts a stored +// row holds. It is the keeper's decision in one place, for a replay: +// +// - the count WaitPings gives for the wait cause; +// - 0 for a thread with no tool call yet (the no-tool rule); +// - 0 for a finished thread (waiting_for done) and for a failed call (retry): the keeper holds nothing for them; +// - MaxPings with WaitPlan off, for OpenAI, for the 1-hour tier, and for a row that stored neither a +// wait cause nor a stop reason (an old row: nothing to replay from). +// +// TestDashboardReplayMatchesTheKeeper (package dash) runs the keeper and this function on the same cases. +func (p Policy) SpanPings(s SpanFacts) int { + if !p.WaitPlan || s.OpenAI || (s.Cause == "" && s.StopReason == "") { + return p.MaxPings + } + w := storedWait(s.Cause, s.StopReason) + if w == waitcause.WaitingDone || w == waitcause.WaitingRetry || !s.ToolSeen { + return 0 + } + if s.OneHour { + return p.MaxPings + } + return p.WaitPings(w, s.Subagent) +} + // calledATool reports whether a response called a tool: a tool_use name or a tool stop reason. A // name list that may be incomplete (a spliced sniff window) counts as a call: the thread is then pinged as // before #858. diff --git a/proxy/keepalive.go b/proxy/keepalive.go index 96e6ad217..e37160968 100644 --- a/proxy/keepalive.go +++ b/proxy/keepalive.go @@ -546,3 +546,27 @@ func (h *Handler) KeepAlivePlans(tenantID string, allTenants bool, session strin } return out } + +// DashWaitPolicy is the dashboard's READ hook for an account's per-wait ping plan settings (#868): +// keepalive_wait_plan and the three counts. The calculator replays the plan with them. Wired in +// cmd/context-guru-proxy beside the keep-alive plans. +func (h *Handler) DashWaitPolicy() func(tenantID string) keepalive.Policy { + return func(id string) keepalive.Policy { + tn := h.static.Load() + if reg := h.registry(); reg != nil && h.opts.Tenants != nil { + t, err := reg.Get(id) + if err != nil { + return dash.DefaultWaitPolicy(0) + } + if tn, err = h.opts.Tenants.ForTenant(t); err != nil { + return dash.DefaultWaitPolicy(0) + } + } + if tn == nil { + return dash.DefaultWaitPolicy(0) + } + c := tn.Cache + return keepalive.Policy{WaitPlan: c.WaitPlan, ToolWaitMaxPings: c.ToolWaitMaxPings, + BackgroundExtraPings: c.BackgroundExtraPings, HumanExtraPings: c.HumanExtraPings} + } +} From 010dce8f19b7001b48a5843b78332147e8b4c2ee Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Tue, 3 Nov 2026 02:33:03 +0200 Subject: [PATCH 2/4] fix(dash): model the 1-hour tier, retry waits and the effective plan (#868) Round 1 fixes for the plan replay. - keepalive: SpanPings returns the count and the spacing. The 1-hour tier uses ForOneHour (MaxPings1h at Idle1h). Idle of 300 s or more keeps the base count, as the keeper does. - dash: the tier is the BILLED tier (ResolveTier rule: asked 1 hour and cache_write_1h >= cache_write, sticky per thread). The calculator states the plan counts, and the plan rung gives way to fixed K at 300 s. - proxy: DashWaitPolicy reports the effective plan. An active campaign or a strategy in the 1-hour band turns it off, and so does a failed account lookup. - Tests: 1-hour, mixed-tier and failed-call cases in the drift test. Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- dash/keepalive.go | 126 +++++++++++++++++++++-------- dash/keepaliveplan_test.go | 130 ++++++++++++++++++++++++------ dash/testdata/golden/ka_calc.json | 6 ++ dash/ui/app.js | 12 ++- keepalive/waitplan.go | 39 +++++---- proxy/dashwaitpolicy_test.go | 59 ++++++++++++++ proxy/keepalive.go | 41 ++++++++-- 7 files changed, 328 insertions(+), 85 deletions(-) create mode 100644 proxy/dashwaitpolicy_test.go diff --git a/dash/keepalive.go b/dash/keepalive.go index c3584cfcd..f5204c3fe 100644 --- a/dash/keepalive.go +++ b/dash/keepalive.go @@ -1291,10 +1291,12 @@ func (s pingSpan) pings(idleSeconds float64, maxPings int) int { return PingsPerSpan(s.gap, idleSeconds, maxPings) } -// planPings is how many pings this span attracts under the per-wait plan (#868). The count comes -// from keepalive.Policy.SpanPings, the keeper's own rule; the span's length caps it as in pings. -func (s pingSpan) planPings(idleSeconds float64, pol keepalive.Policy) int { - return s.pings(idleSeconds, pol.SpanPings(s.facts)) +// planPings is how many pings this span attracts under the per-wait plan (#868). The count and the +// spacing come from keepalive.Policy.SpanPings, the keeper's own rule (pol.Idle is the 5-minute spacing); +// the span's length caps the count as in pings. +func (s pingSpan) planPings(pol keepalive.Policy) int { + n, idle := pol.SpanPings(s.facts) + return s.pings(idle.Seconds(), n) } // toolStopList is keepalive.ToolStopReasons as a SQL list of string literals. The names are fixed @@ -1307,26 +1309,42 @@ func toolStopList() string { return strings.Join(q, ",") } -// factsCTE reads each request with the facts the plan needs. toolSeen is true when this request or an -// earlier one of the same THREAD called a tool: a tool stop reason, or a stored wait on its own tool. -// Pings are excluded by the filter, as in every replay. One extra window, partitioned by thread. +// factsCTE reads each request with the facts the plan needs. +// +// - seen is true when this request or an earlier one of the same THREAD called a tool: a tool stop +// reason, or a stored wait on its own tool. +// - hour is the BILLED tier, as keepalive.ResolveTier decides it: the request asked 1 hour +// (cache_ttl) and the last write of the thread (this row's own, or the latest before it) was billed +// entirely at 1 hour (cache_write_1h >= cache_write). A row that wrote nothing inherits. +// +// Pings are excluded by the filter, as in every replay. Two extra windows, partitioned by thread. func factsCTE(cond string) string { - return `t AS ( + return `a AS ( SELECT r.id AS id, r.tenant_id AS tenant_id, r.session_id AS session_id, r.thread_id AS thread_id, r.ts AS ts, r.cache_read + r.cache_write AS prefix, r.cache_write AS cache_write, r.cost_usd AS cost_usd, r.model AS model, r.provider AS provider, r.cache_miss_reason AS cache_miss_reason, r.cache_ttl AS cache_ttl, COALESCE(r.stop_reason, '') AS stop, COALESCE(r.waiting_for, '') AS wf, - MAX(CASE WHEN r.stop_reason IN (` + toolStopList() + `) OR r.waiting_for = 'own_tool' - THEN 1 ELSE 0 END) - OVER (PARTITION BY r.tenant_id, r.session_id, r.thread_id ORDER BY r.ts, r.id) AS seen - FROM requests r WHERE ` + cond + `)` + CASE WHEN r.stop_reason IN (` + toolStopList() + `) OR r.waiting_for = 'own_tool' THEN 1 ELSE 0 END AS tool, + CASE WHEN r.cache_write > 0 THEN + ROW_NUMBER() OVER (PARTITION BY r.tenant_id, r.session_id, r.thread_id ORDER BY r.ts, r.id) * 2 + + CASE WHEN r.cache_ttl = '` + apply.TTLEphemeral1h + `' AND r.cache_write_1h >= r.cache_write + THEN 1 ELSE 0 END + END AS wcode + FROM requests r WHERE ` + cond + `), + b AS ( + SELECT *, MAX(tool) OVER w AS seen, MAX(wcode) OVER w AS lastw + FROM a WINDOW w AS (PARTITION BY tenant_id, session_id, thread_id ORDER BY ts, id + ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)), + t AS ( + SELECT *, CASE WHEN cache_ttl = '` + apply.TTLEphemeral1h + `' AND COALESCE(lastw, 0) % 2 = 1 + THEN 1 ELSE 0 END AS hour FROM b)` } // scanFacts builds the SpanFacts of one stored row. -func scanFacts(thread, stop, wf, provider, cacheTTL string, seen int64) keepalive.SpanFacts { +func scanFacts(thread, stop, wf, provider string, seen, hour int64) keepalive.SpanFacts { return keepalive.SpanFacts{Cause: wf, StopReason: stop, Subagent: keepalive.IsSubagentThread(thread), - ToolSeen: seen > 0, OpenAI: provider == "openai", OneHour: cacheTTL == apply.TTLEphemeral1h} + ToolSeen: seen > 0, OpenAI: provider == "openai", OneHour: hour > 0} } // DefaultWaitPolicy is the shipped keep-alive policy for the replay: the per-wait plan on, with @@ -1354,9 +1372,9 @@ func DefaultWaitPolicy(base int) keepalive.Policy { // names, so it MOVES when the gate or K moves instead of silently ignoring both. // // The gate is the request path's, in the same order: proxy/keepalive.go's kaEntry.pingable(). It -// gates at 20k (the shipped default since #858) and not at the account's own configured floor, because this -// package does not read tenant configuration — an account that tuned its floor sees a replay of -// the default policy, and the panel's copy names 20k so the page stays self-consistent. +// gates at 20k (the shipped default since #858) and not at the account's own configured floor: the host +// hands this package only the plan settings (SetWaitPolicy), not the floor. An account that tuned its +// floor sees a replay of the default policy, and the panel's copy names 20k so the page stays self-consistent. // // THE COUNT (#858, #868). The live policy picks the count per wait (keepalive.Policy.WaitPings: a tool // wait up to 8, background K+3, a person K+1, no tool call yet 0). With withFacts, every span carries the @@ -1405,11 +1423,11 @@ func (d *DB) pingSpans(f Filter, minPrefix int64, withFacts bool) ([]pingSpan, e // windows run over the same filtered rows. func (d *DB) pingSpansWithFacts(cond string, args []any, minPrefix int64) ([]pingSpan, error) { rows, err := d.sql.QueryContext(d.readCtx(), `WITH `+factsCTE(cond)+`, s AS ( - SELECT session_id, thread_id, provider, cache_ttl, stop, wf, seen, prefix, + SELECT session_id, thread_id, provider, hour, stop, wf, seen, prefix, ROW_NUMBER() OVER (PARTITION BY tenant_id, session_id ORDER BY ts, id) - 1 AS turn, (LEAD(ts) OVER (PARTITION BY tenant_id, session_id ORDER BY ts, id) - ts) / 1000.0 AS gap_s FROM t) - SELECT session_id, gap_s, thread_id, provider, cache_ttl, stop, wf, seen FROM s + SELECT session_id, gap_s, thread_id, provider, hour, stop, wf, seen FROM s WHERE turn >= 1 AND prefix >= ? AND (gap_s IS NULL OR gap_s > 0)`, append(args, minPrefix)...) if err != nil { @@ -1418,14 +1436,14 @@ func (d *DB) pingSpansWithFacts(cond string, args []any, minPrefix int64) ([]pin defer rows.Close() var out []pingSpan for rows.Next() { - var sess, thread, prov, ttl, stop, wf string + var sess, thread, prov, stop, wf string var gap sql.NullFloat64 - var seen int64 - if err := rows.Scan(&sess, &gap, &thread, &prov, &ttl, &stop, &wf, &seen); err != nil { + var seen, hour int64 + if err := rows.Scan(&sess, &gap, &thread, &prov, &hour, &stop, &wf, &seen); err != nil { return nil, err } out = append(out, pingSpan{session: sess, gap: gap.Float64, open: !gap.Valid, - facts: scanFacts(thread, stop, wf, prov, ttl, seen)}) + facts: scanFacts(thread, stop, wf, prov, seen, hour)}) } return out, rows.Err() } @@ -1468,11 +1486,11 @@ func (d *DB) expiries(f Filter, withFacts bool) ([]expiry, error) { rows, err := d.sql.QueryContext(d.readCtx(), `WITH `+factsCTE(cond)+`, s AS ( SELECT session_id, ts, cost_usd, model, provider, cache_miss_reason, cache_write, LAG(ts) OVER w AS prev_ts, LAG(thread_id) OVER w AS p_thread, LAG(stop) OVER w AS p_stop, - LAG(wf) OVER w AS p_wf, LAG(provider) OVER w AS p_prov, LAG(cache_ttl) OVER w AS p_ttl, + LAG(wf) OVER w AS p_wf, LAG(provider) OVER w AS p_prov, LAG(hour) OVER w AS p_hour, LAG(seen) OVER w AS p_seen FROM t WINDOW w AS (PARTITION BY tenant_id, session_id ORDER BY ts, id)) SELECT session_id, (ts - prev_ts) / 1000.0, cost_usd, provider, model, - p_thread, p_stop, p_wf, p_prov, p_ttl, p_seen FROM s + p_thread, p_stop, p_wf, p_prov, p_hour, p_seen FROM s WHERE cache_miss_reason = 'ttl_expiry' AND cache_write > 0 AND prev_ts IS NOT NULL`, args...) if err != nil { return nil, err @@ -1480,18 +1498,29 @@ func (d *DB) expiries(f Filter, withFacts bool) ([]expiry, error) { defer rows.Close() for rows.Next() { var e expiry - var prov, mod, pThread, pStop, pWf, pProv, pTTL string - var pSeen int64 - if err := rows.Scan(&e.session, &e.gap, &e.usd, &prov, &mod, &pThread, &pStop, &pWf, &pProv, &pTTL, &pSeen); err != nil { + var prov, mod, pThread, pStop, pWf, pProv string + var pSeen, pHour int64 + if err := rows.Scan(&e.session, &e.gap, &e.usd, &prov, &mod, &pThread, &pStop, &pWf, &pProv, &pHour, &pSeen); err != nil { return nil, err } e.ttl = missLifetime(prov, mod) - e.facts = scanFacts(pThread, pStop, pWf, pProv, pTTL, pSeen) + e.facts = scanFacts(pThread, pStop, pWf, pProv, pSeen, pHour) out = append(out, e) } return out, rows.Err() } +// planReaches reports whether the plan's schedule for the wait before an expiry reaches it. A thread on +// the 1-hour tier keeps an entry that lives an hour, so its reach is judged against that lifetime. +func planReaches(pol keepalive.Policy, e expiry) bool { + n, idle := pol.SpanPings(e.facts) + ttl := e.ttl + if e.facts.OneHour { + ttl = time.Hour + } + return covers(idle.Seconds(), n, e.gap, ttl) +} + // covers reports whether n pings at idle X reach an expiry that came after `gap` seconds of idle, on a // cache entry of lifetime ttl. No ping, no reach. func covers(x float64, n int, gap float64, ttl time.Duration) bool { @@ -1550,6 +1579,8 @@ type KeepAliveCalc struct { // off (#858). With "wait_plan" the ladder carries a rung of the live plan, marked current, and the // fixed-K rungs are the comparison. Each rung also states its own model. Plan string `json:"plan"` + // Counts are the account's plan counts, present with "wait_plan", so the page can state them. + Counts *PlanCounts `json:"plan_counts,omitempty"` // PrefixSource says where the prefill came from, so nobody reads a number as measured when // it was typed: "session", "account_median" or "given". PrefixSource string `json:"prefix_source,omitempty"` @@ -1578,10 +1609,19 @@ func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model st if k <= 0 { k = recMaxPings } - wait.MaxPings = k + // The keeper applies per-wait counts only below 300 s of Idle: at 300 s or more it keeps the base. + if idleSeconds >= 300 { + wait.WaitPlan = false + } + wait.MaxPings, wait.Idle = k, time.Duration(idleSeconds*float64(time.Second)) out := &KeepAliveCalc{IdleSeconds: idleSeconds, Prefix: prefix, Model: model, Rows: []CalcRow{}, Plan: PlanFixed} if wait.WaitPlan { out.Plan = PlanWait + out.Counts = &PlanCounts{ + ToolWaitMax: orInt(wait.ToolWaitMaxPings, keepalive.DefaultToolWaitMaxPings), + BackgroundExtra: orInt(wait.BackgroundExtraPings, keepalive.DefaultBackgroundExtraPings), + HumanExtra: orInt(wait.HumanExtraPings, keepalive.DefaultHumanExtraPings), + MaxPings: keepalive.MaxWaitPings} } var p modelinfo.Price if price != nil && model != "" { @@ -1642,13 +1682,13 @@ func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model st // is its own count's (a tool wait up to 8, a thread with no tool call yet 0). row := CalcRow{MaxPings: k, Coverage: CoverageSeconds(idleSeconds, k), Current: true, Plan: PlanWait} for _, e := range exps { - if covers(idleSeconds, wait.SpanPings(e.facts), e.gap, e.ttl) { + if planReaches(wait, e) { row.Convertible++ row.ConvertibleUSD += e.usd } } for _, sp := range spans { - row.Pings += int64(sp.planPings(idleSeconds, wait)) + row.Pings += int64(sp.planPings(wait)) } finish(&row) out.Rows = append(out.Rows, row) @@ -1801,6 +1841,22 @@ type KeepAliveRecommendation struct { KeepAliveCoverage } +// PlanCounts are the counts of the per-wait plan in force: a tool wait up to ToolWaitMax pings, +// background work base + BackgroundExtra, a person base + HumanExtra, none above MaxPings (or the base). +type PlanCounts struct { + ToolWaitMax int `json:"tool_wait_max"` + BackgroundExtra int `json:"background_extra"` + HumanExtra int `json:"human_extra"` + MaxPings int `json:"max_pings"` +} + +func orInt(v, def int) int { + if v <= 0 { + return def + } + return v +} + // RecInterval is a 90% range in dollars. Never a point. type RecInterval struct { LoUSD float64 `json:"lo_usd"` @@ -1829,7 +1885,7 @@ const ( // recMaxPings) and the fixed-K interval comes beside it; with WaitPlan off the interval is the // fixed-K one, as before #858. func (d *DB) KeepAliveRecommend(f Filter, wait keepalive.Policy) (*KeepAliveRecommendation, error) { - wait.MaxPings = recMaxPings + wait.MaxPings, wait.Idle = recMaxPings, recIdleSeconds*time.Second out := &KeepAliveRecommendation{ServiceLoUSD: serviceLoUSD, ServiceHiUSD: serviceHiUSD, Plan: PlanFixed} if wait.WaitPlan { out.Plan = PlanWait @@ -1925,7 +1981,7 @@ func (d *DB) KeepAliveRecommend(f Filter, wait keepalive.Policy) (*KeepAliveReco var v float64 for _, m := range bySession[s] { if planned { - if covers(recIdleSeconds, pol.SpanPings(m.facts), m.gap, m.ttl) { + if planReaches(pol, m) { v += m.usd * recRecoverableShare } } else if m.gap > recIdleSeconds && m.gap <= CoverageSecondsFor(recIdleSeconds, k, m.ttl) { @@ -1934,7 +1990,7 @@ func (d *DB) KeepAliveRecommend(f Filter, wait keepalive.Policy) (*KeepAliveReco } for _, sp := range bySpan[s] { if planned { - v -= float64(sp.planPings(recIdleSeconds, pol)) * pingUSD + v -= float64(sp.planPings(pol)) * pingUSD } else { v -= float64(sp.pings(recIdleSeconds, k)) * pingUSD } diff --git a/dash/keepaliveplan_test.go b/dash/keepaliveplan_test.go index 37466676a..8a2641ed3 100644 --- a/dash/keepaliveplan_test.go +++ b/dash/keepaliveplan_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "net/http" + "strings" "testing" "time" @@ -37,16 +38,25 @@ const planBody = `{"model":"aws/claude-sonnet-5","max_tokens":32000,"stream":tru type turn struct { names []string stop string + // status is the HTTP status (0: 200). hour asks the 1-hour tier on every breakpoint; billed1h says the + // response billed its write at 1 hour. read says the turn only reads the cache (writes nothing). + status int + hour, billed1h bool + read bool } +func (t turn) withHour(billed bool) turn { t.hour, t.billed1h = true, billed; return t } +func (t turn) readOnly() turn { t.read = true; return t } +func (t turn) failed() turn { t.status = 500; return t } + var ( - tCall = turn{[]string{"Bash"}, "tool_use"} - tEnd = turn{nil, "end_turn"} + tCall = turn{names: []string{"Bash"}, stop: "tool_use"} + tEnd = turn{stop: "end_turn"} ) // keeperPings is the ping count the KEEPER plans for the wait after the last of seq, on thread. // 0 when the keeper plans no ping (no tool yet, done). -func keeperPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) int { +func keeperPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) (int, time.Duration) { t.Helper() now := time.Date(2026, 8, 22, 12, 0, 0, 0, time.UTC) d := &nopDispatcher{} @@ -57,35 +67,65 @@ func keeperPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) MaxUSDPerPing: 0.25, MinPrefixTokens: 20000, WaitPlan: true} key := keepalive.Key{Tenant: "t1", Session: "s1", Thread: thread} for i, r := range seq { + if i > 0 { + k.Arrive(key) // the next request arrives: the earlier entry is retired, as live + } hdr := http.Header{} hdr.Set("Authorization", "Bearer sk-x") hdr.Set("Anthropic-Version", "2023-06-01") + body, status := planBody, r.status + if status == 0 { + status = 200 + } + if r.hour { + body = strings.ReplaceAll(body, `"cache_control":{"type":"ephemeral"}`, `"cache_control":{"type":"ephemeral","ttl":"1h"}`) + } + u := keepalive.Usage{CacheRead: 48576, CacheWrite: 1200, StopReason: r.stop} + if r.read { + u.CacheWrite = 0 + } + if r.billed1h { + u.CacheWrite1h = u.CacheWrite + } k.Record(keepalive.Record{Key: key, StartedAt: now.Add(-time.Duration(len(seq)-1-i) * time.Second), - Body: []byte(planBody), Provider: bschemas.Anthropic, Route: "/v1/messages", Status: 200, - Usage: keepalive.Usage{CacheRead: 48576, CacheWrite: 1200, StopReason: r.stop}, + Body: []byte(body), Provider: bschemas.Anthropic, Route: "/v1/messages", Status: status, + Usage: u, ToolUseNames: r.names, UsageOK: true, Policy: pol, Header: hdr, Hold: func() dispatch.Ref { d.next++; return dispatch.Ref(fmt.Sprint(d.next)) }}) } - return k.Plans("t1", false, "s1")[thread].PingsLeft + pl := k.Plans("t1", false, "s1")[thread] + return pl.PingsLeft, pl.Idle } // dashPings is the count the DASHBOARD replay gives the same thread: it stores the rows the proxy // would write (the wait cause comes from the same classifier), reads them back through pingSpans, // and asks the span for its count. The last span is open, so it gets its full count. -func dashPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) int { +func dashPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) (int, time.Duration) { t.Helper() db := openTestDB(t) var evs []*Event for i, r := range seq { e := kaAgent(1_000_000+int64(i)*1000, "s1", 0.01) e.CacheRead, e.CacheWrite, e.ThreadID, e.StopReason = 48576, 1200, thread, r.stop + if r.read { + e.CacheWrite = 0 + } + if r.hour { + e.CacheTTL = "ephemeral_1h" + } + if r.billed1h { + e.CacheWrite1h = e.CacheWrite + } + if r.status != 0 { + e.Status = r.status + } // What the proxy stores: the classifier's answer for this response. The last row's open-task // counts are the ones the keeper's OpenTasks hook reports. b, m := 0, 0 if i == len(seq)-1 { b, m = bg, tm } - e.WaitingFor, e.WaitTool = waitcause.ClassifyWaitingFor(r.names, r.stop, false, true, b, m) + e.WaitingFor, e.WaitTool = waitcause.ClassifyWaitingFor(r.names, r.stop, r.status != 0, true, b, m) evs = append(evs, e) } if err := db.insertBatch(evs); err != nil { @@ -95,21 +135,25 @@ func dashPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) in if err != nil { t.Fatal(err) } - pol := keepalive.Policy{WaitPlan: true, MaxPings: base} - n := 0 - for _, sp := range spans { - if sp.open { - n = sp.planPings(280, pol) + pol := keepalive.Policy{WaitPlan: true, MaxPings: base, Idle: 280 * time.Second} + var open *pingSpan + for i := range spans { + if spans[i].open { + open = &spans[i] } } - return n + if open == nil { + return 0, 0 + } + _, idle := pol.SpanPings(open.facts) + return open.planPings(pol), idle } // TestDashboardReplayMatchesTheKeeper pins the dashboard's plan to the keeper's: the same responses // give the same ping count through both paths. It fails when either side changes a rule alone. func TestDashboardReplayMatchesTheKeeper(t *testing.T) { sub := "a:abc" - ask := turn{[]string{"AskUserQuestion"}, "tool_use"} + ask := turn{names: []string{"AskUserQuestion"}, stop: "tool_use"} for _, base := range []int{2, 10} { for _, tc := range []struct { name string @@ -125,19 +169,29 @@ func TestDashboardReplayMatchesTheKeeper(t *testing.T) { {"main, end_turn, teammate", "", []turn{tCall, tEnd}, 0, 1}, {"subagent, end_turn", sub, []turn{tCall, tEnd}, 0, 0}, {"subagent, end_turn, background", sub, []turn{tCall, tEnd}, 1, 0}, - {"subagent, handback", sub, []turn{tCall, {[]string{"SubagentHandback"}, "tool_use"}}, 0, 0}, - {"done", "", []turn{tCall, {[]string{"StructuredOutput"}, "tool_use"}}, 0, 0}, + {"subagent, handback", sub, []turn{tCall, {names: []string{"SubagentHandback"}, stop: "tool_use"}}, 0, 0}, + {"done", "", []turn{tCall, {names: []string{"StructuredOutput"}, stop: "tool_use"}}, 0, 0}, {"no tool call yet", "", []turn{tEnd, tEnd}, 0, 0}, {"no tool call yet, background open", "", []turn{tEnd, tEnd}, 1, 0}, {"earlier tool call counts", "", []turn{tCall, tEnd, tEnd}, 0, 0}, - {"tool stop, no names, background open", "", []turn{tCall, {nil, "tool_use"}}, 1, 0}, + {"tool stop, no names, background open", "", []turn{tCall, {stop: "tool_use"}}, 1, 0}, + {"1-hour tier, own tool", "", []turn{tCall.withHour(true), tCall.withHour(true)}, 0, 0}, + {"1-hour tier, a person", "", []turn{tCall.withHour(true), tEnd.withHour(true)}, 0, 0}, + {"1-hour tier, no tool call yet", "", []turn{tEnd.withHour(true), tEnd.withHour(true)}, 0, 0}, + {"1-hour tier, read-only turn inherits", "", []turn{tCall.withHour(true), tEnd.withHour(true).readOnly()}, 0, 0}, + {"mixed: asked 1 hour, billed 5 minutes", "", []turn{tCall.withHour(false), tCall.withHour(false)}, 0, 0}, + {"mixed: 1 hour then billed 5 minutes", "", []turn{tCall.withHour(true), tCall.withHour(false)}, 0, 0}, + {"failed call after a tool call", "", []turn{tCall, tEnd.failed()}, 0, 0}, } { t.Run(fmt.Sprintf("%s/base%d", tc.name, base), func(t *testing.T) { - want := keeperPings(t, base, tc.thread, tc.bg, tc.tm, tc.seq) - got := dashPings(t, base, tc.thread, tc.bg, tc.tm, tc.seq) + want, wantIdle := keeperPings(t, base, tc.thread, tc.bg, tc.tm, tc.seq) + got, gotIdle := dashPings(t, base, tc.thread, tc.bg, tc.tm, tc.seq) if got != want { t.Errorf("dashboard replay = %d pings, keeper = %d", got, want) } + if want > 0 && gotIdle != wantIdle { + t.Errorf("dashboard replay spaces pings %v apart, keeper %v", gotIdle, wantIdle) + } }) } } @@ -146,13 +200,16 @@ func TestDashboardReplayMatchesTheKeeper(t *testing.T) { // One known gap, stated and pinned (not hidden): a tool stop with NO tool name and nothing open is stored // as a wait for a person, as an ask-a-person tool is. The row cannot tell the two apart. The keeper reads // the first as a tool wait (8 pings), the replay as a person (base + 1). The case needs a spliced -// sniff window, which is rare. +// sniff window, which is rare. The same window gives the second known gap: the keeper counts a response +// with an incomplete name list as a tool call (calledATool), and the stored row has no such fact, so the +// replay's "tool seen" misses it (a thread whose only tool calls were spliced reads "no tool yet" until a +// tool stop is stored). func TestDashboardReplayKnownGapSplicedToolStop(t *testing.T) { - seq := []turn{tCall, {nil, "tool_use"}} - if got := keeperPings(t, 2, "", 0, 0, seq); got != 8 { + seq := []turn{tCall, {stop: "tool_use"}} + if got, _ := keeperPings(t, 2, "", 0, 0, seq); got != 8 { t.Fatalf("keeper = %d, want 8", got) } - if got := dashPings(t, 2, "", 0, 0, seq); got != 3 { + if got, _ := dashPings(t, 2, "", 0, 0, seq); got != 3 { t.Errorf("dashboard replay = %d, want 3 (base + 1)", got) } } @@ -300,3 +357,28 @@ func TestPlanReplayStaysNearTheFixedKCost(t *testing.T) { t.Errorf("plan reads %v against fixed-K reads %v", plan, fixed) } } + +// At an idle of 300 s or more the keeper keeps the base count, so the plan rung models fixed K. +func TestCalculatorAtThreeHundredSecondsModelsFixedK(t *testing.T) { + fx := planFixture(t) + c, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 300, 100_000, "aws/claude-sonnet-5", planPrice, 2, keepalive.Policy{WaitPlan: true}) + if err != nil { + t.Fatal(err) + } + if c.Plan != PlanFixed || len(c.Rows) != 4 { + t.Errorf("plan = %q, rows = %d, want fixed_k and 4", c.Plan, len(c.Rows)) + } +} + +// The plan's counts reach the response, so the page does not hard-code them. +func TestCalculatorStatesThePlanCounts(t *testing.T) { + fx := planFixture(t) + c, err := fx.db.KeepAliveCalc(Filter{TenantAll: true}, 280, 100_000, "aws/claude-sonnet-5", planPrice, 2, + keepalive.Policy{WaitPlan: true, ToolWaitMaxPings: 6}) + if err != nil { + t.Fatal(err) + } + if c.Counts == nil || c.Counts.ToolWaitMax != 6 || c.Counts.HumanExtra != keepalive.DefaultHumanExtraPings { + t.Errorf("plan_counts = %+v", c.Counts) + } +} diff --git a/dash/testdata/golden/ka_calc.json b/dash/testdata/golden/ka_calc.json index af9e75303..969456547 100644 --- a/dash/testdata/golden/ka_calc.json +++ b/dash/testdata/golden/ka_calc.json @@ -6,6 +6,12 @@ "keepalive_recorded_rows": 585, "model": "m-unpriced", "plan": "wait_plan", + "plan_counts": { + "background_extra": 3, + "human_extra": 1, + "max_pings": 11, + "tool_wait_max": 8 + }, "prefix_source": "account_median", "prefix_tokens": 30998, "priced": false, diff --git a/dash/ui/app.js b/dash/ui/app.js index a9f73d150..6b96fa546 100644 --- a/dash/ui/app.js +++ b/dash/ui/app.js @@ -9524,8 +9524,9 @@ function kaRungName(r) { /** kaRungCoverage says how far a rung reaches. The plan's reach depends on the wait, so it has no single figure. */ function kaRungCoverage(r) { + const pc = kaState.calc && kaState.calc.plan_counts; return r.plan === 'wait_plan' - ? 'by wait: a tool up to 8, a person base+1' + ? (pc ? `by wait: a tool up to ${pc.tool_wait_max}, a person base+${pc.human_extra}` : 'by wait') : r.coverage_seconds.toFixed(0) + 's = ' + r.max_pings + '×' + kaState.x + ' + 300'; } @@ -9577,7 +9578,7 @@ async function loadKACalc() { for (const r of c.rows) { // Emphasis by weight, not by a new hue: the current row is the accent, the rest recede. tb.appendChild(el('tr', { class: r.current ? 'is-current' : 'muted-row', - 'data-testid': 'ka-ladder-k' + r.max_pings }, + 'data-testid': r.plan === 'wait_plan' ? 'ka-ladder-plan' : 'ka-ladder-k' + r.max_pings }, el('td', {}, kaRungName(r)), el('td', {}, kaRungCoverage(r)), el('td', { class: 'num' }, num(r.convertible_misses)), @@ -9612,9 +9613,12 @@ async function loadKACalc() { 'what they cost, and what the pings cost. A replay of your past gaps, not a forecast. ' + (c.plan === 'wait_plan' ? 'The first row is the live plan: it picks the count from what each thread waits for ' + - '(up to 8 for a tool wait, the base plus 3 for background work, the base plus 1 for a person, ' + + (c.plan_counts + ? `(up to ${c.plan_counts.tool_wait_max} for a tool wait, the base plus ${c.plan_counts.background_extra} for background work, ` + + `the base plus ${c.plan_counts.human_extra} for a person, ` + : '(') + 'none before the thread has called a tool). The fixed rows send the same count for every wait, ' + - 'for comparison. ' + 'for comparison. It shows your account setting after campaigns and manager strategies, not a per-session override. ' : 'Your account has the per-wait plan switched off, so every row is a fixed count and the row ' + 'with your count is the current one. ') + 'Catching grows sharply from one ping to two, then flattens while ping cost keeps climbing.')); diff --git a/keepalive/waitplan.go b/keepalive/waitplan.go index 96051cc5e..eecdac6a1 100644 --- a/keepalive/waitplan.go +++ b/keepalive/waitplan.go @@ -2,6 +2,7 @@ package keepalive import ( "strings" + "time" "github.com/rossoctl/context-guru/internal/waitcause" ) @@ -133,8 +134,9 @@ type SpanFacts struct { ToolSeen bool // OpenAI is true for an OpenAI request. The plan does not cover OpenAI: it keeps the base count. OpenAI bool - // OneHour is true when the request asked for the 1-hour cache tier. That tier keeps its own - // schedule: the base count. + // OneHour is true when the thread's BILLED tier is the 1-hour one, as ResolveTier decides it + // (the request asked 1 hour on every breakpoint, and the last write was billed at 1 hour). + // The thread then keeps the schedule of Policy.ForOneHour. OneHour bool } @@ -147,28 +149,37 @@ func storedWait(cause, stopReason string) string { return toolStopWait(cause, cause != waitcause.WaitingHuman && cause != waitcause.WaitingOwnTool, stopReason) } -// SpanPings is the ping count the plan gives the idle span after a response, from the facts a stored -// row holds. It is the keeper's decision in one place, for a replay: +// SpanPings is the schedule the plan gives the idle span after a response, from the facts a stored +// row holds: the ping count and the spacing between pings. It is the keeper's decision in one place, +// for a replay. p.Idle is the 5-minute spacing; p.KeepAlive must be on. // -// - the count WaitPings gives for the wait cause; -// - 0 for a thread with no tool call yet (the no-tool rule); -// - 0 for a finished thread (waiting_for done) and for a failed call (retry): the keeper holds nothing for them; -// - MaxPings with WaitPlan off, for OpenAI, for the 1-hour tier, and for a row that stored neither a -// wait cause nor a stop reason (an old row: nothing to replay from). +// - 5-minute tier: the count WaitPings gives for the wait cause, at p.Idle. Only when p.Idle is under +// 300 s, as in the keeper; at 300 s or more the base count stays. +// - 1-hour tier: the schedule of ForOneHour (MaxPings1h pings, Idle1h apart; 1 ping at 59 minutes by default). +// - 0 pings for a thread with no tool call yet (the no-tool rule) and for a finished thread (done). +// The no-tool rule holds on both tiers. +// - A stored "retry" wait (a non-2xx response) gets 0 pings: the keeper does not track a response +// whose status is not 2xx (Record returns first), so it holds nothing for the span. +// - MaxPings with WaitPlan off, for OpenAI, and for a row that stored neither a wait cause nor a stop +// reason (an old row: nothing to replay from). // // TestDashboardReplayMatchesTheKeeper (package dash) runs the keeper and this function on the same cases. -func (p Policy) SpanPings(s SpanFacts) int { +func (p Policy) SpanPings(s SpanFacts) (pings int, idle time.Duration) { if !p.WaitPlan || s.OpenAI || (s.Cause == "" && s.StopReason == "") { - return p.MaxPings + return p.MaxPings, p.Idle } w := storedWait(s.Cause, s.StopReason) if w == waitcause.WaitingDone || w == waitcause.WaitingRetry || !s.ToolSeen { - return 0 + return 0, p.Idle } if s.OneHour { - return p.MaxPings + q := p.ForOneHour() + return q.MaxPings, q.Idle } - return p.WaitPings(w, s.Subagent) + if p.Idle >= 300*time.Second { + return p.MaxPings, p.Idle + } + return p.WaitPings(w, s.Subagent), p.Idle } // calledATool reports whether a response called a tool: a tool_use name or a tool stop reason. A diff --git a/proxy/dashwaitpolicy_test.go b/proxy/dashwaitpolicy_test.go new file mode 100644 index 000000000..6a2181ca7 --- /dev/null +++ b/proxy/dashwaitpolicy_test.go @@ -0,0 +1,59 @@ +package proxy + +import ( + "testing" + + "github.com/rossoctl/context-guru/tenant" +) + +// The dashboard replays the policy that RUNS (#868). DashWaitPolicy says the plan is on only when the +// request path would apply it: an active campaign and a strategy in the 1-hour band turn it off. +func TestDashWaitPolicyReportsTheEffectivePlan(t *testing.T) { + cache := CachePolicy{KeepAlive: true, WaitPlan: true, ToolWaitMaxPings: 6} + setup := func(t *testing.T) (*keeper, func() CachePolicy) { + k, _, _ := testKeeper(t, Limits{}) + k.h.keeper = k + k.h.static.Store(&Tenancy{ID: "t1", Cache: cache}) + return k, func() CachePolicy { return k.h.DashWaitPolicy()("t1") } + } + + k, get := setup(t) + if p := get(); !p.WaitPlan || p.ToolWaitMaxPings != 6 { + t.Errorf("plain account: %+v, want the plan on with its own counts", p) + } + + // An active campaign owns the count: plan off. + k, get = setup(t) + rt, _ := campaignRuntime(t, planSpec, nil, campaignModeActive) + k.pred = rt + if p := get(); p.WaitPlan { + t.Errorf("active campaign: plan still on") + } + // A shadow campaign decides nothing: plan on. + k, get = setup(t) + rt, _ = campaignRuntime(t, planSpec, nil, campaignModeShadow) + k.pred = rt + if p := get(); !p.WaitPlan { + t.Errorf("shadow campaign: plan off") + } + + // A strategy in the 1-hour band (Idle 300 s or more): plan off. A 5-minute strategy: plan on. + k, get = setup(t) + k.setStrategies([]tenant.Strategy{{ID: "s1", Active: true, IdleSeconds: 3540, MaxPings: 1, Target: tenant.Target{Mode: tenant.TargetAll}}}) + if p := get(); p.WaitPlan { + t.Errorf("1-hour strategy: plan still on") + } + k, get = setup(t) + k.setStrategies([]tenant.Strategy{{ID: "s2", Active: true, IdleSeconds: 240, MaxPings: 2, Target: tenant.Target{Mode: tenant.TargetAll}}}) + if p := get(); !p.WaitPlan { + t.Errorf("5-minute strategy: plan off") + } + + // The account switch. + k, get = setup(t) + cache.WaitPlan = false + k.h.static.Store(&Tenancy{ID: "t1", Cache: cache}) + if p := get(); p.WaitPlan { + t.Errorf("keepalive_wait_plan false: plan on") + } +} diff --git a/proxy/keepalive.go b/proxy/keepalive.go index e37160968..63463d574 100644 --- a/proxy/keepalive.go +++ b/proxy/keepalive.go @@ -547,26 +547,51 @@ func (h *Handler) KeepAlivePlans(tenantID string, allTenants bool, session strin return out } -// DashWaitPolicy is the dashboard's READ hook for an account's per-wait ping plan settings (#868): -// keepalive_wait_plan and the three counts. The calculator replays the plan with them. Wired in -// cmd/context-guru-proxy beside the keep-alive plans. +// DashWaitPolicy is the dashboard's READ hook for an account's EFFECTIVE per-wait ping plan (#868). The +// calculator replays the plan with it. It reports WaitPlan false, the fixed-K model, whenever the +// request path would turn the plan off for the account: +// +// - keepalive_wait_plan is false; +// - an active campaign owns the count (resolvePolicyWindows), where the tenant's keep-alive is on; +// - an active manager strategy sits in the 1-hour band (Idle of 300 s or more), where the keeper +// applies no per-wait counts; +// - the account cannot be read (fail to the old model, not to the new one). +// +// A per-session override also turns the plan off, but it is per session: no account-wide figure can +// carry it. Wired in cmd/context-guru-proxy beside the keep-alive plans. func (h *Handler) DashWaitPolicy() func(tenantID string) keepalive.Policy { return func(id string) keepalive.Policy { + off := keepalive.Policy{} tn := h.static.Load() if reg := h.registry(); reg != nil && h.opts.Tenants != nil { t, err := reg.Get(id) if err != nil { - return dash.DefaultWaitPolicy(0) + return off } if tn, err = h.opts.Tenants.ForTenant(t); err != nil { - return dash.DefaultWaitPolicy(0) + return off } } if tn == nil { - return dash.DefaultWaitPolicy(0) + return off } c := tn.Cache - return keepalive.Policy{WaitPlan: c.WaitPlan, ToolWaitMaxPings: c.ToolWaitMaxPings, - BackgroundExtraPings: c.BackgroundExtraPings, HumanExtraPings: c.HumanExtraPings} + pol := keepalive.Policy{WaitPlan: c.WaitPlan, ToolWaitMaxPings: c.ToolWaitMaxPings, + BackgroundExtraPings: c.BackgroundExtraPings, HumanExtraPings: c.HumanExtraPings, KeepAlive: c.KeepAlive} + if pc := h.predRT().forTenant(tn.ID); pc != nil && c.KeepAlive { + pc.mu.Lock() + active := pc.meta.Mode == campaignModeActive + pc.mu.Unlock() + if active { + return off + } + } + if h.keeper != nil { + if best := h.keeper.bestStrategy(tn.ID, h.keeper.now(), true); best != nil && + time.Duration(best.IdleSeconds)*time.Second >= 300*time.Second { + return off + } + } + return pol } } From ab1c0abc5ab6207bd8dbfbfcc7e73a202d0ffaa3 Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Tue, 3 Nov 2026 10:31:56 +0200 Subject: [PATCH 3/4] fix(proxy): a covering strategy decides the dashboard plan before a campaign (#868) DashWaitPolicy checked the campaign first. The request path (resolvePolicyWindows) lets a matching manager strategy win, and a campaign applies only when no strategy did. - proxy: find the covering strategy first. Idle of 300 s or more turns the plan off. Less keeps the account setting. Only with no strategy does an active campaign turn the plan off. - proxy: the doc comment states three limits (campaign price gate, the per-session override, a closed 1-hour window). - keepalive, dash: the comments say that the stored cache_ttl means any breakpoint asked 1 hour, where ResolveTier needs every one. - dash: plan_counts.max_pings is max(base, MaxWaitPings). - Tests: campaign plus 5-minute strategy gives plan on. A third known gap (mixed breakpoint tier) is pinned. Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- dash/keepalive.go | 9 +++++---- dash/keepaliveplan_test.go | 25 +++++++++++++++++++++-- keepalive/waitplan.go | 8 +++++--- proxy/dashwaitpolicy_test.go | 16 +++++++++++++++ proxy/keepalive.go | 39 ++++++++++++++++++++++-------------- 5 files changed, 73 insertions(+), 24 deletions(-) diff --git a/dash/keepalive.go b/dash/keepalive.go index f5204c3fe..cb60d0b65 100644 --- a/dash/keepalive.go +++ b/dash/keepalive.go @@ -1313,8 +1313,8 @@ func toolStopList() string { // // - seen is true when this request or an earlier one of the same THREAD called a tool: a tool stop // reason, or a stored wait on its own tool. -// - hour is the BILLED tier, as keepalive.ResolveTier decides it: the request asked 1 hour -// (cache_ttl) and the last write of the thread (this row's own, or the latest before it) was billed +// - hour is the BILLED tier, close to keepalive.ResolveTier: the request asked 1 hour (cache_ttl: +// ANY breakpoint asked, where ResolveTier needs EVERY breakpoint to ask, a known gap) and the last write of the thread (this row's own, or the latest before it) was billed // entirely at 1 hour (cache_write_1h >= cache_write). A row that wrote nothing inherits. // // Pings are excluded by the filter, as in every replay. Two extra windows, partitioned by thread. @@ -1621,7 +1621,7 @@ func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model st ToolWaitMax: orInt(wait.ToolWaitMaxPings, keepalive.DefaultToolWaitMaxPings), BackgroundExtra: orInt(wait.BackgroundExtraPings, keepalive.DefaultBackgroundExtraPings), HumanExtra: orInt(wait.HumanExtraPings, keepalive.DefaultHumanExtraPings), - MaxPings: keepalive.MaxWaitPings} + MaxPings: max(k, keepalive.MaxWaitPings)} } var p modelinfo.Price if price != nil && model != "" { @@ -1847,7 +1847,8 @@ type PlanCounts struct { ToolWaitMax int `json:"tool_wait_max"` BackgroundExtra int `json:"background_extra"` HumanExtra int `json:"human_extra"` - MaxPings int `json:"max_pings"` + // MaxPings is the cap on any plan count: max(base, MaxWaitPings). + MaxPings int `json:"max_pings"` } func orInt(v, def int) int { diff --git a/dash/keepaliveplan_test.go b/dash/keepaliveplan_test.go index 8a2641ed3..504da8792 100644 --- a/dash/keepaliveplan_test.go +++ b/dash/keepaliveplan_test.go @@ -43,6 +43,7 @@ type turn struct { status int hour, billed1h bool read bool + partial bool // only the first breakpoint asks 1 hour (stored cache_ttl still says 1 hour) } func (t turn) withHour(billed bool) turn { t.hour, t.billed1h = true, billed; return t } @@ -78,7 +79,11 @@ func keeperPings(t *testing.T, base int, thread string, bg, tm int, seq []turn) status = 200 } if r.hour { - body = strings.ReplaceAll(body, `"cache_control":{"type":"ephemeral"}`, `"cache_control":{"type":"ephemeral","ttl":"1h"}`) + n := -1 + if r.partial { + n = 1 + } + body = strings.Replace(body, `"cache_control":{"type":"ephemeral"}`, `"cache_control":{"type":"ephemeral","ttl":"1h"}`, n) } u := keepalive.Usage{CacheRead: 48576, CacheWrite: 1200, StopReason: r.stop} if r.read { @@ -203,7 +208,9 @@ func TestDashboardReplayMatchesTheKeeper(t *testing.T) { // sniff window, which is rare. The same window gives the second known gap: the keeper counts a response // with an incomplete name list as a tool call (calledATool), and the stored row has no such fact, so the // replay's "tool seen" misses it (a thread whose only tool calls were spliced reads "no tool yet" until a -// tool stop is stored). +// tool stop is stored). The third gap: the stored cache_ttl says ANY breakpoint asked 1 hour, where the +// keeper's ResolveTier needs EVERY breakpoint to ask. A mixed request with a billed 1-hour write reads +// as 1-hour in the replay (one ping at 59 minutes) and as 5-minute in the keeper. func TestDashboardReplayKnownGapSplicedToolStop(t *testing.T) { seq := []turn{tCall, {stop: "tool_use"}} if got, _ := keeperPings(t, 2, "", 0, 0, seq); got != 8 { @@ -382,3 +389,17 @@ func TestCalculatorStatesThePlanCounts(t *testing.T) { t.Errorf("plan_counts = %+v", c.Counts) } } + +// Third known gap: the stored cache_ttl means ANY breakpoint asked 1 hour; the keeper's ResolveTier needs +// EVERY breakpoint to ask. A mixed request billed at 1 hour is a 5-minute thread for the keeper (the plan +// counts, 5-minute spacing) and a 1-hour thread for the replay (one ping, 59 minutes apart). +func TestDashboardReplayKnownGapMixedBreakpointTier(t *testing.T) { + mixed := tCall.withHour(true) + mixed.partial = true + seq := []turn{tCall, mixed} + _, kIdle := keeperPings(t, 2, "", 0, 0, seq) + _, dIdle := dashPings(t, 2, "", 0, 0, seq) + if kIdle >= time.Hour || dIdle < 50*time.Minute { + t.Errorf("keeper spacing %v, replay spacing %v: want a 5-minute keeper and a 1-hour replay", kIdle, dIdle) + } +} diff --git a/keepalive/waitplan.go b/keepalive/waitplan.go index eecdac6a1..493668fbd 100644 --- a/keepalive/waitplan.go +++ b/keepalive/waitplan.go @@ -134,9 +134,11 @@ type SpanFacts struct { ToolSeen bool // OpenAI is true for an OpenAI request. The plan does not cover OpenAI: it keeps the base count. OpenAI bool - // OneHour is true when the thread's BILLED tier is the 1-hour one, as ResolveTier decides it - // (the request asked 1 hour on every breakpoint, and the last write was billed at 1 hour). - // The thread then keeps the schedule of Policy.ForOneHour. + // OneHour is true when the thread's BILLED tier is the 1-hour one. ResolveTier needs EVERY + // breakpoint to ask 1 hour; a stored row keeps only "ANY breakpoint asked 1 hour" (cache_ttl), so a + // replay can call a mixed request 1-hour when the keeper keeps it on 5 minutes. The billed write + // (cache_write_1h >= cache_write) is the other half of the rule. The thread then keeps the schedule + // of Policy.ForOneHour. OneHour bool } diff --git a/proxy/dashwaitpolicy_test.go b/proxy/dashwaitpolicy_test.go index 6a2181ca7..194d7f3bf 100644 --- a/proxy/dashwaitpolicy_test.go +++ b/proxy/dashwaitpolicy_test.go @@ -49,6 +49,22 @@ func TestDashWaitPolicyReportsTheEffectivePlan(t *testing.T) { t.Errorf("5-minute strategy: plan off") } + // Coexistence: a covering strategy wins over an active campaign, as in resolvePolicyWindows. + k, get = setup(t) + rt, _ = campaignRuntime(t, planSpec, nil, campaignModeActive) + k.pred = rt + k.setStrategies([]tenant.Strategy{{ID: "s3", Active: true, IdleSeconds: 240, MaxPings: 2, Target: tenant.Target{Mode: tenant.TargetAll}}}) + if p := get(); !p.WaitPlan { + t.Errorf("active campaign + 5-minute strategy: plan off, want on (the strategy wins)") + } + k, get = setup(t) + rt, _ = campaignRuntime(t, planSpec, nil, campaignModeActive) + k.pred = rt + k.setStrategies([]tenant.Strategy{{ID: "s4", Active: true, IdleSeconds: 3540, MaxPings: 1, Target: tenant.Target{Mode: tenant.TargetAll}}}) + if p := get(); p.WaitPlan { + t.Errorf("active campaign + 1-hour strategy: plan on, want off") + } + // The account switch. k, get = setup(t) cache.WaitPlan = false diff --git a/proxy/keepalive.go b/proxy/keepalive.go index 63463d574..a172b3b00 100644 --- a/proxy/keepalive.go +++ b/proxy/keepalive.go @@ -548,17 +548,24 @@ func (h *Handler) KeepAlivePlans(tenantID string, allTenants bool, session strin } // DashWaitPolicy is the dashboard's READ hook for an account's EFFECTIVE per-wait ping plan (#868). The -// calculator replays the plan with it. It reports WaitPlan false, the fixed-K model, whenever the -// request path would turn the plan off for the account: +// calculator replays the plan with it. It follows the request path (resolvePolicyWindows) in the same +// order, and reports WaitPlan false, the fixed-K model, when that path would turn the plan off: // -// - keepalive_wait_plan is false; -// - an active campaign owns the count (resolvePolicyWindows), where the tenant's keep-alive is on; -// - an active manager strategy sits in the 1-hour band (Idle of 300 s or more), where the keeper -// applies no per-wait counts; -// - the account cannot be read (fail to the old model, not to the new one). +// 1. keepalive_wait_plan is false, or the account cannot be read (fail to the old model); +// 2. a covering manager strategy wins first, found as the request path finds it for a wait that is +// not for a person (bestStrategy, windows ignored). Its Idle decides: 300 s or more (the 1-hour +// band) turns the plan off, less keeps the account's own setting. The campaign is then skipped; +// 3. only when no strategy covers the account, an active campaign (keep-alive on) owns the count and +// turns the plan off. // -// A per-session override also turns the plan off, but it is per session: no account-wide figure can -// carry it. Wired in cmd/context-guru-proxy beside the keep-alive plans. +// Limits, stated because one read hook cannot see a single request: +// - A campaign owns the count only on a request that passes its price gate (pdecOK in +// predictive.go). An account with no prices never has its count owned, but the hook reports plan off. +// - A per-session override turns the plan off for that session only. No account-wide figure shows it. +// - The plan flag is one value for the account. A 1-hour strategy whose window is closed leaves a +// person's wait on the account plan (the windows apply to that wait), but the hook reports plan off. +// +// Wired in cmd/context-guru-proxy beside the keep-alive plans. func (h *Handler) DashWaitPolicy() func(tenantID string) keepalive.Policy { return func(id string) keepalive.Policy { off := keepalive.Policy{} @@ -578,6 +585,14 @@ func (h *Handler) DashWaitPolicy() func(tenantID string) keepalive.Policy { c := tn.Cache pol := keepalive.Policy{WaitPlan: c.WaitPlan, ToolWaitMaxPings: c.ToolWaitMaxPings, BackgroundExtraPings: c.BackgroundExtraPings, HumanExtraPings: c.HumanExtraPings, KeepAlive: c.KeepAlive} + if h.keeper != nil { + if best := h.keeper.bestStrategy(tn.ID, h.keeper.now(), true); best != nil { + if time.Duration(best.IdleSeconds)*time.Second >= 300*time.Second { + return off + } + return pol // a covering strategy decides; the campaign is skipped + } + } if pc := h.predRT().forTenant(tn.ID); pc != nil && c.KeepAlive { pc.mu.Lock() active := pc.meta.Mode == campaignModeActive @@ -586,12 +601,6 @@ func (h *Handler) DashWaitPolicy() func(tenantID string) keepalive.Policy { return off } } - if h.keeper != nil { - if best := h.keeper.bestStrategy(tn.ID, h.keeper.now(), true); best != nil && - time.Duration(best.IdleSeconds)*time.Second >= 300*time.Second { - return off - } - } return pol } } From 5e0cc32face1fe42dfafe8f6a049ed486a67a4ed Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Tue, 3 Nov 2026 12:57:00 +0200 Subject: [PATCH 4/4] fix(proxy): respect strategy windows in the dashboard plan while a campaign is active (#868) With an active campaign the request path does not retry a closed window, so a strategy whose window is closed falls through to the campaign. DashWaitPolicy ignored windows and reported plan on. - proxy: while a campaign is active, look up the strategy with windows respected. A closed window gives plan off. An open window keeps the rule (300 s or more gives off, less gives the account setting). Without a campaign the window-ignoring lookup stays. - proxy: the doc comment says the flag then depends on the hour. - Tests: campaign plus a 5-minute strategy, window closed (off) and open (on), and no campaign with a closed window (on). Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- proxy/dashwaitpolicy_test.go | 30 ++++++++++++++++++++++++++++-- proxy/keepalive.go | 25 ++++++++++++++++--------- 2 files changed, 44 insertions(+), 11 deletions(-) diff --git a/proxy/dashwaitpolicy_test.go b/proxy/dashwaitpolicy_test.go index 194d7f3bf..0f1ec3e8a 100644 --- a/proxy/dashwaitpolicy_test.go +++ b/proxy/dashwaitpolicy_test.go @@ -53,18 +53,44 @@ func TestDashWaitPolicyReportsTheEffectivePlan(t *testing.T) { k, get = setup(t) rt, _ = campaignRuntime(t, planSpec, nil, campaignModeActive) k.pred = rt - k.setStrategies([]tenant.Strategy{{ID: "s3", Active: true, IdleSeconds: 240, MaxPings: 2, Target: tenant.Target{Mode: tenant.TargetAll}}}) + k.setStrategies([]tenant.Strategy{{ID: "s3", Active: true, IdleSeconds: 240, MaxPings: 2, Windows: []tenant.Window{{Start: "11:00", End: "13:00", TZ: "UTC"}}, Target: tenant.Target{Mode: tenant.TargetAll}}}) if p := get(); !p.WaitPlan { t.Errorf("active campaign + 5-minute strategy: plan off, want on (the strategy wins)") } k, get = setup(t) rt, _ = campaignRuntime(t, planSpec, nil, campaignModeActive) k.pred = rt - k.setStrategies([]tenant.Strategy{{ID: "s4", Active: true, IdleSeconds: 3540, MaxPings: 1, Target: tenant.Target{Mode: tenant.TargetAll}}}) + k.setStrategies([]tenant.Strategy{{ID: "s4", Active: true, IdleSeconds: 3540, MaxPings: 1, Windows: []tenant.Window{{Start: "11:00", End: "13:00", TZ: "UTC"}}, Target: tenant.Target{Mode: tenant.TargetAll}}}) if p := get(); p.WaitPlan { t.Errorf("active campaign + 1-hour strategy: plan on, want off") } + // Windows with a campaign (the test clock is 12:00 UTC): a closed window falls through to the + // campaign, an open one keeps the strategy. With no campaign a closed window is still ignored. + win := func(start, end string) []tenant.Window { return []tenant.Window{{Start: start, End: end, TZ: "UTC"}} } + five := func(w []tenant.Window) []tenant.Strategy { + return []tenant.Strategy{{ID: "s5", Active: true, IdleSeconds: 240, MaxPings: 2, Windows: w, Target: tenant.Target{Mode: tenant.TargetAll}}} + } + k, get = setup(t) + rt, _ = campaignRuntime(t, planSpec, nil, campaignModeActive) + k.pred = rt + k.setStrategies(five(win("02:00", "03:00"))) + if p := get(); p.WaitPlan { + t.Errorf("active campaign + 5-minute strategy, window closed: plan on, want off (the campaign owns it)") + } + k, get = setup(t) + rt, _ = campaignRuntime(t, planSpec, nil, campaignModeActive) + k.pred = rt + k.setStrategies(five(win("11:00", "13:00"))) + if p := get(); !p.WaitPlan { + t.Errorf("active campaign + 5-minute strategy, window open: plan off, want on") + } + k, get = setup(t) + k.setStrategies(five(win("02:00", "03:00"))) + if p := get(); !p.WaitPlan { + t.Errorf("no campaign + 5-minute strategy, window closed: plan off, want on") + } + // The account switch. k, get = setup(t) cache.WaitPlan = false diff --git a/proxy/keepalive.go b/proxy/keepalive.go index a172b3b00..b38543d60 100644 --- a/proxy/keepalive.go +++ b/proxy/keepalive.go @@ -553,11 +553,14 @@ func (h *Handler) KeepAlivePlans(tenantID string, allTenants bool, session strin // // 1. keepalive_wait_plan is false, or the account cannot be read (fail to the old model); // 2. a covering manager strategy wins first, found as the request path finds it for a wait that is -// not for a person (bestStrategy, windows ignored). Its Idle decides: 300 s or more (the 1-hour +// not for a person (bestStrategy, windows ignored; respected while a campaign is active). Its Idle decides: 300 s or more (the 1-hour // band) turns the plan off, less keeps the account's own setting. The campaign is then skipped; // 3. only when no strategy covers the account, an active campaign (keep-alive on) owns the count and // turns the plan off. // +// While a campaign is active the answer depends on the hour: a strategy whose window is closed falls +// through to the campaign, so the flag is right at the moment the page is read. +// // Limits, stated because one read hook cannot see a single request: // - A campaign owns the count only on a request that passes its price gate (pdecOK in // predictive.go). An account with no prices never has its count owned, but the hook reports plan off. @@ -585,21 +588,25 @@ func (h *Handler) DashWaitPolicy() func(tenantID string) keepalive.Policy { c := tn.Cache pol := keepalive.Policy{WaitPlan: c.WaitPlan, ToolWaitMaxPings: c.ToolWaitMaxPings, BackgroundExtraPings: c.BackgroundExtraPings, HumanExtraPings: c.HumanExtraPings, KeepAlive: c.KeepAlive} + // With an active campaign the request path does not retry a closed window (resolvePolicyWindows), + // so a strategy counts only while its window is open and a closed one falls through to the + // campaign. Without one, the window-ignoring lookup stands. + campaignOn := false + if pc := h.predRT().forTenant(tn.ID); pc != nil && c.KeepAlive { + pc.mu.Lock() + campaignOn = pc.meta.Mode == campaignModeActive + pc.mu.Unlock() + } if h.keeper != nil { - if best := h.keeper.bestStrategy(tn.ID, h.keeper.now(), true); best != nil { + if best := h.keeper.bestStrategy(tn.ID, h.keeper.now(), !campaignOn); best != nil { if time.Duration(best.IdleSeconds)*time.Second >= 300*time.Second { return off } return pol // a covering strategy decides; the campaign is skipped } } - if pc := h.predRT().forTenant(tn.ID); pc != nil && c.KeepAlive { - pc.mu.Lock() - active := pc.meta.Mode == campaignModeActive - pc.mu.Unlock() - if active { - return off - } + if campaignOn { + return off } return pol }