From 156b7203a1757d56e16e1b13a28e34f44884f26c Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Mon, 2 Nov 2026 22:01:23 +0200 Subject: [PATCH 1/5] feat(keepalive): choose the ping count from what the thread waits for Pick the ping count when a turn ends, from the wait cause (#858): a tool wait is pinged until the result comes (at most 8), background work gets K+3, a person K+1, a teammate or a main agent K, and a thread with no tool call yet gets 0. Manager-strategy time windows apply only to a wait for a person. The sweep question is not asked during a tool wait. The default keepalive_min_prefix_tokens drops from 80000 to 20000. One switch, keepalive_wait_plan: false, restores the fixed count, the windows for every wait, the sweep questions and the no-tool-call pings. Metadata only: stop reason, tool names, open-task counts, thread id shape, token counts and time of day. Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- cmd/context-guru-proxy/main.go | 38 +- cmd/context-guru-proxy/terminal_tools_test.go | 15 + config/config.go | 54 ++- config/terminal_tools_test.go | 42 +++ dash/keepalive.go | 4 +- dash/keepalivetab_test.go | 4 +- docs/data-dependence.md | 2 +- docs/how-to/cache-keepalive.md | 37 +- keepalive/keeper.go | 37 +- keepalive/manualhold.go | 2 +- keepalive/substitute.go | 7 + keepalive/types.go | 12 + keepalive/types_test.go | 12 + keepalive/waitplan.go | 119 +++++++ keepalive/waitplan_test.go | 327 ++++++++++++++++++ proxy/keepalive.go | 24 +- proxy/keepaliveoverride.go | 3 + proxy/keepalivestrategy.go | 24 +- proxy/keepalivewaitplan_test.go | 121 +++++++ 19 files changed, 854 insertions(+), 30 deletions(-) create mode 100644 keepalive/waitplan.go create mode 100644 keepalive/waitplan_test.go create mode 100644 proxy/keepalivewaitplan_test.go diff --git a/cmd/context-guru-proxy/main.go b/cmd/context-guru-proxy/main.go index 1f49be0b4..239ee86a6 100644 --- a/cmd/context-guru-proxy/main.go +++ b/cmd/context-guru-proxy/main.go @@ -1158,25 +1158,39 @@ func buildHostedTenantConfig(doc []byte, e components.Emitter) (proxy.BuiltConfi // The mapping lives here rather than in `proxy` for the reason ConfigBuilder exists: the // proxy package does not depend on the configuration loader. func cachePolicy(c config.CacheConfig) proxy.CachePolicy { + r := c.Resolved() + // The per-wait ping plan (#858) rides on EVERY policy, a parked one too, because a manager + // strategy can switch the keep-alive on for an account whose own block has it off, and the plan + // then still has to apply. These fields alone never turn the keep-alive on (Policy.On reads + // KeepAlive). + plan := proxy.CachePolicy{ + WaitPlan: c.WaitPlanOn(), + ToolWaitMaxPings: r.KeepAliveToolWaitMaxPings, + BackgroundExtraPings: r.KeepAliveBackgroundExtraPings, + HumanExtraPings: r.KeepAliveHumanExtraPings, + } // Resolve the defaults only when something is actually switched on. A parked block // (`keepalive: false` with a tuned interval) must not come out looking enabled, and a // zero HeadTTLMinBytes must stay zero so a host that never configured this asks for // nothing rather than asking on every request. if !c.KeepAlive && !c.HeadTTL1h { - return proxy.CachePolicy{} + return plan } - r := c.Resolved() return proxy.CachePolicy{ - KeepAlive: c.KeepAlive, - Idle: time.Duration(r.KeepAliveIdleSeconds) * time.Second, - OpenAIIdle: time.Duration(r.KeepAliveOpenAIIdleSeconds) * time.Second, - MaxPings: r.KeepAliveMaxPings, - Idle1h: time.Duration(r.KeepAliveIdleSeconds1h) * time.Second, - MaxPings1h: r.KeepAliveMaxPings1h, - MaxUSDPerPing: r.KeepAliveMaxUSDPerPing, - MinPrefixTokens: r.KeepAliveMinPrefixTokens, - HeadTTL1h: c.HeadTTL1h, - HeadTTLMinBytes: headTTLMin(c.HeadTTL1h, r.HeadTTLMinBytes), + WaitPlan: plan.WaitPlan, + ToolWaitMaxPings: plan.ToolWaitMaxPings, + BackgroundExtraPings: plan.BackgroundExtraPings, + HumanExtraPings: plan.HumanExtraPings, + KeepAlive: c.KeepAlive, + Idle: time.Duration(r.KeepAliveIdleSeconds) * time.Second, + OpenAIIdle: time.Duration(r.KeepAliveOpenAIIdleSeconds) * time.Second, + MaxPings: r.KeepAliveMaxPings, + Idle1h: time.Duration(r.KeepAliveIdleSeconds1h) * time.Second, + MaxPings1h: r.KeepAliveMaxPings1h, + MaxUSDPerPing: r.KeepAliveMaxUSDPerPing, + MinPrefixTokens: r.KeepAliveMinPrefixTokens, + HeadTTL1h: c.HeadTTL1h, + HeadTTLMinBytes: headTTLMin(c.HeadTTL1h, r.HeadTTLMinBytes), } } diff --git a/cmd/context-guru-proxy/terminal_tools_test.go b/cmd/context-guru-proxy/terminal_tools_test.go index 71f3784f7..672b85d5b 100644 --- a/cmd/context-guru-proxy/terminal_tools_test.go +++ b/cmd/context-guru-proxy/terminal_tools_test.go @@ -5,6 +5,7 @@ import ( "testing" "github.com/rossoctl/context-guru/components" + "github.com/rossoctl/context-guru/config" ) const docWithKey = "preset: off\ncache:\n keepalive_terminal_tools: []\n" @@ -34,3 +35,17 @@ func TestHostConfigWithTerminalToolsStillBuilds(t *testing.T) { t.Fatalf("local rebuild of a host config with the key: %v", err) } } + +// #858: the wait-plan fields reach the policy, also for a parked account block (a manager strategy can +// switch the keep-alive on for it), and the off switch reaches it. +func TestCachePolicyCarriesTheWaitPlan(t *testing.T) { + p := cachePolicy(config.CacheConfig{}) + if !p.WaitPlan || p.On() || p.ToolWaitMaxPings != 8 || p.BackgroundExtraPings != 3 || p.HumanExtraPings != 1 { + t.Errorf("parked block: %+v", p) + } + off := false + p = cachePolicy(config.CacheConfig{KeepAlive: true, KeepAliveWaitPlan: &off}) + if p.WaitPlan || !p.On() || p.MinPrefixTokens != 20000 { + t.Errorf("switch off: %+v", p) + } +} diff --git a/config/config.go b/config/config.go index c5b75b586..81a229e67 100644 --- a/config/config.go +++ b/config/config.go @@ -171,6 +171,26 @@ type CacheConfig struct { // (StructuredOutput). An empty list (`keepalive_terminal_tools: []`) turns the rule off, which // is the behavior before #857: such a thread gets its pings. KeepAliveTerminalTools []string `yaml:"keepalive_terminal_tools"` + // KeepAliveWaitPlan turns on the per-wait ping plan (#858). Absent: on. `false` is the one switch + // that brings back the behavior before #858: MaxPings for every wait, time windows for every wait, + // sweep questions during tool waits, and pings for a thread with no tool call yet. It does not + // bring back the 80k minimum prefix, which is only a default (keepalive_min_prefix_tokens). + // + // With the plan on, the ping count follows what the thread waits for. "Base" is + // keepalive_max_pings: + // - its own tool (main or subagent): keep pinging until the result comes, at most + // keepalive_tool_wait_max_pings (default 8); + // - main, end_turn, background work: base + keepalive_background_extra_pings (default 3); + // - main, end_turn, a person: base + keepalive_human_extra_pings (default 1); + // - main waiting for a teammate, or a subagent waiting for its main agent: base; + // - a finished thread, or one with no tool call yet: 0. + // Manager-strategy time windows then apply only to a wait for a person, and the sweep question is + // not asked while a thread waits for its own tool. It applies to the 5-minute tier. 0 on the three + // counts means the default. The three counts are at most 11, the highest strategy ping count. + KeepAliveWaitPlan *bool `yaml:"keepalive_wait_plan"` + KeepAliveToolWaitMaxPings int `yaml:"keepalive_tool_wait_max_pings"` + KeepAliveBackgroundExtraPings int `yaml:"keepalive_background_extra_pings"` + KeepAliveHumanExtraPings int `yaml:"keepalive_human_extra_pings"` // HeadTTL1h asks for the ONE-HOUR tier on the request's HEAD breakpoints (`tools` and // `system`) while leaving the trailing message breakpoint at 5 minutes — the provider's @@ -236,8 +256,17 @@ const ( DefaultKeepAliveIdle1h = int(modelinfo.AnthropicDefaultKeepAliveIdle1h / time.Second) DefaultKeepAliveMaxPings1h = modelinfo.AnthropicDefaultKeepAliveMaxPings1h DefaultKeepAliveMaxUSDPerPing = 0.50 - DefaultKeepAliveMinPrefix = 80000 - DefaultHeadTTLMinBytes = 200000 + // DefaultKeepAliveMinPrefix is 20000 since #858. No long subagent idle span had a prefix under + // 20k, so any floor at or under 20k gives the same result there. It was 80000 (#435). + DefaultKeepAliveMinPrefix = 20000 + // The per-wait ping plan defaults (#858). They mirror keepalive.DefaultToolWaitMaxPings, + // DefaultBackgroundExtraPings and DefaultHumanExtraPings, pinned equal by a test. + DefaultKeepAliveToolWaitMaxPings = 8 + DefaultKeepAliveBackgroundExtraPings = 3 + DefaultKeepAliveHumanExtraPings = 1 + // MaxKeepAliveWaitPings bounds each of the three counts: the highest strategy ping count. + MaxKeepAliveWaitPings = 11 + DefaultHeadTTLMinBytes = 200000 // DefaultHeadTTLMinTokens is DEPRECATED, kept for old callers: the old default of 50,000. DefaultHeadTTLMinTokens = DefaultHeadTTLMinBytes / HeadTTLBytesPerToken // HeadTTLBytesPerToken converts the deprecated head_ttl_min_tokens into bytes. It is the @@ -279,6 +308,15 @@ func (c CacheConfig) Resolved() CacheConfig { if c.KeepAliveMinPrefixTokens <= 0 { c.KeepAliveMinPrefixTokens = DefaultKeepAliveMinPrefix } + if c.KeepAliveToolWaitMaxPings <= 0 { + c.KeepAliveToolWaitMaxPings = DefaultKeepAliveToolWaitMaxPings + } + if c.KeepAliveBackgroundExtraPings <= 0 { + c.KeepAliveBackgroundExtraPings = DefaultKeepAliveBackgroundExtraPings + } + if c.KeepAliveHumanExtraPings <= 0 { + c.KeepAliveHumanExtraPings = DefaultKeepAliveHumanExtraPings + } if c.HeadTTLMinBytes <= 0 { if c.HeadTTLMinTokens > 0 { c.HeadTTLMinBytes = c.HeadTTLMinTokens * HeadTTLBytesPerToken @@ -298,9 +336,16 @@ func (c CacheConfig) validate() error { if c.KeepAliveIdleSeconds < 0 || c.KeepAliveOpenAIIdleSeconds < 0 || c.KeepAliveMaxPings < 0 || c.HeadTTLMinTokens < 0 || c.HeadTTLMinBytes < 0 || c.KeepAliveMaxUSDPerPing < 0 || c.KeepAliveMinPrefixTokens < 0 || - c.KeepAliveIdleSeconds1h < 0 || c.KeepAliveMaxPings1h < 0 { + c.KeepAliveIdleSeconds1h < 0 || c.KeepAliveMaxPings1h < 0 || + c.KeepAliveToolWaitMaxPings < 0 || c.KeepAliveBackgroundExtraPings < 0 || c.KeepAliveHumanExtraPings < 0 { return fmt.Errorf("config: cache: negative values are not meaningful") } + if c.KeepAliveToolWaitMaxPings > MaxKeepAliveWaitPings || c.KeepAliveBackgroundExtraPings > MaxKeepAliveWaitPings || + c.KeepAliveHumanExtraPings > MaxKeepAliveWaitPings { + return fmt.Errorf("config: cache: keepalive_tool_wait_max_pings, keepalive_background_extra_pings and "+ + "keepalive_human_extra_pings must be at most %d (the credential hold is (K+1) x the idle time)", + MaxKeepAliveWaitPings) + } if c.HeadTTLMinBytes > MaxHeadTTLMinBytes || c.HeadTTLMinTokens > MaxHeadTTLMinBytes/HeadTTLBytesPerToken { return fmt.Errorf("config: cache: head_ttl_min_bytes must be at most %d (head_ttl_min_tokens at most %d)", MaxHeadTTLMinBytes, MaxHeadTTLMinBytes/HeadTTLBytesPerToken) @@ -979,3 +1024,6 @@ func (c *Config) NewStore() store.Store { } return store.NewMemory(c.Store) } + +// WaitPlanOn reports whether the per-wait ping plan is on (#858). Absent is on. +func (c CacheConfig) WaitPlanOn() bool { return c.KeepAliveWaitPlan == nil || *c.KeepAliveWaitPlan } diff --git a/config/terminal_tools_test.go b/config/terminal_tools_test.go index f6f75f1ba..f04d2e1df 100644 --- a/config/terminal_tools_test.go +++ b/config/terminal_tools_test.go @@ -29,3 +29,45 @@ func TestKeepAliveTerminalToolsYAML(t *testing.T) { }) } } + +// #858: the wait-plan keys. Absent: the plan is on with the shipped counts. Present: kept. +func TestKeepAliveWaitPlanYAML(t *testing.T) { + cfg, err := LoadBytes([]byte("preset: off\ncache:\n keepalive: true\n")) + if err != nil { + t.Fatal(err) + } + if !cfg.Cache.WaitPlanOn() || cfg.Cache.KeepAliveWaitPlan != nil { + t.Errorf("absent: plan on = %v, raw %v; want on and nil", cfg.Cache.WaitPlanOn(), cfg.Cache.KeepAliveWaitPlan) + } + r := cfg.Cache.Resolved() + if r.KeepAliveToolWaitMaxPings != 8 || r.KeepAliveBackgroundExtraPings != 3 || r.KeepAliveHumanExtraPings != 1 { + t.Errorf("absent: resolved = %d/%d/%d, want 8/3/1", r.KeepAliveToolWaitMaxPings, + r.KeepAliveBackgroundExtraPings, r.KeepAliveHumanExtraPings) + } + cfg, err = LoadBytes([]byte("preset: off\ncache:\n keepalive: true\n keepalive_wait_plan: false\n" + + " keepalive_tool_wait_max_pings: 5\n keepalive_background_extra_pings: 4\n keepalive_human_extra_pings: 2\n")) + if err != nil { + t.Fatal(err) + } + r = cfg.Cache.Resolved() + if cfg.Cache.WaitPlanOn() || r.KeepAliveToolWaitMaxPings != 5 || r.KeepAliveBackgroundExtraPings != 4 || + r.KeepAliveHumanExtraPings != 2 { + t.Errorf("present: on=%v %d/%d/%d, want off and 5/4/2", cfg.Cache.WaitPlanOn(), + r.KeepAliveToolWaitMaxPings, r.KeepAliveBackgroundExtraPings, r.KeepAliveHumanExtraPings) + } + for _, bad := range []string{"keepalive_tool_wait_max_pings: 12", "keepalive_human_extra_pings: -1"} { + if _, err := LoadBytes([]byte("preset: off\ncache:\n " + bad + "\n")); err == nil { + t.Errorf("%q was accepted", bad) + } + } +} + +// #858: the default minimum prefix is 20000; an explicit value stays. +func TestKeepAliveMinPrefixDefaultIs20k(t *testing.T) { + if got := (CacheConfig{}).Resolved().KeepAliveMinPrefixTokens; got != 20000 { + t.Errorf("default = %d, want 20000", got) + } + if got := (CacheConfig{KeepAliveMinPrefixTokens: 80000}).Resolved().KeepAliveMinPrefixTokens; got != 80000 { + t.Errorf("explicit 80000 became %d", got) + } +} diff --git a/dash/keepalive.go b/dash/keepalive.go index da1d4e751..4e96ac4d4 100644 --- a/dash/keepalive.go +++ b/dash/keepalive.go @@ -1262,9 +1262,9 @@ func PingsPerSpan(gap, idleSeconds float64, maxPings int) int { } // kaGateMinPrefix is the billed-prefix floor a replay gates on: the shipped default, mirroring -// config.DefaultKeepAliveMinPrefix, which this package may not import. Pinned by +// config.DefaultKeepAliveMinPrefix (20000 since #858), which this package may not import. Pinned by // TestTheReplayGateMatchesTheShippedPolicy. -const kaGateMinPrefix = 80000 +const kaGateMinPrefix = 20000 // pingSpan is one idle span a live keep-alive would send pings in. type pingSpan struct { diff --git a/dash/keepalivetab_test.go b/dash/keepalivetab_test.go index 69389147d..fb3654baf 100644 --- a/dash/keepalivetab_test.go +++ b/dash/keepalivetab_test.go @@ -879,8 +879,8 @@ func smallPrefix(e *Event) *Event { // The replay gate is the request path's gate. A drift here is a calculator modelling a policy // nobody runs, which is the whole of F5. func TestTheReplayGateMatchesTheShippedPolicy(t *testing.T) { - if kaGateMinPrefix != 80000 { - t.Errorf("kaGateMinPrefix = %d; config.DefaultKeepAliveMinPrefix is 80000 and the replay "+ + if kaGateMinPrefix != 20000 { + t.Errorf("kaGateMinPrefix = %d; config.DefaultKeepAliveMinPrefix is 20000 and the replay "+ "has to gate on what the request path gates on", kaGateMinPrefix) } } diff --git a/docs/data-dependence.md b/docs/data-dependence.md index 5c123cbc6..a101c747f 100644 --- a/docs/data-dependence.md +++ b/docs/data-dependence.md @@ -224,7 +224,7 @@ One row for each registered component. The test `components/all/datadep_test.go` | Thread tracker | metadata | 🟡 Works, metadata | 🟡 Works, metadata | 🟡 Works, metadata | It tells the main agent from its subagents. It uses hashes to check if a new request continues an earlier one. | | Session tracker | metadata | 🟡 Works, metadata | 🟡 Works, metadata | 🟡 Works, metadata | It needs a stable label for each conversation. When the client sends none, it hashes the start of the conversation. | | Cache tracker | metadata | 🟡 Works, metadata | 🟡 Works, metadata | 🟡 Works, metadata | It must know how much of the conversation the provider already cached, so components never edit that part. After a restart it asks the dashboard (when on) for the time each thread's cached prefix was last refreshed — by its own last request or a keep-alive ping that read the cache, under either session id the conversation uses (the row stores the content-derived alias id, a hash like the session id itself); timestamps and ids, metadata only — and treats a thread refreshed inside the request's cache lifetime as cached, not as new, until that thread's first request. Without the dashboard — and for a tenant running observe mode, whose rows carry no session id — it uses the proxy's start time instead. | -| Wait-cause and lifecycle signals | metadata | ❌ Off (it needs the dashboard) | 🟡 Works, metadata | 🟡 Works, metadata | To tell why the agent paused, it reads the request once. It keeps only counts and ids. Since #857 it also reads the **tool name** of the last tool call, to tell that an agent has finished (`done`). The keep-alive then sends no pings for that thread. A tool name is metadata, and the rule works at every storage level. | +| Wait-cause and lifecycle signals | metadata | ❌ Off (it needs the dashboard) | 🟡 Works, metadata | 🟡 Works, metadata | To tell why the agent paused, it reads the request once. It keeps only counts and ids. Since #857 it also reads the **tool name** of the last tool call, to tell that an agent has finished (`done`). The keep-alive then sends no pings for that thread. A tool name is metadata, and the rule works at every storage level. Since #858 the keep-alive also uses the **wait cause** (from the stop reason, the tool names and the open-task counts), whether the thread id starts with `a:`, the token count of the prefix and the time of day, to pick the ping count, to ignore a strategy's time windows for a wait that is not for a person, and to skip the sweep question during a tool wait. It also keeps one flag per thread: "has a tool call been seen". All of it is metadata, and it works at every storage level. | | Costly settings check (`/context-guru:save-more settings`, `GET /api/adapt/settings`) | metadata | ❌ Off (the cost needs the dashboard). The plugin still finds the settings and says "cost not measured yet". | 🟡 Works, metadata | 🟡 Works, metadata | It counts and prices dashboard rows by request shape: model name, tool count, `max_tokens` and the stream flag. One check also uses the upstream status (400) of a Haiku row in the same session. It reads no request text and keeps nothing new. The plugin reads only the named keys of the user's own settings files (the betas setting, the recap setting, `ANTHROPIC_SMALL_FAST_MODEL` and `ANTHROPIC_DEFAULT_HAIKU_MODEL`) on the user's machine, and sends none of them to the proxy (#615). | | Beep (`beep.sh`, the `beep` plugin option) | none | 🟡 Works, no data | 🟡 Works, no data | 🟡 Works, no data | It plays a sound when Claude finishes or waits for you. It reads no stored data and never reads the hook payload. It sends nothing to the proxy or anywhere else. The level changes nothing. | | Metrics (`/stats`, `/metrics`) | metadata | 🟡 Works, metadata | 🟡 Works, metadata | 🟡 Works, metadata | Totals and rates need only numbers. | diff --git a/docs/how-to/cache-keepalive.md b/docs/how-to/cache-keepalive.md index 4d8861448..1a1914bd6 100644 --- a/docs/how-to/cache-keepalive.md +++ b/docs/how-to/cache-keepalive.md @@ -60,13 +60,48 @@ cache: keepalive_idle_seconds: 240 # X — must be under 300 keepalive_openai_idle_seconds: 1680 # GPT-5.6+ Responses; must be under 1730 keepalive_max_pings: 2 # K - keepalive_min_prefix_tokens: 80000 # the gate + keepalive_min_prefix_tokens: 20000 # the gate (80000 before #858) keepalive_max_usd_per_ping: 0.50 keepalive_idle_seconds_1h: 3540 # X for a 1-hour cache (59 min); must be under 3600 keepalive_max_pings_1h: 1 # K for a 1-hour cache keepalive_terminal_tools: [StructuredOutput] # default; [] sends pings after the final answer, as before #857 + keepalive_wait_plan: true # false: one fixed K for every wait, as before #858 + keepalive_tool_wait_max_pings: 8 # a tool wait: ping until the result comes, at most this many + keepalive_background_extra_pings: 3 # end_turn, waiting for background work: K + 3 + keepalive_human_extra_pings: 1 # end_turn, waiting for a person: K + 1 ``` +### The ping count follows what the thread waits for + +When a turn ends, the proxy picks the ping count from what the thread waits for (#858). "K" is +`keepalive_max_pings` (2 by default). It reads only metadata: the stop reason, the tool names, the +open-task counts, and whether the thread id starts with `a:` (a subagent). + +| The thread waits for | Pings | Time windows apply? | +|---|---|---| +| Its own tool (main or subagent) | until the result comes, at most 8 | No | +| Main, `end_turn`, background work | K + 3 | No | +| Main, `end_turn`, a teammate | K | No | +| Subagent, `end_turn` or `SubagentHandback` (its main agent) | K | No | +| Main, `end_turn`, a person | K + 1 | **Yes, only this row** | +| Finished (`done`, #857), or **no tool call yet on the thread** | 0 | - | + +- **Time windows** (manager strategies) describe when a person answers. They gate only a wait for a + person. For any other wait, the highest-priority active strategy that targets the tenant applies at any + hour. With no such strategy, the account config applies, as before. +- **The sweep question** is not asked while a thread waits for its own tool. That span stays warm until + the result comes back, so a cut can never follow. It is asked on a span that began with `end_turn`. + The rule "keep-alive never shortens the cache strategy" is unchanged. +- **The hold limit** is still `(K+1) x X` with K at most 11 and X under 290 s, so at most 58 minutes. A + tool wait of 8 pings at 240 to 280 s holds at most 42 minutes. The plan never goes above 11 pings, unless + K itself is higher, and it never lowers K. +- **Not covered:** a 1-hour thread keeps its own schedule. A session override or an active learned + campaign sets its own count, and the plan does not change it. +- **One switch:** `keepalive_wait_plan: false` brings back one fixed K for every wait, time windows for + every wait, sweep questions as before, and pings for a thread with no tool call yet. It does not bring + back the 80,000 minimum prefix. The default is now 20,000 (`keepalive_min_prefix_tokens`). A value set + by a user or a strategy is not changed. + ### No pings after the final answer A response whose **last** tool call is a terminal tool (default: `StructuredOutput`) ends the agent. diff --git a/keepalive/keeper.go b/keepalive/keeper.go index d45029193..371692ed1 100644 --- a/keepalive/keeper.go +++ b/keepalive/keeper.go @@ -134,6 +134,11 @@ type kaEntry struct { usageOK bool // terminal is Options.TerminalTools as of record time (#857). terminal []string + // wait is the wait cause this entry's pings were planned for (waitplan.go, #858). "" when the + // plan did not apply (plan off, a 1-hour thread, OpenAI). The sweep question reads it. + wait string + // noTool: the plan is on and the thread has had no tool_use response yet, so it is not pinged. + noTool bool // endsAtText is whether the recorded request's last non-system message is not a tool_result turn // (endsAtTextTurn, #776): roles and block types only. Never on OpenAI: its cache has no breakpoints // and no text-turn rule, so a substitute there renews the prefix like a ping. Then the sweep's question cannot renew the @@ -322,6 +327,10 @@ func (e *kaEntry) pingable() bool { if e.isDone() { return false } + // A thread with no tool call yet is not pinged (#858): it may be a side call. + if e.noTool { + return false + } if e.pol.PredictorID == "" { return true } @@ -456,6 +465,9 @@ type Keeper struct { // turn that only READS the cache bills no write, so it carries no tier of its own and // inherits this. Written beside turns, so it is bounded and reset exactly as turns is. tier1h map[string]bool + // toolSeen remembers, per THREAD (keyed like live), that a response of it called a tool (#858). + // Bounded and reset exactly as turns is. + toolSeen map[string]bool // lastVary is, per THREAD (keyed like live), the output-budget step of the last ping actually // handed to the Dispatcher. It outlives the entry, as turns does, so a new entry on the same // thread never repeats the step its predecessor last sent (#706, #712 review). Bounded and reset @@ -552,7 +564,7 @@ func New(o Options) *Keeper { o.Go = func(f func()) { go f() } } return &Keeper{o: o, stop: make(chan struct{}), done: make(chan struct{}), - live: map[string]*kaEntry{}, turns: map[string]int{}, tier1h: map[string]bool{}, lastVary: map[string]int{}, inflight: map[string]int{}, + live: map[string]*kaEntry{}, turns: map[string]int{}, tier1h: map[string]bool{}, toolSeen: map[string]bool{}, lastVary: map[string]int{}, inflight: map[string]int{}, routeFails: map[string]int{}, routeTripped: map[string]time.Time{}, credDead: map[string]time.Time{}} } @@ -629,6 +641,7 @@ func (k *Keeper) Shutdown() { k.bytes = 0 k.turns = map[string]int{} k.tier1h = map[string]bool{} + k.toolSeen = map[string]bool{} k.lastVary = map[string]int{} } @@ -881,6 +894,20 @@ func (k *Keeper) Record(r Record) { if ttl1h && pol.On() { pol = pol.ForOneHour() } + // The per-wait ping plan (#858): the count follows what the thread waits for. The wait cause is read + // for every thread the plan covers (the sweep gate and the dashboard use it); the COUNT changes only + // on the 5-minute tier. A 1-hour thread keeps its own bounded schedule, and OpenAI pings 28 minutes + // apart, so more pings would hold the credential for hours. A strategy that asks for the 1-hour head + // tier carries a 1-hour Idle even when the gateway granted 5 minutes; its count is the manager's. + guard := pol.WaitPlan && pol.On() && provider != bschemas.OpenAI + planned := guard && !ttl1h && pol.Idle < 300*time.Second + wait := "" + if guard { + wait = k.WaitingFor(r.Key, r.ToolUseNames, r.ToolUseIncomplete, u.StopReason, usageOK) + } + if planned { + pol.MaxPings = pol.WaitPings(wait, IsSubagentThread(threadID)) + } // A keeper-OWNED copy, not the slice serve just used. It costs one allocation per tracked // session — and tracked sessions are gated down to a handful — in exchange for being able to // overwrite the bytes on release without corrupting anything else that might still read @@ -905,6 +932,7 @@ func (k *Keeper) Record(r Record) { emitter: emitter, provider: provider, model: model, route: route, preset: r.Preset, agent: r.Agent, pol: pol, ttl1h: ttl1h, prefix: prefix, turnRead: u.CacheRead, stopReason: u.StopReason, + wait: wait, toolUseNames: r.ToolUseNames, usageOK: usageOK, terminal: terminalFor(k.o.TerminalTools, r.ToolUseIncomplete), endsAtText: provider != bschemas.OpenAI && endsAtTextTurn(body), pingUSD: k.projectedPingUSD(model, prefix, route, r.Endpoint), hdr: hdr, tag: r.Tag, @@ -917,6 +945,9 @@ func (k *Keeper) Record(r Record) { // what lets a session's second request be recognised as such. k.turns[key]++ e.turn = k.turns[key] - 1 + seen := k.toolSeen[key] || calledATool(r.ToolUseNames, r.ToolUseIncomplete, u.StopReason) + k.toolSeen[key] = seen + e.noTool = guard && !seen // A write bills a tier; a read-only turn bills none and leaves the remembered one alone. if usageOK && (u.CacheWrite > 0 || ttl1h) { k.tier1h[key] = ttl1h @@ -926,6 +957,7 @@ func (k *Keeper) Record(r Record) { // read-only turn back to the 5-minute schedule. k.turns = map[string]int{key: k.turns[key]} k.tier1h = map[string]bool{} + k.toolSeen = map[string]bool{key: seen} // Keep the step of every thread that still holds a live entry, and of this one: a live // thread that lost its step would send its next ping at step 0, the same bytes as a ping 1 // it may have just sent (#712 review). k.live is bounded, so this stays bounded too. @@ -1157,6 +1189,7 @@ func (k *Keeper) Forget(tenantID string) { delete(k.live, key) delete(k.turns, key) delete(k.tier1h, key) + delete(k.toolSeen, key) offload.ClearKeepAliveCandidate(thread.Key(e.session, e.thread)) } // Every step the tenant's threads left, including threads with no live entry: lastVary has no @@ -2079,7 +2112,7 @@ func (k *Keeper) extraReadyLocked(e *kaEntry, now time.Time) bool { } tk := thread.Key(e.session, e.thread) _, _, hasSummary := offload.KeepAliveCandidateState(tk) - return hasSummary || offload.HasSweepPingCandidate(tk) + return hasSummary || (!e.toolWait() && offload.HasSweepPingCandidate(tk)) } // fireExtra sends the one extra request (kaEntry.extraDue) under the same claim, ceilings and latches diff --git a/keepalive/manualhold.go b/keepalive/manualhold.go index dfa472d91..6e1ae6368 100644 --- a/keepalive/manualhold.go +++ b/keepalive/manualhold.go @@ -58,7 +58,7 @@ const ( // "not opted in": that stays for a request that carried no opt-in and met no automatic policy. ReasonPingsUsed = "pings used" // the planned pings ran out and the keeper let go of the thread ReasonStopped = "stopped" // the keeper stopped the thread after a miss, a refused ping or a failure - ReasonBelowLimits = "below limits" // the automatic gates refused it: first request, under the minimum prefix, or the predictor + ReasonBelowLimits = "below limits" // the automatic gates refused it: first request, under the minimum prefix, no tool call yet on the thread (#858), or the predictor ReasonDone = "done" // the thread's last response ended with a terminal tool (#857): the agent is finished, so nothing is pinged ReasonNoEntry = "no entry" // nothing is known about the thread: a restart, or no request since ReasonUnavailable = "unavailable" // the route is failing in a row, or the provider refused this credential diff --git a/keepalive/substitute.go b/keepalive/substitute.go index 8136490ba..fef9dad20 100644 --- a/keepalive/substitute.go +++ b/keepalive/substitute.go @@ -413,6 +413,13 @@ func (k *Keeper) fireSweepSubstitute(j pingJob) (substituted bool) { substituted = false } }() + // No sweep question while the thread waits for its own tool (#858): that span stays warm until the + // result comes back, so a cut can never follow. The ordinary ping goes out. + if j.e.toolWait() { + slog.Debug("context-guru: sweep keep-alive question not offered: the thread waits for its own tool", + "tenant", tenantLabel(j.tenant), "session", j.session, "thread", j.thread) + return false + } dispatch, info, reason, ok := offload.SweepPingSubstitute(thread.Key(j.session, j.thread)) if !ok { slog.Debug("context-guru: sweep keep-alive question not offered", diff --git a/keepalive/types.go b/keepalive/types.go index 8d9f597fe..930e3ff0a 100644 --- a/keepalive/types.go +++ b/keepalive/types.go @@ -159,6 +159,18 @@ type Policy struct { // PredictorThreshold is the minimum probability PredictorID must return. Ignored // while PredictorID is "". PredictorThreshold float64 + // WaitPlan turns on the per-wait ping plan (waitplan.go, #858): the ping count follows what the + // thread waits for, time windows apply only to a wait for a person, the sweep question is not + // asked during a tool wait, and a thread with no tool call yet is not pinged. False is the + // behavior before #858: MaxPings for every wait. The zero Policy has it off, so a host that builds a + // Policy by hand keeps the old behavior until it asks. + WaitPlan bool + // ToolWaitMaxPings, BackgroundExtraPings and HumanExtraPings tune the plan. Zero means the shipped + // default (DefaultToolWaitMaxPings, DefaultBackgroundExtraPings, DefaultHumanExtraPings). Read only + // while WaitPlan is on. + ToolWaitMaxPings int + BackgroundExtraPings int + HumanExtraPings int // ManualHold is the host's reading of an opt-in on THIS request (the proxy: the // x-context-guru-ping header): keep this thread for the Ping button even though no automatic // ping will be sent for it. It never turns the schedule on, and never widens what is held: the diff --git a/keepalive/types_test.go b/keepalive/types_test.go index d43716b4b..292315dc0 100644 --- a/keepalive/types_test.go +++ b/keepalive/types_test.go @@ -4,6 +4,8 @@ import ( "math" "testing" "time" + + "github.com/rossoctl/context-guru/config" ) func TestPolicyCeilingResolvesZeroToTheDefault(t *testing.T) { @@ -46,3 +48,13 @@ func TestDisabledReadsTheKillSwitch(t *testing.T) { t.Error("empty CONTEXT_GURU_KEEPALIVE disabled the keep-alive") } } + +// The wait-plan defaults mirror config's; keepalive may not import config. +func TestWaitPlanDefaultsMatchConfig(t *testing.T) { + if DefaultToolWaitMaxPings != config.DefaultKeepAliveToolWaitMaxPings || + DefaultBackgroundExtraPings != config.DefaultKeepAliveBackgroundExtraPings || + DefaultHumanExtraPings != config.DefaultKeepAliveHumanExtraPings || + MaxWaitPings != config.MaxKeepAliveWaitPings { + t.Error("the keepalive and config wait-plan defaults drifted") + } +} diff --git a/keepalive/waitplan.go b/keepalive/waitplan.go new file mode 100644 index 000000000..ad02e39b8 --- /dev/null +++ b/keepalive/waitplan.go @@ -0,0 +1,119 @@ +package keepalive + +import ( + "strings" + + "github.com/rossoctl/context-guru/internal/waitcause" +) + +// The per-wait ping plan (#858). The ping count is chosen when a turn ends, from what the thread +// waits for. Metadata only: stop reason, tool names, the open-task counts and the shape of the +// thread id. "Base" is Policy.MaxPings. +// +// own tool (main or subagent) keep pinging until the result comes, at most 8 +// main, background work base + 3 +// main, a person base + 1 +// main, a teammate base +// subagent, its main agent base +// done, or no tool call yet on the thread 0 +// +// The plan applies to the 5-minute tier only. A 1-hour thread keeps its own schedule +// (Policy.ForOneHour), whose credential hold is already bounded at (K+1) x 59 min. +const ( + // DefaultToolWaitMaxPings is the most pings a tool wait gets. A tool almost never runs longer + // than 13 minutes (0 of 805 main waits locally, 5 of 5,202 on the hosted proxy), so in practice + // the count stays near the base. Mirrored by config.DefaultKeepAliveToolWaitMaxPings. + DefaultToolWaitMaxPings = 8 + // DefaultBackgroundExtraPings and DefaultHumanExtraPings are added to the base. Mirrored by + // config.DefaultKeepAliveBackgroundExtraPings and DefaultKeepAliveHumanExtraPings. + DefaultBackgroundExtraPings = 3 + DefaultHumanExtraPings = 1 + // MaxWaitPings bounds any plan result unless the base itself is higher. It equals the highest + // strategy ping count, so the credential hold stays at (11+1) x 290 s = 58 min or less. + MaxWaitPings = 11 +) + +// IsSubagentThread reports whether a thread id names a subagent. The main thread is "". +func IsSubagentThread(threadID string) bool { return strings.HasPrefix(threadID, "a:") } + +// WindowsApply reports whether a manager strategy's time windows apply to a thread that waits for +// cause. Only a wait for a person follows working hours. A tool, a background task, a teammate and +// a main agent answer at night as fast as in the day. A subagent's "human" wait (end_turn, or the +// SubagentHandback tool) is a wait for its main agent, not for a person. +func WindowsApply(cause string, subagent bool) bool { + switch cause { + case waitcause.WaitingOwnTool, waitcause.WaitingBackground, waitcause.WaitingTeammate: + return false + case waitcause.WaitingHuman: + return !subagent + } + return true +} + +// WaitPings is the ping count for a thread that waits for cause. With WaitPlan off it is MaxPings. +func (p Policy) WaitPings(cause string, subagent bool) int { + base := p.MaxPings + if !p.WaitPlan { + return base + } + n := base + switch cause { + case waitcause.WaitingOwnTool: + n = max(base, orDefault(p.ToolWaitMaxPings, DefaultToolWaitMaxPings)) + case waitcause.WaitingBackground: + if !subagent { + n = base + orDefault(p.BackgroundExtraPings, DefaultBackgroundExtraPings) + } + case waitcause.WaitingHuman: + if !subagent { + n = base + orDefault(p.HumanExtraPings, DefaultHumanExtraPings) + } + } + return min(n, max(base, MaxWaitPings)) +} + +func orDefault(v, def int) int { + if v <= 0 { + return def + } + return v +} + +// WaitingFor is the wait cause of a response, from its tool names and stop reason and the thread's +// open-task counts (Options.OpenTasks). The open counts are as of the last request the host's +// tracker observed, which can lag by one request: a stale count reads as "a person", the cheaper plan. +func (k *Keeper) WaitingFor(key Key, names []string, incomplete bool, stopReason string, usageOK bool) string { + if k == nil { + return waitcause.WaitingUnknown + } + bg, tm := 0, 0 + if k.o.OpenTasks != nil { + bg, tm = k.o.OpenTasks(key.Tenant, key.Session, key.Thread) + } + w, _ := waitcause.ClassifyWaitingForTerminal(terminalFor(k.o.TerminalTools, incomplete), + names, stopReason, false, usageOK, bg, tm) + // 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. + if len(names) == 0 && w == waitcause.WaitingHuman && toolStop(stopReason) { + return waitcause.WaitingOwnTool + } + return w +} + +func toolStop(stopReason string) bool { + switch stopReason { + case "tool_use", "tool_calls", "function_call": + return true + } + return false +} + +// 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. +func calledATool(names []string, incomplete bool, stopReason string) bool { + return len(names) > 0 || incomplete || toolStop(stopReason) +} + +// toolWait reports whether this entry's pings were planned for a wait on the thread's own tool. +func (e *kaEntry) toolWait() bool { return e.wait == waitcause.WaitingOwnTool } diff --git a/keepalive/waitplan_test.go b/keepalive/waitplan_test.go new file mode 100644 index 000000000..a24aea68c --- /dev/null +++ b/keepalive/waitplan_test.go @@ -0,0 +1,327 @@ +package keepalive + +import ( + "net/http" + "testing" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + + "github.com/rossoctl/context-guru/store" +) + +// resp is one finished response of a thread: the tool names it called and its stop reason. +type resp struct { + names []string + stop string +} + +var ( + toolCall = resp{[]string{"Bash"}, "tool_use"} + endTurn = resp{nil, "end_turn"} +) + +// planPolicy is kaPolicy with the per-wait plan on and its shipped defaults. +func planPolicy() Policy { + p := kaPolicy() + p.WaitPlan = true + return p +} + +// recordSeq records the responses in order on key, one second apart, the last at `at`. +func recordSeq(k *Keeper, key Key, pol Policy, body string, at time.Time, seq ...resp) { + for i, r := range seq { + ts := at.Add(-time.Duration(len(seq)-1-i) * time.Second) + k.Record(Record{Key: key, StartedAt: ts, Body: []byte(body), Provider: bschemas.Anthropic, + Route: "/v1/messages", Status: http.StatusOK, + Usage: Usage{CacheRead: 48576, CacheWrite: 1200, StopReason: r.stop}, + ToolUseNames: r.names, UsageOK: true, Policy: pol, Header: kaHeader(), Hold: fakeOf(k).hold(key)}) + } +} + +// plannedPings is the ping count the entry on key was planned for. +func plannedPings(t *testing.T, k *Keeper, key Key) int { + t.Helper() + v, ok := k.Entry(key) + if !ok { + t.Fatalf("no live entry for %+v", key) + } + return v.Policy.MaxPings +} + +// Every row of the issue's table, at the shipped defaults (base 2). +func TestWaitPlanPingBudgetPerWaitCause(t *testing.T) { + sub := Key{Tenant: "t1", Session: "sess-1", Thread: "a:abc"} + for _, tc := range []struct { + name string + key Key + seq []resp + bg, tm int + want int + wantEntry bool + }{ + {"main, own tool", sess1, []resp{toolCall, toolCall}, 0, 0, 8, true}, + {"subagent, own tool", sub, []resp{toolCall, toolCall}, 0, 0, 8, true}, + {"main, end_turn, background", sess1, []resp{toolCall, endTurn}, 1, 0, 5, true}, + {"main, end_turn, a person", sess1, []resp{toolCall, endTurn}, 0, 0, 3, true}, + {"main, ask-human tool", sess1, []resp{toolCall, {[]string{"AskUserQuestion"}, "tool_use"}}, 0, 0, 3, true}, + {"main, end_turn, teammate", sess1, []resp{toolCall, endTurn}, 0, 1, 2, true}, + {"subagent, end_turn", sub, []resp{toolCall, endTurn}, 0, 0, 2, true}, + {"subagent, SubagentHandback", sub, []resp{toolCall, {[]string{"SubagentHandback"}, "tool_use"}}, 0, 0, 2, true}, + {"subagent, end_turn, background", sub, []resp{toolCall, endTurn}, 1, 0, 2, true}, + {"done", sess1, []resp{toolCall, {[]string{"StructuredOutput"}, "tool_use"}}, 0, 0, 0, false}, + } { + t.Run(tc.name, func(t *testing.T) { + k, _, clock := testKeeperWith(t, func(o *Options) { + o.OpenTasks = func(string, string, string) (int, int) { return tc.bg, tc.tm } + }) + recordSeq(k, tc.key, planPolicy(), kaBody, clock.now(), tc.seq...) + if !tc.wantEntry { + if _, ok := k.Entry(tc.key); ok { + t.Fatal("a done thread is held") + } + return + } + if got := plannedPings(t, k, tc.key); got != tc.want { + t.Errorf("planned pings = %d, want %d", got, tc.want) + } + }) + } +} + +// A tool wait is pinged until the result comes, and stops at 8. +func TestWaitPlanToolWaitIsPingedUntilTheResultComesAtMostEight(t *testing.T) { + k, fd, clock := testKeeper(t) + recordSeq(k, sess1, planPolicy(), kaBody, clock.now(), toolCall, toolCall) + for i := 1; i <= 12; i++ { + k.Sweep(clock.advance(281 * time.Second)) + if i == 5 { + if n := fd.n(); n != 5 { + t.Fatalf("after 5 intervals: %d pings, want 5 (the base is 2)", n) + } + } + } + if n := fd.n(); n != 8 { + t.Fatalf("pings = %d, want 8 (the cap)", n) + } +} + +func TestWaitPlanResultArrivingStopsThePings(t *testing.T) { + k, fd, clock := testKeeper(t) + recordSeq(k, sess1, planPolicy(), kaBody, clock.now(), toolCall, toolCall) + for i := 0; i < 4; i++ { + k.Sweep(clock.advance(281 * time.Second)) + } + k.Arrive(sess1) // the tool result comes back + if _, ok := k.Entry(sess1); ok { + t.Fatal("entry still held after the next request arrived") + } + k.Sweep(clock.advance(281 * time.Second)) + if n := fd.n(); n != 4 { + t.Errorf("pings = %d, want 4", n) + } +} + +// The base is the floor: a configured tool-wait cap under the base never shortens it. +func TestWaitPlanToolWaitNeverBelowTheBase(t *testing.T) { + k, _, clock := testKeeper(t) + pol := planPolicy() + pol.MaxPings, pol.ToolWaitMaxPings = 10, 3 + recordSeq(k, sess1, pol, kaBody, clock.now(), toolCall, toolCall) + if got := plannedPings(t, k, sess1); got != 10 { + t.Errorf("planned = %d, want the base 10", got) + } +} + +// Configured extra pings replace the defaults. +func TestWaitPlanConfiguredCounts(t *testing.T) { + k, _, clock := testKeeperWith(t, func(o *Options) { + o.OpenTasks = func(string, string, string) (int, int) { return 1, 0 } + }) + pol := planPolicy() + pol.BackgroundExtraPings, pol.ToolWaitMaxPings = 5, 4 + recordSeq(k, sess1, pol, kaBody, clock.now(), toolCall, endTurn) + if got := plannedPings(t, k, sess1); got != 7 { + t.Errorf("background = %d, want 7", got) + } + k2, _, clock2 := testKeeper(t) + recordSeq(k2, sess1, pol, kaBody, clock2.now(), toolCall, toolCall) + if got := plannedPings(t, k2, sess1); got != 4 { + t.Errorf("tool wait = %d, want 4", got) + } +} + +// A plan result never passes 11 unless the base itself is higher. +func TestWaitPlanCapsAtEleven(t *testing.T) { + p := planPolicy() + p.MaxPings, p.ToolWaitMaxPings = 9, 50 + if got := p.WaitPings("own_tool", false); got != 11 { + t.Errorf("got %d, want 11", got) + } + p.MaxPings = 20 + if got := p.WaitPings("background", false); got != 20 { + t.Errorf("got %d, want the base 20", got) + } +} + +// A thread with no tool call yet gets no pings. After one tool call it does. +func TestWaitPlanNoToolCallYetGetsNoPings(t *testing.T) { + k, fd, clock := testKeeper(t) + recordSeq(k, sess1, planPolicy(), kaBody, clock.now(), endTurn, endTurn) + if n := k.Sweep(clock.advance(281 * time.Second)); n != 0 || fd.n() != 0 { + t.Fatalf("a thread with no tool call was pinged (%d, %d)", n, fd.n()) + } + k2, fd2, clock2 := testKeeper(t) + recordSeq(k2, sess1, planPolicy(), kaBody, clock2.now(), toolCall, endTurn) + if n := k2.Sweep(clock2.advance(281 * time.Second)); n != 1 || fd2.n() != 1 { + t.Fatalf("a thread that called a tool was not pinged (%d, %d)", n, fd2.n()) + } +} + +// The thread remembers a tool call from an earlier turn. +func TestWaitPlanEarlierToolCallCounts(t *testing.T) { + k, fd, clock := testKeeper(t) + recordSeq(k, sess1, planPolicy(), kaBody, clock.now().Add(-10*time.Second), toolCall) + recordSeq(k, sess1, planPolicy(), kaBody, clock.now(), endTurn, endTurn) + if n := k.Sweep(clock.advance(281 * time.Second)); n != 1 || fd.n() != 1 { + t.Fatalf("a thread with an earlier tool call was not pinged (%d, %d)", n, fd.n()) + } +} + +// The off switch: today's behavior. A fixed MaxPings for every wait, and the no-tool guard is off. +func TestWaitPlanOffIsTodaysBehavior(t *testing.T) { + k, fd, clock := testKeeper(t) + recordSeq(k, sess1, kaPolicy(), kaBody, clock.now(), toolCall, toolCall) + if got := plannedPings(t, k, sess1); got != 2 { + t.Fatalf("planned = %d, want the fixed 2", got) + } + for i := 0; i < 6; i++ { + k.Sweep(clock.advance(281 * time.Second)) + } + if fd.n() != 2 { + t.Errorf("pings = %d, want 2", fd.n()) + } + k2, fd2, clock2 := testKeeper(t) + recordSeq(k2, sess1, kaPolicy(), kaBody, clock2.now(), endTurn, endTurn) + if n := k2.Sweep(clock2.advance(281 * time.Second)); n != 1 || fd2.n() != 1 { + t.Fatalf("with the plan off, a thread with no tool call is pinged as before (%d, %d)", n, fd2.n()) + } +} + +// A 1-hour thread keeps its own bounded schedule: the plan does not stretch the hold. +func TestWaitPlanDoesNotChangeAOneHourThread(t *testing.T) { + k, _, clock := testKeeper(t) + body := ttlBody1h + for i, ts := range []time.Time{clock.now().Add(-time.Second), clock.now()} { + k.Record(Record{Key: sess1, StartedAt: ts, Body: []byte(body), Provider: bschemas.Anthropic, + Route: "/v1/messages", Status: http.StatusOK, + Usage: Usage{CacheRead: 48576, CacheWrite: 1200, CacheWrite1h: 1200, StopReason: "tool_use"}, + ToolUseNames: []string{"Bash"}, UsageOK: true, Policy: planPolicy(), Header: kaHeader(), + Hold: fakeOf(k).hold(sess1)}) + _ = i + } + v, ok := k.Entry(sess1) + if !ok || !v.TTL1h { + t.Fatalf("not a 1-hour entry: %+v ok=%v", v, ok) + } + if v.Policy.MaxPings > 3 { + t.Errorf("1-hour thread planned %d pings, want at most 3", v.Policy.MaxPings) + } +} + +// The sweep question is not asked while a thread waits for its own tool. It takes the planned ping on +// a tool_result span that began with end_turn. +func TestWaitPlanNoSweepQuestionDuringAToolWait(t *testing.T) { + k, fs, clock := testKeeper(t) + st := store.NewMemory(store.Options{MaxEntries: 400}) + asker := &sweepAsker{cacheRead: 48576} + registerSweep(t, st, asker, "claude-sonnet-5", sweepConversation(3)) + recordSeq(k, sess1, planPolicy(), kaToolBody, clock.now(), toolCall, toolCall) + k.Sweep(clock.advance(281 * time.Second)) + waitPings(t, k, 1) + if asker.calls.Load() != 0 || fs.n() != 1 { + t.Fatalf("tool wait: asks %d, plain %d; want 0 and 1", asker.calls.Load(), fs.n()) + } +} + +func TestWaitPlanSweepQuestionOnAnEndTurnSpan(t *testing.T) { + k, fs, clock := testKeeper(t) + st := store.NewMemory(store.Options{MaxEntries: 400}) + asker := &sweepAsker{cacheRead: 48576} + registerSweep(t, st, asker, "claude-sonnet-5", sweepConversation(3)) + recordSeq(k, sess1, planPolicy(), kaToolBody, clock.now(), toolCall, endTurn) + k.Sweep(clock.advance(281 * time.Second)) + waitSweepOutcome(t, k, fs) + waitPings(t, k, 1) + if asker.calls.Load() != 1 || fs.n() != 0 { + t.Fatalf("end_turn span: asks %d, plain %d; want 1 and 0", asker.calls.Load(), fs.n()) + } +} + +// The extra slot after the last planned ping: no question on a tool wait, one on an end_turn span. +func TestWaitPlanNoExtraQuestionAfterAToolWaitButOneAfterEndTurn(t *testing.T) { + for _, tc := range []struct { + name string + seq []resp + pings int + asks int64 + }{ + {"tool wait", []resp{toolCall, toolCall}, 2, 0}, + {"end_turn", []resp{toolCall, endTurn}, 3, 1}, + } { + t.Run(tc.name, func(t *testing.T) { + k, fs, clock := testKeeper(t) + st := store.NewMemory(store.Options{MaxEntries: 400}) + asker := &sweepAsker{cacheRead: 48576} + registerSweep(t, st, asker, "claude-sonnet-5", sweepConversation(3)) + pol := planPolicy() + pol.ToolWaitMaxPings = 2 // the tool-wait cap, kept at the base so the slot comes soon + recordSeq(k, sess1, pol, kaBody, clock.now(), tc.seq...) + for i := 1; i <= tc.pings; i++ { + k.Sweep(clock.advance(281 * time.Second)) + waitPings(t, k, int64(i)) + } + k.Sweep(clock.advance(250 * time.Second)) + time.Sleep(20 * time.Millisecond) + if asker.calls.Load() != tc.asks || fs.n() != tc.pings { + t.Errorf("asks %d, plain %d; want %d and %d", asker.calls.Load(), fs.n(), tc.asks, tc.pings) + } + }) + } +} + +func TestWindowsApply(t *testing.T) { + for _, tc := range []struct { + cause string + sub bool + want bool + }{ + {"human", false, true}, {"human", true, false}, {"own_tool", false, false}, {"own_tool", true, false}, + {"background", false, false}, {"teammate", false, false}, {"done", false, true}, {"unknown", false, true}, + {"retry", false, true}, + } { + if got := WindowsApply(tc.cause, tc.sub); got != tc.want { + t.Errorf("WindowsApply(%q, sub=%v) = %v, want %v", tc.cause, tc.sub, got, tc.want) + } + } +} + +// The plan the dashboard reads (Plans: pings_left, next ping) is the per-wait one. The hold of the +// longest plan stays inside the strategy bound: (8+1) x 240 s = 36 min; at the cap of 11 and 290 s, 58 min. +func TestWaitPlanShowsInThePlanTheDashboardReads(t *testing.T) { + k, _, clock := testKeeper(t) + start := clock.now() + recordSeq(k, sess1, planPolicy(), kaBody, start, toolCall, toolCall) + pl := k.Plans("t1", false, "sess-1")[""] + if pl.PingsLeft != 8 || !pl.NextPing.Equal(start.Add(280*time.Second)) { + t.Fatalf("plan = %+v, want 8 pings left", pl) + } + if h := time.Duration(MaxWaitPings+1) * 290 * time.Second; h > time.Hour { + t.Errorf("longest plan holds %v, over the 1 h override ceiling", h) + } + k2, _, clock2 := testKeeper(t) + recordSeq(k2, sess1, planPolicy(), kaBody, clock2.now(), endTurn, endTurn) + if len(k2.Plans("t1", false, "sess-1")) != 0 { + t.Error("a thread with no tool call yet shows a plan") + } +} diff --git a/proxy/keepalive.go b/proxy/keepalive.go index e7a458c84..ffd4ecf1c 100644 --- a/proxy/keepalive.go +++ b/proxy/keepalive.go @@ -369,7 +369,15 @@ func (k *keeper) record(tn *Tenancy, session, threadID string, startedAt time.Ti pdec, pdecOK = k.predictiveDecide(arr, !tn.Cache.KeepAlive, tn.ID, session, threadID, string(provider), startedAt, body, r.UserAgent(), u) } - pol, applied = k.resolvePolicy(tn, session, provider, pol, pdec, pdecOK) + // Time windows gate only a wait for a person (#858), so the strategy is resolved with the wait + // cause of the response that ended this turn. + ignoreWindows := false + if pol.WaitPlan && tracked { + cause := k.Keeper.WaitingFor(keepalive.Key{Tenant: tn.ID, Session: session, Thread: threadID}, + u.ToolUseNames, u.ToolUseIncomplete, u.StopReason, usageOK) + ignoreWindows = !keepalive.WindowsApply(cause, keepalive.IsSubagentThread(threadID)) + } + pol, applied = k.resolvePolicyWindows(tn, session, provider, pol, pdec, pdecOK, ignoreWindows) // The Ping button's opt-in (manualping.go): this REQUEST asked to be held for it. Set only here, // on the path that already passed the no-audit-sink rule, the client-ended check and the kill // switch, so the header can never widen what those refuse. It does not turn the schedule on. @@ -407,7 +415,8 @@ func (k *keeper) canPing(tn *Tenancy, session string, provider bschemas.ModelPro provider == bschemas.OpenAI || session == "" { return false } - pol, _ := k.resolvePolicy(tn, session, provider, tn.Cache, predDecision{}, false) + // Windows are ignored: the wait cause is not known yet, and a non-person wait is pinged at any hour (#858). + pol, _ := k.resolvePolicyWindows(tn, session, provider, tn.Cache, predDecision{}, false, tn.Cache.WaitPlan) return pol.On() } @@ -415,6 +424,12 @@ func (k *keeper) canPing(tn *Tenancy, session string, provider bschemas.ModelPro // policy — the old record's policy block, unchanged. func (k *keeper) resolvePolicy(tn *Tenancy, session string, provider bschemas.ModelProvider, pol CachePolicy, pdec predDecision, pdecOK bool) (CachePolicy, string) { + return k.resolvePolicyWindows(tn, session, provider, pol, pdec, pdecOK, false) +} + +// resolvePolicyWindows is resolvePolicy with the strategy's time windows optionally ignored (#858). +func (k *keeper) resolvePolicyWindows(tn *Tenancy, session string, provider bschemas.ModelProvider, + pol CachePolicy, pdec predDecision, pdecOK bool, ignoreWindows bool) (CachePolicy, string) { // The manager-controlled strategy layer, resolved once here and generalizing the one // hook the session override already had — see keepalivestrategy.go's applyStrategy. // A matching ACTIVE strategy forces KeepAlive on and replaces Idle/MaxPings/ @@ -422,11 +437,12 @@ func (k *keeper) resolvePolicy(tn *Tenancy, session string, provider bschemas.Mo // override, which still wins on top and still cannot widen MaxUSDPerPing. applied := "" if provider != bschemas.OpenAI { - pol, applied = k.applyStrategy(tn.ID, pol, k.now()) + pol, applied = k.applyStrategyWindows(tn.ID, pol, k.now(), ignoreWindows) // An ACTIVE campaign decides the ping count unless a manager strategy matched (the more - // explicit instruction). + // explicit instruction). The campaign owns the count, so the per-wait plan is off for it (#858). if pdecOK && applied == "" && pdec.active { pol, applied = pdec.apply(pol), campaignStrategyID(pdec.campaign.meta.ID) + pol.WaitPlan = false } } // A per-session manual override, if one is armed and unexpired. Everything below reads diff --git a/proxy/keepaliveoverride.go b/proxy/keepaliveoverride.go index a7f547706..6db63ffa2 100644 --- a/proxy/keepaliveoverride.go +++ b/proxy/keepaliveoverride.go @@ -184,6 +184,9 @@ func (k *keeper) overrideFor(tenantID, session string, pol CachePolicy) CachePol pol.KeepAlive = true pol.Idle, pol.MaxPings = o.pol.Idle, o.pol.MaxPings pol.MinPrefixTokens = o.pol.MinPrefixTokens + // The person armed this exact count and was shown the hold it buys: the per-wait plan (#858) + // does not change it. + pol.WaitPlan = false // The override's ping count bounds a 1-hour thread too, but only downward: its 60-290 s band // cannot express an hour, so the 1-hour timing stays the account's, and the hold the person was // shown is then (K+1) x 59 min at most. 0 means the proxy default, which is resolved here so diff --git a/proxy/keepalivestrategy.go b/proxy/keepalivestrategy.go index b6dd37634..3686147f2 100644 --- a/proxy/keepalivestrategy.go +++ b/proxy/keepalivestrategy.go @@ -74,13 +74,24 @@ func (k *keeper) clockNow() time.Time { // resolved at record time) and resolveHeadTTL (the write-tier half, resolved on the // request path) need, so the two can never disagree about which strategy won. func (k *keeper) bestStrategyFor(tenantID string, now time.Time) *tenant.Strategy { + return k.bestStrategy(tenantID, now, false) +} + +// bestStrategy is bestStrategyFor with the time windows optionally ignored (#858): the +// highest-priority ACTIVE strategy that targets tenantID, as if its window matched. The same +// priority rule (betterStrategy) picks among them. +func (k *keeper) bestStrategy(tenantID string, now time.Time, ignoreWindows bool) *tenant.Strategy { k.mu.Lock() strategies := k.strategies k.mu.Unlock() var best *tenant.Strategy for i := range strategies { s := &strategies[i] - if !s.Matches(now, tenantID) { + if ignoreWindows { + if !s.Active || !s.Target.Covers(tenantID) { + continue + } + } else if !s.Matches(now, tenantID) { continue } if best == nil || betterStrategy(s, best) { @@ -121,7 +132,16 @@ func (k *keeper) strategyTZFor(tenantID string) string { // for its whole lifetime, exactly as overrideFor's own resolution already is: a strategy // whose window closes mid-hold does not retroactively un-arm a ping already scheduled. func (k *keeper) applyStrategy(tenantID string, pol CachePolicy, now time.Time) (CachePolicy, string) { - best := k.bestStrategyFor(tenantID, now) + return k.applyStrategyWindows(tenantID, pol, now, false) +} + +// applyStrategyWindows is applyStrategy with the time windows optionally ignored (#858). Windows +// describe when a person answers, so they gate only a wait for a person. For any other wait the +// caller passes ignoreWindows, and the strategy that targets the tenant applies at any hour; with no +// such strategy the account config applies, as before. The wait cause is known here because the +// strategy is resolved at record time, from the response that ended the turn. +func (k *keeper) applyStrategyWindows(tenantID string, pol CachePolicy, now time.Time, ignoreWindows bool) (CachePolicy, string) { + best := k.bestStrategy(tenantID, now, ignoreWindows) if best == nil { return pol, "" } diff --git a/proxy/keepalivewaitplan_test.go b/proxy/keepalivewaitplan_test.go new file mode 100644 index 000000000..9f619a416 --- /dev/null +++ b/proxy/keepalivewaitplan_test.go @@ -0,0 +1,121 @@ +package proxy + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + bschemas "github.com/maximhq/bifrost/core/schemas" + + "github.com/rossoctl/context-guru/keepalive" + "github.com/rossoctl/context-guru/tenant" +) + +// recordResp records two turns of "sess-1" on thread th for tenant t1, the second ending as u says. +func recordResp(t *testing.T, k *keeper, pol CachePolicy, th string, at time.Time, u Usage) { + t.Helper() + tn := &Tenancy{ID: "t1", Cache: pol} + for i, ts := range []time.Time{at.Add(-time.Second), at} { + r := httptest.NewRequest(http.MethodPost, "/anthropic/v1/messages", strings.NewReader("")) + r.Header.Set("Authorization", "Bearer sk-caller-secret") + use := Usage{CacheRead: 48576, CacheWrite: 1200, StopReason: "tool_use", ToolUseNames: []string{"Bash"}} + if i == 1 { + use = u + use.CacheRead, use.CacheWrite = 48576, 1200 + } + k.record(tn, "sess-1", th, ts, []byte(kaBody), upstream{base: "http://up", path: "/v1/messages"}, r, + bschemas.Anthropic, "/v1/messages", http.StatusOK, use, true) + } +} + +// #858: a strategy whose window is shut still pings a tool wait (any hour), and a wait for a person +// outside the window gets today's result: the account config, which here is off. +func TestStrategyWindowsApplyOnlyToAWaitForAPerson(t *testing.T) { + k, _, clock := testKeeper(t, Limits{}) + now := clock.now() + closed := tenant.Strategy{ + ID: "s", Active: true, IdleSeconds: 240, MaxPings: 4, MinPrefixTokens: 20000, MaxUSDPerPing: 0.5, + Windows: []tenant.Window{{Start: "00:00", End: "00:01"}}, + Target: tenant.Target{Mode: tenant.TargetList, TenantIDs: []string{"t1"}}, UpdatedAt: now, + } + k.setStrategies([]tenant.Strategy{closed}) + account := CachePolicy{WaitPlan: true} // keep-alive off on the account + + for _, tc := range []struct { + name string + u Usage + wantHeld bool + wantPings int + }{ + {"tool wait, outside the window", Usage{StopReason: "tool_use", ToolUseNames: []string{"Bash"}}, true, 8}, + {"background wait is not asked here: end_turn is a person", Usage{StopReason: "end_turn"}, false, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + k.Keeper.Forget("t1") + recordResp(t, k, account, "", now, tc.u) + v, ok := k.Keeper.Entry(keepalive.Key{Tenant: "t1", Session: "sess-1"}) + if ok != tc.wantHeld { + t.Fatalf("held = %v, want %v", ok, tc.wantHeld) + } + if ok { + if v.Tag != "s" { + t.Errorf("tag = %q, want the strategy", v.Tag) + } + if v.Policy.MaxPings != tc.wantPings { + t.Errorf("pings = %d, want %d", v.Policy.MaxPings, tc.wantPings) + } + } + }) + } + + // With the plan off, even a tool wait outside the window is not pinged (today). + k.Keeper.Forget("t1") + recordResp(t, k, CachePolicy{}, "", now, Usage{StopReason: "tool_use", ToolUseNames: []string{"Bash"}}) + if _, ok := k.Keeper.Entry(keepalive.Key{Tenant: "t1", Session: "sess-1"}); ok { + t.Fatal("with the plan off, a tool wait outside the window was held") + } +} + +// A subagent that waits for its main agent is not a wait for a person: the windows are ignored. +func TestStrategyWindowsIgnoredForASubagentEndTurn(t *testing.T) { + k, _, clock := testKeeper(t, Limits{}) + now := clock.now() + k.setStrategies([]tenant.Strategy{{ + ID: "s", Active: true, IdleSeconds: 240, MaxPings: 2, MinPrefixTokens: 20000, MaxUSDPerPing: 0.5, + Windows: []tenant.Window{{Start: "00:00", End: "00:01"}}, + Target: tenant.Target{Mode: tenant.TargetAll}, UpdatedAt: now, + }}) + recordResp(t, k, CachePolicy{WaitPlan: true}, "a:sub1", now, Usage{StopReason: "end_turn"}) + v, ok := k.Keeper.Entry(keepalive.Key{Tenant: "t1", Session: "sess-1", Thread: "a:sub1"}) + if !ok || v.Policy.MaxPings != 2 { + t.Fatalf("subagent end_turn: held=%v policy=%+v, want held with 2 pings", ok, v.Policy) + } +} + +// The strategy picked ignoring windows follows the same priority rule, and no targeting strategy +// leaves the account config. +func TestBestStrategyIgnoringWindowsKeepsPriorityAndTarget(t *testing.T) { + k, _, clock := testKeeper(t, Limits{}) + now := clock.now() + win := []tenant.Window{{Start: "00:00", End: "00:01"}} + k.setStrategies([]tenant.Strategy{ + {ID: "all", Active: true, Windows: win, Target: tenant.Target{Mode: tenant.TargetAll}, UpdatedAt: now}, + {ID: "list", Active: true, Windows: win, Target: tenant.Target{Mode: tenant.TargetList, TenantIDs: []string{"t1"}}, UpdatedAt: now.Add(-time.Hour)}, + {ID: "off", Active: false, Windows: win, Target: tenant.Target{Mode: tenant.TargetList, TenantIDs: []string{"t2"}}, UpdatedAt: now}, + }) + if b := k.bestStrategy("t1", now, true); b == nil || b.ID != "list" { + t.Errorf("t1 got %+v, want list", b) + } + if b := k.bestStrategy("t2", now, true); b == nil || b.ID != "all" { + t.Errorf("t2 got %+v, want all (the list strategy is inactive)", b) + } + if b := k.bestStrategy("t1", now, false); b != nil { + t.Errorf("windows respected: got %+v, want none", b) + } + k.setStrategies(nil) + if b := k.bestStrategy("t1", now, true); b != nil { + t.Errorf("no strategies: got %+v", b) + } +} From 118f80598bf611a697c19d07f49298cc937cc36e Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Mon, 2 Nov 2026 22:08:06 +0200 Subject: [PATCH 2/5] test(config): pin the 20000 default minimum prefix Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- config/config_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/config/config_test.go b/config/config_test.go index 4d6ac3f89..316351fea 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -266,12 +266,12 @@ func TestPresetLadderSummarizerModes(t *testing.T) { } } -// The recommended keep-alive policy (#435, B-policy-r2) is the default; pin the four knobs. +// The recommended keep-alive policy (#435, B-policy-r2; floor 20000 since #858) is the default; pin the four knobs. func TestKeepAliveRecommendedDefaults(t *testing.T) { r := CacheConfig{}.Resolved() if r.KeepAliveIdleSeconds != 240 || r.KeepAliveMaxPings != 2 || - r.KeepAliveMinPrefixTokens != 80000 || r.KeepAliveMaxUSDPerPing != 0.50 { - t.Errorf("Resolved() = idle %d, K %d, floor %d, ceiling %v; want 240, 2, 80000, 0.50", + r.KeepAliveMinPrefixTokens != 20000 || r.KeepAliveMaxUSDPerPing != 0.50 { + t.Errorf("Resolved() = idle %d, K %d, floor %d, ceiling %v; want 240, 2, 20000, 0.50", r.KeepAliveIdleSeconds, r.KeepAliveMaxPings, r.KeepAliveMinPrefixTokens, r.KeepAliveMaxUSDPerPing) } } From 9d31e0e0fefacd59544377b636c08deee1c30b17 Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Mon, 2 Nov 2026 22:27:30 +0200 Subject: [PATCH 3/5] test(dash): move the replay goldens to the 20000 floor Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- dash/testdata/golden/ka_calc.json | 8 ++++---- dash/testdata/golden/ka_recommend.json | 4 ++-- docs/how-to/cache-keepalive.md | 2 +- keepalive/waitplan.go | 2 +- keepalive/waitplan_test.go | 4 ++-- 5 files changed, 10 insertions(+), 10 deletions(-) diff --git a/dash/testdata/golden/ka_calc.json b/dash/testdata/golden/ka_calc.json index 6d91a3807..0bbe382c9 100644 --- a/dash/testdata/golden/ka_calc.json +++ b/dash/testdata/golden/ka_calc.json @@ -15,7 +15,7 @@ "convertible_usd": 0.6000000000000004, "coverage_seconds": 580, "max_pings": 1, - "pings": 0, + "pings": 281, "share_of_addressable_pct": 70.75471698113208 }, { @@ -24,7 +24,7 @@ "coverage_seconds": 860, "current": true, "max_pings": 2, - "pings": 0, + "pings": 433, "share_of_addressable_pct": 70.75471698113208 }, { @@ -32,7 +32,7 @@ "convertible_usd": 0.6000000000000004, "coverage_seconds": 1140, "max_pings": 3, - "pings": 0, + "pings": 585, "share_of_addressable_pct": 70.75471698113208 }, { @@ -40,7 +40,7 @@ "convertible_usd": 0.6000000000000004, "coverage_seconds": 1420, "max_pings": 4, - "pings": 0, + "pings": 737, "share_of_addressable_pct": 70.75471698113208 } ] diff --git a/dash/testdata/golden/ka_recommend.json b/dash/testdata/golden/ka_recommend.json index a31c288e2..91f3e7a92 100644 --- a/dash/testdata/golden/ka_recommend.json +++ b/dash/testdata/golden/ka_recommend.json @@ -1,9 +1,9 @@ { - "hi_usd": 0.5299840000000002, + "hi_usd": 0.33678400000000003, "idle_seconds": 240, "keepalive_recorded_from": 1767606380000, "keepalive_recorded_rows": 585, - "lo_usd": 0.40297600000000033, + "lo_usd": 0.22627199999999986, "max_pings": 2, "n": 204, "requests": 591, diff --git a/docs/how-to/cache-keepalive.md b/docs/how-to/cache-keepalive.md index 1a1914bd6..5c5dd7dd3 100644 --- a/docs/how-to/cache-keepalive.md +++ b/docs/how-to/cache-keepalive.md @@ -92,7 +92,7 @@ open-task counts, and whether the thread id starts with `a:` (a subagent). - **The sweep question** is not asked while a thread waits for its own tool. That span stays warm until the result comes back, so a cut can never follow. It is asked on a span that began with `end_turn`. The rule "keep-alive never shortens the cache strategy" is unchanged. -- **The hold limit** is still `(K+1) x X` with K at most 11 and X under 290 s, so at most 58 minutes. A +- **The hold limit** is still `(K+1) x X` with the plan at most 11 pings (unless K itself is higher) and X under 300 s, so at most 60 minutes. A tool wait of 8 pings at 240 to 280 s holds at most 42 minutes. The plan never goes above 11 pings, unless K itself is higher, and it never lowers K. - **Not covered:** a 1-hour thread keeps its own schedule. A session override or an active learned diff --git a/keepalive/waitplan.go b/keepalive/waitplan.go index ad02e39b8..6dcbbc4b7 100644 --- a/keepalive/waitplan.go +++ b/keepalive/waitplan.go @@ -29,7 +29,7 @@ const ( DefaultBackgroundExtraPings = 3 DefaultHumanExtraPings = 1 // MaxWaitPings bounds any plan result unless the base itself is higher. It equals the highest - // strategy ping count, so the credential hold stays at (11+1) x 290 s = 58 min or less. + // strategy ping count, so the credential hold stays at (11+1) x 299 s = 60 min or less. MaxWaitPings = 11 ) diff --git a/keepalive/waitplan_test.go b/keepalive/waitplan_test.go index a24aea68c..30739fb32 100644 --- a/keepalive/waitplan_test.go +++ b/keepalive/waitplan_test.go @@ -307,7 +307,7 @@ func TestWindowsApply(t *testing.T) { } // The plan the dashboard reads (Plans: pings_left, next ping) is the per-wait one. The hold of the -// longest plan stays inside the strategy bound: (8+1) x 240 s = 36 min; at the cap of 11 and 290 s, 58 min. +// longest plan stays inside the strategy bound: (8+1) x 240 s = 36 min; at the cap of 11 and 299 s, 60 min. func TestWaitPlanShowsInThePlanTheDashboardReads(t *testing.T) { k, _, clock := testKeeper(t) start := clock.now() @@ -316,7 +316,7 @@ func TestWaitPlanShowsInThePlanTheDashboardReads(t *testing.T) { if pl.PingsLeft != 8 || !pl.NextPing.Equal(start.Add(280*time.Second)) { t.Fatalf("plan = %+v, want 8 pings left", pl) } - if h := time.Duration(MaxWaitPings+1) * 290 * time.Second; h > time.Hour { + if h := time.Duration(MaxWaitPings+1) * 299 * time.Second; h > time.Hour { t.Errorf("longest plan holds %v, over the 1 h override ceiling", h) } k2, _, clock2 := testKeeper(t) From 84957948e6be1bfe930a8e29162471c44d58a09e Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Mon, 2 Nov 2026 22:56:49 +0200 Subject: [PATCH 4/5] fix(keepalive): address round-1 review of the wait plan - canPing keeps the strategy windows and refuses a thread with no tool call yet, so no sweep copy is kept for a thread that is never pinged. - A tool stop with no names always gets the tool budget, also with open background tasks. Tests cover the spliced-window rows. - A strategy matched only because its window was ignored never displaces an active campaign. - The dash calculator and recommendation model a fixed count. They now say so (plan: fixed_k), and no rung is marked current. Follow-up: model the plan per span. - Fix stale 80k comments and the 80000 doc example. Update the bestStrategyFor comment. Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- dash/keepalive.go | 26 +++++++---- dash/keepalivetab_test.go | 16 ++++--- dash/testdata/golden/ka_calc.json | 2 +- dash/ui/app.js | 8 ++-- docs/how-to/kv-cache-keepalive-budget.md | 2 +- keepalive/keeper.go | 2 +- keepalive/waitplan.go | 14 +++++- keepalive/waitplan_test.go | 20 +++++++++ proxy/keepalive.go | 25 ++++++++--- proxy/keepalivehost_test.go | 12 ++--- proxy/keepalivestrategy.go | 5 ++- proxy/keepalivewaitplan_test.go | 56 ++++++++++++++++++++++++ proxy/modes.go | 2 +- 13 files changed, 157 insertions(+), 33 deletions(-) diff --git a/dash/keepalive.go b/dash/keepalive.go index 4e96ac4d4..a681b2f31 100644 --- a/dash/keepalive.go +++ b/dash/keepalive.go @@ -550,7 +550,7 @@ type KeepAliveBehaviour struct { Gaps []AccountGaps `json:"account_gaps"` // PrefixP10/P50/P90 in tokens, over the lapsed entries. p50 is 292,527 on the corpus and // 347 of 356 are past 20k — which is the measured reason a `prefix >= 20k` gate - // costs almost nothing at 20k (the pre-#435 gate; the shipped floor is now 80k), and the panel says so. + // costs almost nothing at 20k (the pre-#435 gate; the shipped floor is 20k since #858), and the panel says so. PrefixP10 int64 `json:"prefix_p10"` PrefixP50 int64 `json:"prefix_p50"` PrefixP90 int64 `json:"prefix_p90"` @@ -570,7 +570,7 @@ type KeepAliveBehaviour struct { // gapEdges are the idle-gap bands' fixed edges in seconds. The last is unbounded. var gapEdges = []float64{0, 300, 580, 600, 1800, 3600, 14400} -// prefixEdges are the billed-prefix bands' fixed edges in tokens. 20,000 was the pre-#435 gate (the shipped floor is now 80,000). +// prefixEdges are the billed-prefix bands' fixed edges in tokens. 20,000 was the pre-#435 gate and is the shipped floor again since #858 (80,000 in between). var prefixEdges = []float64{0, 20000, 50000, 100000, 200000, 400000, 800000} // bandLabels renders the edge list as human labels, with a unit formatter. @@ -1306,9 +1306,15 @@ func (s pingSpan) pings(idleSeconds float64, maxPings int) int { // 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 pre-#435 default; the shipped floor is now 80k) and not at the account's own configured floor, because this +// 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 80k so the page stays self-consistent. +// 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) { cond, args := f.where() // LEAD, not LAG: a span belongs to the request that OPENS it, which is the request the gate @@ -1377,6 +1383,8 @@ 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 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". PrefixSource string `json:"prefix_source,omitempty"` @@ -1393,11 +1401,11 @@ type KeepAliveCalc struct { // The rates themselves are the operator's price list; a model missing from it leaves every // dollar field omitted (Priced above). func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model string, - price func(string) (modelinfo.Price, bool), currentPings int) (*KeepAliveCalc, error) { + price func(string) (modelinfo.Price, bool), _ int) (*KeepAliveCalc, error) { if idleSeconds <= 0 { idleSeconds = recIdleSeconds } - out := &KeepAliveCalc{IdleSeconds: idleSeconds, Prefix: prefix, Model: model, Rows: []CalcRow{}} + out := &KeepAliveCalc{IdleSeconds: idleSeconds, Prefix: prefix, Model: model, Rows: []CalcRow{}, Plan: "fixed_k"} var p modelinfo.Price if price != nil && model != "" { if pr, ok := price(model); ok && !pr.Zero() { @@ -1460,7 +1468,8 @@ func (d *DB) KeepAliveCalc(f Filter, idleSeconds float64, prefix int64, model st // 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), - Current: k == currentPings} + // 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++ @@ -1604,7 +1613,8 @@ type KeepAliveRecommendation struct { KeepAliveCoverage } -// The shipped policy, which is also what is recommended when the rule admits an account. +// 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. const ( recIdleSeconds = 240 recMaxPings = 2 diff --git a/dash/keepalivetab_test.go b/dash/keepalivetab_test.go index fb3654baf..743ff9cfe 100644 --- a/dash/keepalivetab_test.go +++ b/dash/keepalivetab_test.go @@ -507,9 +507,15 @@ 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) } - // And the current row is marked, which is what the panel emphasises. - if !out.Rows[1].Current { - t.Error("K=2 was not marked as the current policy") + // 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. + 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 out.Plan != "fixed_k" { + t.Errorf("plan = %q, want fixed_k", out.Plan) } } @@ -822,12 +828,12 @@ func TestTheCalculatorChargesThePingsAPolicyWouldActuallySend(t *testing.T) { const t0 = int64(1_700_000_000_000) sec := func(n int64) int64 { return n * 1000 } evs := []*Event{ - // A gated session: prefix 100k throughout (above the 80k floor). turn 0, then a 300 s gap, then a 1000 s gap, + // A gated session: prefix 100k throughout (above the 20k floor). turn 0, then a 300 s gap, then a 1000 s gap, // then nothing — the last request opens a span a live policy pings in and never closes. bigPrefix(kaAgent(t0, "big", 0.10)), bigPrefix(kaAgent(t0+sec(300), "big", 0.10)), bigPrefix(kaAgent(t0+sec(1300), "big", 0.10)), - // Below the 80k prefix floor: the shipped gate never pings this session at all. + // Below the 20k prefix floor: the shipped gate never pings this session at all. smallPrefix(kaAgent(t0, "small", 0.10)), smallPrefix(kaAgent(t0+sec(1000), "small", 0.10)), } diff --git a/dash/testdata/golden/ka_calc.json b/dash/testdata/golden/ka_calc.json index 0bbe382c9..7505ddf8c 100644 --- a/dash/testdata/golden/ka_calc.json +++ b/dash/testdata/golden/ka_calc.json @@ -5,6 +5,7 @@ "keepalive_recorded_from": 1767606380000, "keepalive_recorded_rows": 585, "model": "m-unpriced", + "plan": "fixed_k", "prefix_source": "account_median", "prefix_tokens": 30998, "priced": false, @@ -22,7 +23,6 @@ "convertible_misses": 112, "convertible_usd": 0.6000000000000004, "coverage_seconds": 860, - "current": true, "max_pings": 2, "pings": 433, "share_of_addressable_pct": 70.75471698113208 diff --git a/dash/ui/app.js b/dash/ui/app.js index 962f50796..cdb6ca74f 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: 'The shipped default, offered only when your own history can tell its effect apart ' + 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 ' + '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 ' @@ -9597,10 +9597,12 @@ async function loadKACalc() { host.appendChild(el('p', { class: 'hint' }, 'Each row is one cap on pings. 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.')); + '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.')); host.appendChild(el('details', { class: 'why' }, el('summary', {}, 'Technical details'), el('p', {}, - 'PINGS is every ping the shipped policy would have sent over this window, gated as the ' + + 'PINGS is every ping a fixed-count 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 ' + diff --git a/docs/how-to/kv-cache-keepalive-budget.md b/docs/how-to/kv-cache-keepalive-budget.md index 712e96b1e..49cd83c23 100644 --- a/docs/how-to/kv-cache-keepalive-budget.md +++ b/docs/how-to/kv-cache-keepalive-budget.md @@ -30,7 +30,7 @@ cache: keepalive: true keepalive_idle_seconds: 240 # ping interval — keep this under the tier's TTL keepalive_max_pings: 2 # flat cap per idle span - keepalive_min_prefix_tokens: 80000 # skip small prefixes — a ping's fixed cost isn't worth it below this + keepalive_min_prefix_tokens: 20000 # skip small prefixes — a ping's fixed cost isn't worth it below this keepalive_max_usd_per_ping: 0.50 # per-ping cost ceiling ``` diff --git a/keepalive/keeper.go b/keepalive/keeper.go index 371692ed1..40ab5ebaa 100644 --- a/keepalive/keeper.go +++ b/keepalive/keeper.go @@ -895,7 +895,7 @@ func (k *Keeper) Record(r Record) { pol = pol.ForOneHour() } // The per-wait ping plan (#858): the count follows what the thread waits for. The wait cause is read - // for every thread the plan covers (the sweep gate and the dashboard use it); the COUNT changes only + // for every thread the plan covers (the sweep gate uses it); the COUNT changes only // on the 5-minute tier. A 1-hour thread keeps its own bounded schedule, and OpenAI pings 28 minutes // apart, so more pings would hold the credential for hours. A strategy that asks for the 1-hour head // tier carries a 1-hour Idle even when the gateway granted 5 minutes; its count is the manager's. diff --git a/keepalive/waitplan.go b/keepalive/waitplan.go index 6dcbbc4b7..7c7d5f1f0 100644 --- a/keepalive/waitplan.go +++ b/keepalive/waitplan.go @@ -94,7 +94,8 @@ func (k *Keeper) WaitingFor(key Key, names []string, incomplete bool, stopReason names, stopReason, false, usageOK, bg, tm) // 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. - if len(names) == 0 && w == waitcause.WaitingHuman && toolStop(stopReason) { + // 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 waitcause.WaitingOwnTool } return w @@ -117,3 +118,14 @@ func calledATool(names []string, incomplete bool, stopReason string) bool { // toolWait reports whether this entry's pings were planned for a wait on the thread's own tool. func (e *kaEntry) toolWait() bool { return e.wait == waitcause.WaitingOwnTool } + +// ToolSeen reports whether a response of the thread has called a tool (#858). Cheap: one map read under +// the keeper lock. The host uses it to keep nothing for a thread that the plan will not ping. +func (k *Keeper) ToolSeen(key Key) bool { + if k == nil { + return false + } + k.mu.Lock() + defer k.mu.Unlock() + return k.toolSeen[key.id()] +} diff --git a/keepalive/waitplan_test.go b/keepalive/waitplan_test.go index 30739fb32..3e20f2841 100644 --- a/keepalive/waitplan_test.go +++ b/keepalive/waitplan_test.go @@ -61,6 +61,8 @@ func TestWaitPlanPingBudgetPerWaitCause(t *testing.T) { wantEntry bool }{ {"main, own tool", sess1, []resp{toolCall, toolCall}, 0, 0, 8, true}, + {"tool stop, no names (spliced window)", sess1, []resp{toolCall, {nil, "tool_use"}}, 0, 0, 8, true}, + {"tool stop, no names, open background", sess1, []resp{toolCall, {nil, "tool_use"}}, 1, 0, 8, true}, {"subagent, own tool", sub, []resp{toolCall, toolCall}, 0, 0, 8, true}, {"main, end_turn, background", sess1, []resp{toolCall, endTurn}, 1, 0, 5, true}, {"main, end_turn, a person", sess1, []resp{toolCall, endTurn}, 0, 0, 3, true}, @@ -325,3 +327,21 @@ func TestWaitPlanShowsInThePlanTheDashboardReads(t *testing.T) { t.Error("a thread with no tool call yet shows a plan") } } + +// A response whose name list may be incomplete (a spliced sniff window) counts as a tool call: the thread is +// pinged as before #858, not refused as "no tool call yet". +func TestWaitPlanIncompleteNamesCountAsAToolCall(t *testing.T) { + k, fd, clock := testKeeper(t) + pol := planPolicy() + for i, ts := range []time.Time{clock.now().Add(-time.Second), clock.now()} { + k.Record(Record{Key: sess1, StartedAt: ts, Body: []byte(kaBody), Provider: bschemas.Anthropic, + Route: "/v1/messages", Status: http.StatusOK, Usage: Usage{CacheRead: 48576, CacheWrite: 1200, StopReason: "end_turn"}, + ToolUseIncomplete: i == 1, UsageOK: true, Policy: pol, Header: kaHeader(), Hold: fakeOf(k).hold(sess1)}) + } + if !k.ToolSeen(sess1) { + t.Fatal("an incomplete name list did not count as a tool call") + } + if n := k.Sweep(clock.advance(281 * time.Second)); n != 1 || fd.n() != 1 { + t.Fatalf("pings = %d/%d, want 1", n, fd.n()) + } +} diff --git a/proxy/keepalive.go b/proxy/keepalive.go index ffd4ecf1c..96e6ad217 100644 --- a/proxy/keepalive.go +++ b/proxy/keepalive.go @@ -410,14 +410,23 @@ func (k *keeper) record(tn *Tenancy, session, threadID string, startedAt time.Ti // record-time gate, evaluated before the request goes out and without a campaign decision (that one // needs the response), so a thread a learned campaign would turn on later gets a plain ping. That is // the safe direction: nothing extra is kept for it. -func (k *keeper) canPing(tn *Tenancy, session string, provider bschemas.ModelProvider) bool { +func (k *keeper) canPing(tn *Tenancy, session, threadID string, provider bschemas.ModelProvider) bool { if k == nil || tn == nil || k.h == nil || k.h.rec == nil || keepAliveDisabled() || provider == bschemas.OpenAI || session == "" { return false } - // Windows are ignored: the wait cause is not known yet, and a non-person wait is pinged at any hour (#858). - pol, _ := k.resolvePolicyWindows(tn, session, provider, tn.Cache, predDecision{}, false, tn.Cache.WaitPlan) - return pol.On() + // The wait cause is not known yet, so the windows count, as before #858: for an unknown wait the + // pessimistic answer is the safe one. A tool wait outside a window then registers no conversation copy + // and still gets its plain ping; it loses only the sweep substitute. A copy is content kept in memory. + pol, _ := k.resolvePolicy(tn, session, provider, tn.Cache, predDecision{}, false) + if !pol.On() { + return false + } + // A thread with no tool call yet is not pinged (#858), so it keeps no copy either. + if pol.WaitPlan && !k.Keeper.ToolSeen(keepalive.Key{Tenant: tn.ID, Session: session, Thread: threadID}) { + return false + } + return true } // resolvePolicy layers the manager strategy and the session override over the account's own @@ -437,7 +446,13 @@ func (k *keeper) resolvePolicyWindows(tn *Tenancy, session string, provider bsch // override, which still wins on top and still cannot widen MaxUSDPerPing. applied := "" if provider != bschemas.OpenAI { - pol, applied = k.applyStrategyWindows(tn.ID, pol, k.now(), ignoreWindows) + account := pol + pol, applied = k.applyStrategyWindows(tn.ID, account, k.now(), false) + // A strategy that matches only because its window was ignored (#858) never displaces an active + // campaign: with a campaign and a closed window, the campaign wins, as before #858. + if applied == "" && ignoreWindows && !(pdecOK && pdec.active) { + pol, applied = k.applyStrategyWindows(tn.ID, account, k.now(), true) + } // An ACTIVE campaign decides the ping count unless a manager strategy matched (the more // explicit instruction). The campaign owns the count, so the per-wait plan is off for it (#858). if pdecOK && applied == "" && pdec.active { diff --git a/proxy/keepalivehost_test.go b/proxy/keepalivehost_test.go index 22e5f0f82..5a42e1194 100644 --- a/proxy/keepalivehost_test.go +++ b/proxy/keepalivehost_test.go @@ -653,23 +653,23 @@ func TestCanPingFollowsTheKeepAlivePolicy(t *testing.T) { offPol.KeepAlive = false off := &Tenancy{ID: "t1", Cache: offPol} k, _, _ := testKeeper(t, Limits{}) - if !k.canPing(on, "sess-1", bschemas.Anthropic) { + if !k.canPing(on, "sess-1", "", bschemas.Anthropic) { t.Error("keep-alive on: canPing = false") } - if k.canPing(off, "sess-1", bschemas.Anthropic) { + if k.canPing(off, "sess-1", "", bschemas.Anthropic) { t.Error("keep-alive off: canPing = true, so a conversation copy would be kept for nothing") } - if k.canPing(on, "", bschemas.Anthropic) { + if k.canPing(on, "", "", bschemas.Anthropic) { t.Error("no session: canPing = true") } - if k.canPing(on, "sess-1", bschemas.OpenAI) { + if k.canPing(on, "sess-1", "", bschemas.OpenAI) { t.Error("OpenAI: the sweep's question is Anthropic only") } - if k.canPing(nil, "sess-1", bschemas.Anthropic) { + if k.canPing(nil, "sess-1", "", bschemas.Anthropic) { t.Error("nil tenancy: canPing = true") } k.h.rec = nil - if k.canPing(on, "sess-1", bschemas.Anthropic) { + if k.canPing(on, "sess-1", "", bschemas.Anthropic) { t.Error("no audit sink: canPing = true, but the keeper holds nothing without one") } } diff --git a/proxy/keepalivestrategy.go b/proxy/keepalivestrategy.go index 3686147f2..85cf4967c 100644 --- a/proxy/keepalivestrategy.go +++ b/proxy/keepalivestrategy.go @@ -72,7 +72,10 @@ func (k *keeper) clockNow() time.Time { // bestStrategyFor resolves the highest-priority ACTIVE strategy matching tenantID at // `now`, if any — the one matching walk both applyStrategy (the ping-scheduling half, // resolved at record time) and resolveHeadTTL (the write-tier half, resolved on the -// request path) need, so the two can never disagree about which strategy won. +// request path) need. They resolve at the same instant, so they agree on which strategy won, with one +// exception since #858: for a wait that is not for a person, applyStrategyWindows ignores the windows +// (bestStrategy), while resolveHeadTTL still respects them. The split is wanted: the head-TTL write tier +// is a cost decision taken at write time, before the response, and a tool wait does not change it. func (k *keeper) bestStrategyFor(tenantID string, now time.Time) *tenant.Strategy { return k.bestStrategy(tenantID, now, false) } diff --git a/proxy/keepalivewaitplan_test.go b/proxy/keepalivewaitplan_test.go index 9f619a416..48e4f37a4 100644 --- a/proxy/keepalivewaitplan_test.go +++ b/proxy/keepalivewaitplan_test.go @@ -119,3 +119,59 @@ func TestBestStrategyIgnoringWindowsKeepsPriorityAndTarget(t *testing.T) { t.Errorf("no strategies: got %+v", b) } } + +// #858 review: canPing keeps the windows, so a thread that will not be pinged keeps no sweep copy. A +// person wait outside the window, and a thread with no tool call yet, register nothing. +func TestCanPingKeepsWindowsAndRefusesAThreadWithNoToolCall(t *testing.T) { + k, _, clock := testKeeper(t, Limits{}) + now := clock.now() + k.setStrategies([]tenant.Strategy{{ + ID: "s", Active: true, IdleSeconds: 240, MaxPings: 2, MinPrefixTokens: 20000, MaxUSDPerPing: 0.5, + Windows: []tenant.Window{{Start: "00:00", End: "00:01"}}, + Target: tenant.Target{Mode: tenant.TargetAll}, UpdatedAt: now, + }}) + acct := &Tenancy{ID: "t1", Cache: CachePolicy{WaitPlan: true}} // keep-alive off on the account + if k.canPing(acct, "sess-1", "", bschemas.Anthropic) { + t.Error("window closed, account off: canPing = true, so a copy would be kept for a thread that never pings") + } + // Open window, but no tool call yet on the thread. + k.setStrategies([]tenant.Strategy{{ + ID: "s", Active: true, IdleSeconds: 240, MaxPings: 2, MinPrefixTokens: 20000, MaxUSDPerPing: 0.5, + Windows: []tenant.Window{{Start: "00:00", End: "24:00"}}, + Target: tenant.Target{Mode: tenant.TargetAll}, UpdatedAt: now, + }}) + if k.canPing(acct, "sess-1", "", bschemas.Anthropic) { + t.Error("no tool call yet: canPing = true") + } + recordResp(t, k, CachePolicy{WaitPlan: true}, "", now, Usage{StopReason: "tool_use", ToolUseNames: []string{"Bash"}}) + if !k.canPing(acct, "sess-1", "", bschemas.Anthropic) { + t.Error("a tool call was seen and the window is open: canPing = false") + } + // The plan off: the guard is off too, as before #858. + k.Keeper.Forget("t1") + if !k.canPing(&Tenancy{ID: "t1", Cache: CachePolicy{}}, "sess-1", "", bschemas.Anthropic) { + t.Error("plan off, strategy matching: canPing = false") + } +} + +// #858 review: a strategy whose window is closed never displaces an active campaign. +func TestClosedWindowStrategyDoesNotDisplaceACampaign(t *testing.T) { + k, _, clock := testKeeper(t, Limits{}) + now := clock.now() + k.setStrategies([]tenant.Strategy{{ + ID: "s", Active: true, IdleSeconds: 240, MaxPings: 4, MinPrefixTokens: 20000, MaxUSDPerPing: 0.5, + Windows: []tenant.Window{{Start: "00:00", End: "00:01"}}, + Target: tenant.Target{Mode: tenant.TargetAll}, UpdatedAt: now, + }}) + tn := &Tenancy{ID: "t1", Cache: CachePolicy{WaitPlan: true, Idle: 240 * time.Second, MaxPings: 2}} + pdec := predDecision{campaign: &predCampaign{}, k: 3, active: true} + _, applied := k.resolvePolicyWindows(tn, "sess-1", bschemas.Anthropic, tn.Cache, pdec, true, true) + if applied == "s" || applied == "" { + t.Fatalf("applied = %q, want the campaign", applied) + } + // With no campaign the strategy applies at any hour, as designed. + _, applied = k.resolvePolicyWindows(tn, "sess-1", bschemas.Anthropic, tn.Cache, predDecision{}, false, true) + if applied != "s" { + t.Fatalf("no campaign: applied = %q, want s", applied) + } +} diff --git a/proxy/modes.go b/proxy/modes.go index 54ace3a4c..b3e7a6124 100644 --- a/proxy/modes.go +++ b/proxy/modes.go @@ -69,7 +69,7 @@ func (h *Handler) applyMode(r *reqInfo) ([]byte, time.Duration, apply.Trace) { // Whether the keep-alive can ping this thread, so a component that prepares something for a // ping keeps nothing for a thread that will never get one. See keeper.canPing. CanKeepAlive: func(session, thread string) bool { - return h.keeper.canPing(r.tn, session, r.provider) + return h.keeper.canPing(r.tn, session, thread, r.provider) }, }) added := time.Since(start) From 7c0d4843ed0d66da0d0a7c1e33fe6f75afcd5de5 Mon Sep 17 00:00:00 2001 From: DAVID AMID Date: Mon, 2 Nov 2026 22:58:39 +0200 Subject: [PATCH 5/5] test(proxy): make the canPing window test independent of the tool guard Assisted-By: Claude Opus 5.5 Signed-off-by: DAVID AMID --- proxy/keepalivewaitplan_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/proxy/keepalivewaitplan_test.go b/proxy/keepalivewaitplan_test.go index 48e4f37a4..7c8c8bd22 100644 --- a/proxy/keepalivewaitplan_test.go +++ b/proxy/keepalivewaitplan_test.go @@ -131,9 +131,12 @@ func TestCanPingKeepsWindowsAndRefusesAThreadWithNoToolCall(t *testing.T) { Target: tenant.Target{Mode: tenant.TargetAll}, UpdatedAt: now, }}) acct := &Tenancy{ID: "t1", Cache: CachePolicy{WaitPlan: true}} // keep-alive off on the account + // A tool call was seen, so only the window decides here. + recordResp(t, k, CachePolicy{WaitPlan: true}, "", now, Usage{StopReason: "tool_use", ToolUseNames: []string{"Bash"}}) if k.canPing(acct, "sess-1", "", bschemas.Anthropic) { t.Error("window closed, account off: canPing = true, so a copy would be kept for a thread that never pings") } + k.Keeper.Forget("t1") // Open window, but no tool call yet on the thread. k.setStrategies([]tenant.Strategy{{ ID: "s", Active: true, IdleSeconds: 240, MaxPings: 2, MinPrefixTokens: 20000, MaxUSDPerPing: 0.5,