diff --git a/crates/buzz-agent/README.md b/crates/buzz-agent/README.md index 0bc03db7813..b441d5ea554 100644 --- a/crates/buzz-agent/README.md +++ b/crates/buzz-agent/README.md @@ -153,10 +153,10 @@ Everything is environment variables. No flags, no config files. (We are a subpro | `BUZZ_AGENT_SYSTEM_PROMPT` | built-in | Inline system prompt. | | `BUZZ_AGENT_SYSTEM_PROMPT_FILE` | — | File path. Mutually exclusive with the above. | | `BUZZ_AGENT_MAX_ROUNDS` | `0` | Tool-loop iteration cap. 0 = unlimited. | -| `BUZZ_AGENT_MAX_OUTPUT_TOKENS` | `65536` | Desired per-call ceiling. Set this at or below the served model's output limit for each agent deployment. Proactive handoff is independently based on 90% of `BUZZ_AGENT_MAX_CONTEXT_TOKENS`. | +| `BUZZ_AGENT_MAX_OUTPUT_TOKENS` | `65536` | Desired per-call ceiling. Set this at or below the served model's output limit for each agent deployment. Proactive handoff runs at the end of a turn whose final request used 90% or more of `BUZZ_AGENT_MAX_CONTEXT_TOKENS`. | | `BUZZ_AGENT_MAX_TOKEN_RECOVERIES` | `3` | Retries after a successful response is truncated at the output-token limit. `0` disables recovery; the finite value and `BUZZ_AGENT_MAX_ROUNDS` prevent infinite retries. | | `BUZZ_AGENT_MAX_CONTEXT_TOKENS` | `200000` | Provider context window used by the handoff gate. | -| `BUZZ_AGENT_MAX_HANDOFFS` | `10` | Max context handoffs per session before falling back to truncation. | +| `BUZZ_AGENT_MAX_HANDOFFS` | `10` | `0` disables proactive (end-of-turn) context handoff; any positive value enables it. Reactive recovery from a provider context-window error is unaffected. | | `BUZZ_AGENT_LLM_TIMEOUT_SECS` | `240` | Max seconds with no response bytes before abandoning an LLM call (per-read inactivity, not wall-clock). | | `BUZZ_AGENT_TOOL_TIMEOUT_SECS` | `660` | Per-tool call timeout in seconds | | `BUZZ_AGENT_MAX_PARALLEL_TOOLS` | `8` | Max concurrent tool calls per turn (1 = sequential) | diff --git a/crates/buzz-agent/src/agent.rs b/crates/buzz-agent/src/agent.rs index 9258ce449f3..6d3937fe811 100644 --- a/crates/buzz-agent/src/agent.rs +++ b/crates/buzz-agent/src/agent.rs @@ -9,7 +9,7 @@ use crate::builtin; use crate::config::{ pricing_authority, Config, MAX_PROMPT_BYTES, MAX_TOOL_CALLS_PER_TURN, MAX_TOOL_RESULT_BYTES, }; -use crate::handoff::{ContextRecovery, HandoffOutcome}; +use crate::handoff::ContextRecovery; use crate::hints::SkillEntry; use crate::llm::Llm; use crate::mcp::McpRegistry; @@ -166,12 +166,6 @@ pub struct RunCtx<'a> { /// context. The handoff gate reads this to compare against the token /// budget; falls back to the byte heuristic when `None`. pub last_request_input_tokens: &'a mut Option, - /// History byte size at the moment `last_request_input_tokens` was - /// measured. Paired with it so the gate can add a conservative token - /// estimate of history that has grown since (tool results, next prompt), - /// which the exact-but-stale token count would otherwise miss. Cleared and - /// preserved in lockstep with `last_request_input_tokens`. - pub last_request_history_bytes: &'a mut Option, /// Accumulated input tokens across all LLM rounds in this turn, for /// NIP-AM metric publishing. Reset to `Unseen` at turn start in `run()`. pub turn_input_tokens: &'a mut TurnIOState, @@ -319,14 +313,6 @@ impl RunCtx<'_> { *self.turn_cache_write_tokens = CacheTotalState::Unseen; *self.turn_pricing_identity = None; *self.turn_total_state = TurnTotalState::Unseen; - // Per-turn handoff-attempt counter. Scoped here (not persisted in the - // session) so `BUZZ_AGENT_MAX_HANDOFFS` bounds compactions per - // `session/prompt` turn rather than per session lifetime. A - // long-lived session legitimately needs unbounded handoffs across - // prompts; the cap only exists to stop runaway within a single turn. - // The session-cumulative `handoff_count` (used in log lines) is not - // reset: it reflects total compactions since session start. - let mut handoff_attempts: usize = 0; let mut round = 0u32; // Per-prompt `_Stop` objection count. Bounded per prompt (not per @@ -361,21 +347,12 @@ impl RunCtx<'_> { // its next request — the turn continues, it is not restarted. Drain // non-blocking; an empty queue is the common case. self.drain_steers(); - match self.maybe_handoff(&mut handoff_attempts).await { - HandoffOutcome::Cancelled => return Ok(StopReason::Cancelled), - // Context was just reset — the prior request's token count no - // longer describes the (now much smaller) history. Clear both - // the token count and its byte baseline so a stale over- - // threshold reading can't immediately re-fire the handoff - // before the next response reports fresh usage. - HandoffOutcome::Performed => { - *self.last_request_input_tokens = None; - *self.last_request_history_bytes = None; - } - HandoffOutcome::Skipped => { - truncate_history(self.history, self.cfg.max_history_bytes) - } - } + // No proactive compaction mid-turn: the working set is what the + // model is using right now, and summarizing it away forces a + // re-read storm. Proactive compaction runs once at the end of the + // turn (`end_of_turn_handoff`, called by `run_prompt`); mid-turn + // overflow is caught reactively by the context-400 arm below. + truncate_history(self.history, self.cfg.max_history_bytes); let mut tools = self.mcp.tools(); // Inject the built-in load_skill tool when skills are available. @@ -461,13 +438,10 @@ impl RunCtx<'_> { // many rounds can ever be refunded in one turn. An // ordinary round is never refunded. round = round.saturating_sub(1); - // Same reset as the proactive path (see - // `HandoffOutcome::Performed` above): the frozen - // token reading describes history that no longer - // exists. Clearing it is what lets the gate work - // again on later rounds. + // The frozen token reading describes history that + // no longer exists; clearing it is what lets the + // end-of-turn gate read fresh usage. *self.last_request_input_tokens = None; - *self.last_request_history_bytes = None; continue; } ContextRecovery::Cancelled => return Ok(StopReason::Cancelled), @@ -482,19 +456,11 @@ impl RunCtx<'_> { } Err(error) => return Err(error), }; - // Record provider-reported input usage so the next loop iteration's - // handoff gate can compare it against the token budget. We capture - // it together with the history byte size AT THIS MOMENT — which is - // exactly the history that was just sent to `complete()` (the - // assistant response is appended below, after this point). Pairing - // them lets the gate add a conservative estimate for any history - // appended before the next request. Uses `context_pressure_bytes` - // (the same measure the gate's `current_bytes` uses) so the - // `grown` delta is coherent — an image contributes its visual- - // token equivalent here, not its base64 length. Preserve both when - // a response omits usage (`None`) rather than clobbering — a - // one-off missing field shouldn't blind the gate or zero the - // growth baseline. + // Record provider-reported input usage so the end-of-turn handoff + // gate can compare the turn's FINAL request against the token + // budget. Preserve the prior reading when a response omits usage + // (`None`) rather than clobbering — a one-off missing field + // shouldn't blind the gate. if response.input_tokens_overflowed { // The Anthropic-style inclusive sum (input_tokens + // cache_read_input_tokens + cache_creation_input_tokens) @@ -507,12 +473,6 @@ impl RunCtx<'_> { *self.turn_input_tokens = TurnIOState::Poisoned; } else if let Some(tokens) = response.input_tokens { *self.last_request_input_tokens = Some(tokens); - *self.last_request_history_bytes = Some( - self.history - .iter() - .map(HistoryItem::context_pressure_bytes) - .sum(), - ); // Accumulate per-turn input tokens for NIP-AM metric publishing. // fold_round uses checked_add; overflow permanently poisons the // turn accumulator (and, via merge_session, the session cumulative). diff --git a/crates/buzz-agent/src/config.rs b/crates/buzz-agent/src/config.rs index 67d7c593b56..24d9fcb9fa4 100644 --- a/crates/buzz-agent/src/config.rs +++ b/crates/buzz-agent/src/config.rs @@ -472,11 +472,12 @@ pub struct Config { /// operators lower/raise it for other models. Set via /// `BUZZ_AGENT_MAX_CONTEXT_TOKENS`. pub max_context_tokens: u64, - /// Maximum context-handoff attempts permitted within a single - /// `session/prompt` turn. Caps runaway compaction loops inside one turn; - /// does NOT limit handoffs across a session's lifetime — a long-lived - /// session can compact on every successive turn without hitting this bound. - /// Set via `BUZZ_AGENT_MAX_HANDOFFS`. Default 10. + /// Proactive context handoff switch: `0` disables the end-of-turn + /// compaction gate entirely (history then grows until the byte cap or a + /// provider context-400, which reactive recovery handles regardless of + /// this setting). Any positive value enables it; the gate runs at most + /// once per turn by construction, so the magnitude no longer bounds + /// anything. Set via `BUZZ_AGENT_MAX_HANDOFFS`. Default 10. pub max_handoffs: usize, pub max_parallel_tools: usize, pub hook_timeout: Duration, diff --git a/crates/buzz-agent/src/handoff.rs b/crates/buzz-agent/src/handoff.rs index 869fbe06664..2afb230a419 100644 --- a/crates/buzz-agent/src/handoff.rs +++ b/crates/buzz-agent/src/handoff.rs @@ -52,43 +52,51 @@ static HANDOFF_SYSTEM_PROMPT: std::sync::LazyLock = std::sync::LazyLock: ) }); +/// Whether [`RunCtx::handoff`] re-appends the live user prompt after the reset. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Reseat { + /// Mid-turn: the prompt is still being answered, so the fresh context must + /// end with it or the model has nothing to respond to. + LivePrompt, + /// End of turn: the prompt was already answered. Re-seating it would hand + /// the next turn a dangling user message that reads as a repeated request. + None, +} + impl RunCtx<'_> { - pub(crate) async fn maybe_handoff(&mut self, handoff_attempts: &mut usize) -> HandoffOutcome { - if !self.should_handoff() { + /// Proactive compaction, run once after a turn completes. Gates on the + /// provider's measured input usage for the turn's final request — and only + /// that (see [`Self::should_handoff`]) — and compacts so the NEXT turn + /// starts on a fresh context. + /// + /// Deliberately not run mid-turn: a turn's working set (file reads, tool + /// results) is exactly what the model is still using, and compacting it + /// forces a re-read storm that refills the window. Mid-turn overflow is + /// handled reactively by [`Self::recover_from_context_overflow`]. A turn + /// boundary is the cheapest place to lose the working set. + pub(crate) async fn end_of_turn_handoff(&mut self) -> HandoffOutcome { + if self.cfg.max_handoffs == 0 || !self.should_handoff() { return HandoffOutcome::Skipped; } - if *handoff_attempts >= self.cfg.max_handoffs { - let projected = self.projected_handoff_input_tokens(); - let threshold = token_threshold(self.cfg.max_context_tokens); - tracing::warn!( - session_id = self.session_id, - reason = "preflight", - handoff_attempts = *handoff_attempts, - max_handoffs = self.cfg.max_handoffs, - projected_tokens = projected, - threshold_tokens = threshold, - "handoff cap reached; using truncation", - ); - return HandoffOutcome::Skipped; + let outcome = self.handoff(None, Reseat::None).await; + if matches!(outcome, HandoffOutcome::Performed) { + // The measured usage describes history that no longer exists; a + // stale over-threshold reading must not re-fire the gate. + *self.last_request_input_tokens = None; } - // Consume one attempt slot before calling handoff(). This ensures - // that empty-summary, summarize-error, and cancellation outcomes all - // burn budget — not just successful compactions — so the cap cannot - // be bypassed by a flaky summarizer. - *handoff_attempts += 1; - self.handoff(None).await + outcome } - /// Handoff forced by a provider context-window rejection, bypassing both - /// gates in [`Self::maybe_handoff`]. + /// Handoff forced by a provider context-window rejection, bypassing the + /// gate in [`Self::end_of_turn_handoff`]. /// - /// The gates exist to *predict* overflow; a 400 naming a context-length - /// overflow is overflow already observed, so neither prediction applies. - /// `should_handoff()` reads a token count frozen at the last SUCCESSFUL - /// request (a failed request reports no usage), so it is under threshold by - /// construction — that frozen reading is the permanent stick. And - /// `max_handoffs` is a cost cap whose only alternative here is a request - /// that cannot succeed. + /// The gate exists to *predict* overflow; a 400 naming a context-length + /// overflow is overflow already observed, so the prediction does not + /// apply. `should_handoff()` reads a token count frozen at the last + /// SUCCESSFUL request (a failed request reports no usage), so it is under + /// threshold by construction — that frozen reading is the permanent stick. + /// And `max_handoffs` is a cost cap whose only alternative here is a + /// request that cannot succeed. /// /// `history_budget_bytes` is explicit rather than derived from /// `cfg.max_context_tokens`: that window is the quantity the provider just @@ -97,7 +105,8 @@ impl RunCtx<'_> { tracing::warn!( "provider reported context overflow; forcing handoff (history budget {history_budget_bytes} bytes)" ); - self.handoff(Some(history_budget_bytes)).await + self.handoff(Some(history_budget_bytes), Reseat::LivePrompt) + .await } /// The reactive context-recovery ladder, run after the provider rejected a @@ -171,11 +180,20 @@ impl RunCtx<'_> { } } - /// The handoff mechanism itself: summarize, reset, re-seat the live prompt. - /// Holds no gate — callers decide whether a handoff is warranted. - async fn handoff(&mut self, history_budget_bytes: Option) -> HandoffOutcome { + /// The handoff mechanism itself: summarize, reset, and (per `reseat`) + /// re-seat the live prompt. Holds no gate — callers decide whether a + /// handoff is warranted. + async fn handoff( + &mut self, + history_budget_bytes: Option, + reseat: Reseat, + ) -> HandoffOutcome { let prompt = self.build_handoff_prompt(history_budget_bytes); - let tokens_before = self.projected_handoff_input_tokens(); + // For the log line: the provider's measurement when we have one, else + // the byte-derived upper bound the byte fallback gates on. + let tokens_before = self + .last_request_input_tokens + .unwrap_or_else(|| estimate_history_tokens(self.history)); let summary = tokio::select! { biased; _ = self.cancel.changed() => return HandoffOutcome::Cancelled, @@ -197,10 +215,13 @@ impl RunCtx<'_> { } }, }; - let current_prompt = self.history.iter().rev().find_map(|item| match item { - HistoryItem::User(s) => Some(s.clone()), - _ => None, - }); + let current_prompt = match reseat { + Reseat::LivePrompt => self.history.iter().rev().find_map(|item| match item { + HistoryItem::User(s) => Some(s.clone()), + _ => None, + }), + Reseat::None => None, + }; let prior = self.history.len(); // Reset history first; the _PostCompact hook is meant to inject // state into the FRESH context, not the old one we're discarding. @@ -241,11 +262,17 @@ impl RunCtx<'_> { HandoffOutcome::Performed } + /// Measured-usage gate: the provider's input-token count for the turn's + /// final request against 90% of the window. No growth estimate is added — + /// at end of turn the only history appended since that measurement is the + /// final assistant reply, and charging it at a byte-per-token would compact + /// an under-threshold turn just because its answer was long. Before any + /// usage is known (first request, or right after a reset) fall back to a + /// conservative byte cap so a single pre-usage turn can't blow the window. fn should_handoff(&self) -> bool { match *self.last_request_input_tokens { - Some(_) => { - self.projected_handoff_input_tokens() - >= token_threshold(self.cfg.max_context_tokens) + Some(measured_tokens) => { + measured_tokens >= token_threshold(self.cfg.max_context_tokens) } None => { let bytes: usize = self @@ -262,42 +289,6 @@ impl RunCtx<'_> { } } - fn projected_handoff_input_tokens(&self) -> u64 { - let current_tokens = estimate_history_tokens(self.history); - match *self.last_request_input_tokens { - // Token-first: the provider told us exactly how many input tokens - // the PREVIOUS request used. But history has grown since that - // measurement — new assistant text, tool results, and the next - // user prompt are appended before the next `complete()`. The exact - // count alone would miss "previous request was under threshold, but - // newly appended content pushes the next one over" (the stale-usage - // cousin of the original stale-bytes bug). So we add a conservative - // token estimate of the bytes added since the measurement. - Some(measured_tokens) => { - let measured_bytes = self.last_request_history_bytes.unwrap_or(0); - let current_bytes: usize = self - .history - .iter() - .map(HistoryItem::context_pressure_bytes) - .sum(); - let grown = current_bytes.saturating_sub(measured_bytes); - measured_tokens.saturating_add(estimate_tokens_from_bytes(grown)) - } - // No usage yet (first request, or just after a handoff reset). - // Fall back to the byte heuristic, capped conservatively so a - // single pre-usage request can't blow the window. We map the token - // threshold to bytes using a deliberately LOW bytes/token ratio: - // a low ratio implies more tokens per byte, so the byte cap is - // small and the handoff fires early rather than late. Never raise - // the cap above the configured byte budget. - // - // Caveat: this can't shrink a single oversized current prompt, - // since a handoff re-adds the current prompt verbatim — that is a - // prompt-cap concern (MAX_PROMPT_BYTES), not this gate. - None => current_tokens, - } - } - /// Build the summarizer prompt. `history_budget_bytes` overrides the /// budget normally derived from `cfg.max_context_tokens`; `None` keeps the /// derived value, which is what the proactive path uses. @@ -473,8 +464,8 @@ pub(crate) fn clamp_bytes(s: &str, max_bytes: usize) -> String { /// every byte as a whole token is an unconditional UPPER bound on the true /// token count — it can never undercount, regardless of content density (even /// the densest real content sits at ~1.4 bytes/token). That over-estimate is -/// exactly what a fail-early preflight gate wants: it hands off sooner rather -/// than risk the next request exceeding the window. +/// exactly what the pre-usage byte fallback and the summarizer prompt budget +/// want: err toward handing off sooner rather than risk exceeding the window. const CONSERVATIVE_BYTES_PER_TOKEN: u64 = 1; fn estimate_history_tokens(history: &[HistoryItem]) -> u64 { diff --git a/crates/buzz-agent/src/lib.rs b/crates/buzz-agent/src/lib.rs index 98fa99ca5bf..68455f7b8c5 100644 --- a/crates/buzz-agent/src/lib.rs +++ b/crates/buzz-agent/src/lib.rs @@ -43,7 +43,7 @@ use crate::config::{Config, MAX_SYSTEM_PROMPT_BYTES, PROTOCOL_VERSION}; use crate::hints::SkillEntry; use crate::llm::Llm; use crate::mcp::McpRegistry; -use crate::types::{ContentBlock, HistoryItem}; +use crate::types::{ContentBlock, HistoryItem, StopReason}; use crate::wire::{ classify, goose_session_update, Inbound, InitializeParams, SessionCancelParams, SessionNewParams, SessionPromptParams, SessionSetModelParams, SessionSteerParams, WireMsg, @@ -84,12 +84,9 @@ struct Session { handoff_count: usize, /// Cache-summed input tokens the provider reported for this session's most /// recent request, or `None` before the first response (or after a handoff - /// resets the context). Drives the token-based handoff gate; see - /// [`RunCtx::should_handoff`]. + /// resets the context). Drives the end-of-turn handoff gate; see + /// [`RunCtx::end_of_turn_handoff`]. last_request_input_tokens: Option, - /// History byte size when `last_request_input_tokens` was measured, paired - /// with it so the gate can account for history appended since. - last_request_history_bytes: Option, effective_system_prompt: Arc, /// Per-session model override set by `session/set_model`. When `Some`, /// overrides `App::cfg.model` for all LLM calls on this session. Persists @@ -490,7 +487,6 @@ async fn session_new(app: &Arc, id: Value, params: Value, wire_tx: &WireSen original_task: None, handoff_count: 0, last_request_input_tokens: None, - last_request_history_bytes: None, effective_system_prompt, effective_model: None, accumulated_input_tokens: crate::types::TurnIOState::Unseen, @@ -679,7 +675,6 @@ async fn run_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender mut original_task, mut handoff_count, mut last_request_input_tokens, - mut last_request_history_bytes, mut cancel_rx, effective_system_prompt, effective_model_override, @@ -743,7 +738,6 @@ async fn run_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender handoff_count: &mut handoff_count, run_id, last_request_input_tokens: &mut last_request_input_tokens, - last_request_history_bytes: &mut last_request_history_bytes, turn_input_tokens: &mut turn_input_tokens, turn_output_tokens: &mut turn_output_tokens, turn_cached_input_tokens: &mut turn_cached_input_tokens, @@ -753,6 +747,13 @@ async fn run_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender usage_baseline, }; let result = ctx.run(p.prompt).await; + // Proactive compaction lives at the turn boundary, not mid-turn. Gate on + // the stop reason, not `is_ok()`: `run()` reports cancellation as + // `Ok(StopReason::Cancelled)`, and a cancelled turn must not spend a + // summarize round trip the user just asked us to stop. + if matches!(result, Ok(reason) if reason != StopReason::Cancelled) { + ctx.end_of_turn_handoff().await; + } if let Some(s) = app.sessions.lock().await.get_mut(&sid) { s.busy = false; // Clear run state so a late steer can't queue into a finished turn. @@ -762,7 +763,6 @@ async fn run_prompt(app: Arc, id: Value, params: Value, wire_tx: WireSender s.original_task = original_task; s.handoff_count = handoff_count; s.last_request_input_tokens = last_request_input_tokens; - s.last_request_history_bytes = last_request_history_bytes; } // Update session-cumulative token counters and emit the usage notification // BEFORE sending the session/prompt response. buzz-acp's UsageTracker @@ -866,7 +866,6 @@ async fn acquire_session( Option, usize, Option, - Option, watch::Receiver, Arc, Option, @@ -909,7 +908,6 @@ async fn acquire_session( s.original_task.take(), s.handoff_count, s.last_request_input_tokens, - s.last_request_history_bytes, rx, Arc::clone(&s.effective_system_prompt), effective_model, diff --git a/crates/buzz-agent/tests/regressions.rs b/crates/buzz-agent/tests/regressions.rs index 6a4f347f6bb..4fbdfc6e725 100644 --- a/crates/buzz-agent/tests/regressions.rs +++ b/crates/buzz-agent/tests/regressions.rs @@ -20,13 +20,18 @@ struct CapturingLlm { captured: Arc>>, } +/// Sentinel body for [`spawn_capturing_llm_with_status`]: never respond. +const HANG: &str = "__hang__"; + async fn spawn_capturing_llm(responses: Vec) -> CapturingLlm { spawn_capturing_llm_with_status(responses.into_iter().map(|v| (200u16, v)).collect()).await } /// Like `spawn_capturing_llm` but each canned response carries its own HTTP /// status, so a test can serve a real provider rejection (e.g. a context-window -/// 400) instead of only success bodies. +/// 400) instead of only success bodies. A response whose body is [`HANG`] is +/// captured but never answered: the socket is held open until the agent drops +/// the request, so a test can cancel a turn while a provider call is in flight. async fn spawn_capturing_llm_with_status(responses: Vec<(u16, Value)>) -> CapturingLlm { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("http://{}", listener.local_addr().unwrap()); @@ -78,6 +83,11 @@ async fn spawn_capturing_llm_with_status(responses: Vec<(u16, Value)>) -> Captur .await .pop_front() .unwrap_or_else(|| (200, json!({ "error": "no canned response" }))); + if body == json!(HANG) { + // Hold the connection until the peer goes away. + let _ = sock.read(&mut tmp).await; + return; + } let body_s = serde_json::to_string(&body).unwrap(); let reason = match status { 200 => "OK", @@ -1142,23 +1152,23 @@ async fn hook_tools_hidden_from_llm() { /// *new* context, not the discarded one. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn hook_post_compact_injects_after_handoff() { - // Sequence of canned LLM responses consumed in order: - // 1-3. Three `session/prompt` rounds returning short text. Each - // prompt body is ~300 KB, so by the 4th prompt we'll be over - // the 90% (= ~922 KB) threshold of a 1 MB budget. - // 4. Handoff `summarize()` call returns the summary text. - // 5. Next regular `complete()` call after the handoff returns - // a final "done" message; we inspect this request's body. + // Canned LLM responses, consumed in order. No response carries usage, so + // the end-of-turn gate runs on the byte fallback; each ~300 KB prompt is + // over that threshold on its own, so the first turn ends with a + // `summarize()` and the next `complete()` carries the fresh context. The + // loop below tolerates the handoff landing later; spares keep the queue + // from being what ends a turn. let llm = spawn_capturing_llm(vec![ openai_text("ack-1"), - openai_text("ack-2"), - openai_text("ack-3"), openai_text("handoff summary text"), openai_text("done"), + openai_text("spare-1"), + openai_text("spare-2"), + openai_text("spare-3"), + openai_text("spare-4"), ]) .await; - // 1 MB budget = MIN allowed. Threshold = ~922 KB. Each ~300 KB prompt - // fills the budget on the 4th turn, triggering handoff. + // 1 MB budget = MIN allowed. let mut h = Harness::spawn_with_env( &llm.url, &[ @@ -1211,6 +1221,16 @@ async fn hook_post_compact_injects_after_handoff() { "no handoff observed after {prompts_sent} prompts (captured={})", llm.captured.lock().await.len() ); + // The handoff ran at the END of the last turn, so the fresh context has + // not been sent to the LLM yet. One more prompt produces the request that + // must carry it. + let p = h + .send( + "session/prompt", + json!({"sessionId": sid, "prompt": [{"type":"text","text":"after handoff"}]}), + ) + .await; + let _ = h.recv_until(|v| v["id"] == json!(p)).await; // The first LLM call AFTER the handoff is the one we inspect. Find it: // it's the one where the messages array is short (history just reset) @@ -1254,10 +1274,12 @@ async fn hook_post_compact_injects_after_handoff() { /// to the old fixed tail of five tiny snippets. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn handoff_summary_prompt_includes_full_history_within_context_budget() { + // Turn 1 stays under threshold; turn 2 reports usage over it, so the + // summarize() lands at the end of turn 2 with both turns in history. let llm = spawn_capturing_llm(vec![ - openai_text_with_usage("ack-0", 9500), + openai_text_with_usage("ack-0", 10), + openai_text_with_usage("ack-1", 9500), openai_text("handoff summary text"), - openai_text_with_usage("done", 10), ]) .await; let mut h = Harness::spawn_with_env( @@ -1292,8 +1314,8 @@ async fn handoff_summary_prompt_includes_full_history_within_context_budget() { let _ = h.recv_until(|v| v["id"] == json!(p1)).await; let captured = llm.captured.lock().await; - assert_eq!(captured.len(), 3, "expected prompt, handoff, prompt"); - let handoff_messages = captured[1]["messages"].as_array().unwrap(); + assert_eq!(captured.len(), 3, "expected prompt, prompt, handoff"); + let handoff_messages = captured[2]["messages"].as_array().unwrap(); let handoff_prompt = handoff_messages[1]["content"].as_str().unwrap(); assert!( handoff_prompt.contains("# Session History (oldest first)"), @@ -1322,10 +1344,14 @@ async fn handoff_summary_prompt_includes_full_history_within_context_budget() { /// form of the most recent item instead of sending an empty history block. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn handoff_summary_prompt_keeps_latest_item_when_one_item_exceeds_budget() { + // At the end of a turn the newest history item is the assistant's final + // reply, so that is the item made oversized here: it alone exceeds the + // summarizer budget and must be kept in clamped form. + let huge = format!("oversize-latest-marker {}", "x".repeat(12000)); let llm = spawn_capturing_llm(vec![ - openai_text_with_usage("ack-0", 9500), + openai_text_with_usage("ack-0", 10), + openai_text_with_usage(&huge, 9500), openai_text("handoff summary text"), - openai_text_with_usage("done", 10), ]) .await; let mut h = Harness::spawn_with_env( @@ -1343,7 +1369,6 @@ async fn handoff_summary_prompt_keeps_latest_item_when_one_item_exceeds_budget() .await; let sid = init_session(&mut h, json!([])).await; - let huge = format!("oversize-latest-marker {}", "x".repeat(12000)); let p0 = h .send( "session/prompt", @@ -1355,14 +1380,14 @@ async fn handoff_summary_prompt_keeps_latest_item_when_one_item_exceeds_budget() let p1 = h .send( "session/prompt", - json!({"sessionId": sid, "prompt": [{"type":"text","text": huge}]}), + json!({"sessionId": sid, "prompt": [{"type":"text","text":"late-history-marker"}]}), ) .await; let _ = h.recv_until(|v| v["id"] == json!(p1)).await; let captured = llm.captured.lock().await; - assert_eq!(captured.len(), 3, "expected prompt, handoff, prompt"); - let handoff_messages = captured[1]["messages"].as_array().unwrap(); + assert_eq!(captured.len(), 3, "expected prompt, prompt, handoff"); + let handoff_messages = captured[2]["messages"].as_array().unwrap(); let handoff_prompt = handoff_messages[1]["content"].as_str().unwrap(); assert!( handoff_prompt.contains("oversize-latest-marker"), @@ -1377,20 +1402,21 @@ async fn handoff_summary_prompt_keeps_latest_item_when_one_item_exceeds_budget() /// Regression for the original bug: context fills, the provider 400s on the /// next request, and the handoff never fires because the old gate measured -/// BYTES (16 MiB threshold) while the limit is in TOKENS. The fix gates on +/// BYTES (16 MiB threshold) while the limit is in TOKENS. The gate reads /// provider-reported input tokens. Here the prompts are tiny (bytes nowhere /// near any byte threshold), but the fake LLM reports `usage.prompt_tokens` /// over the configured token budget — so the handoff MUST fire on the token -/// signal alone, before the next normal `complete()`. +/// signal alone. +/// +/// Proactive compaction runs at the END of the turn whose final request was +/// over threshold, so the next turn starts fresh: the summarize() call lands +/// before prompt #0's `session/prompt` response, not at the start of prompt #1. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn token_usage_over_budget_triggers_handoff() { - // Context window 1000 tokens, output 100 -> threshold = min(900, 900) = 900. - // First response reports 950 input tokens (> 900). The agent stores that; - // the next prompt's pre-flight gate sees 950 >= 900 and hands off, which - // inserts an extra summarize() call we didn't issue. - // req 1: prompt #0 -> text + usage(950) - // req 2: summarize() (the handoff) -> summary text - // req 3: prompt #1's actual complete() -> done + // Context window 1000 tokens -> threshold = 900. + // req 1: prompt #0 complete() -> text + usage(950) + // req 2: summarize() (end-of-turn handoff, still inside prompt #0) + // req 3: prompt #1 complete() -> done let llm = spawn_capturing_llm(vec![ openai_text_with_usage("ack-0", 950), openai_text("handoff summary text"), @@ -1414,7 +1440,8 @@ async fn token_usage_over_budget_triggers_handoff() { .await; let sid = init_session(&mut h, json!([])).await; - // Prompt #0: small body; response carries usage(950) -> over threshold. + // Prompt #0: small body; response carries usage(950) -> over threshold, + // so the turn ends with a summarize() before the response is sent. let p0 = h .send( "session/prompt", @@ -1424,13 +1451,23 @@ async fn token_usage_over_budget_triggers_handoff() { let _ = h.recv_until(|v| v["id"] == json!(p0)).await; assert_eq!( llm.captured.lock().await.len(), - 1, - "first prompt should produce exactly one LLM request (no handoff yet)" + 2, + "over-threshold turn should end with a summarize() (2 reqs) — \ + token gate did not fire on usage over budget" + ); + let stderr = h.stderr_text(); + assert!( + stderr.contains("handoff #1 (history"), + "expected handoff log line in stderr, got: {stderr}" + ); + assert!( + stderr.contains(" -> ") && stderr.contains(" tokens"), + "expected handoff log to include before/after token counts, got: {stderr}" ); - // Prompt #1: also small. The pre-flight gate sees the stored 800 tokens - // and hands off BEFORE issuing this prompt's complete() -> an extra - // summarize request appears (3 total, not 2). + // Prompt #1 runs on the compacted history: exactly one more request, and + // its messages are the handoff block followed by the new prompt — the + // answered prompt #0 must not be re-seated after the summary. let p1 = h .send( "session/prompt", @@ -1438,44 +1475,50 @@ async fn token_usage_over_budget_triggers_handoff() { ) .await; let _ = h.recv_until(|v| v["id"] == json!(p1)).await; - let captured = llm.captured.lock().await.len(); + let captured = llm.captured.lock().await; assert_eq!( - captured, 3, - "expected handoff summarize() between the two prompts (3 reqs), saw {captured} — \ - token gate did not fire on usage over budget" + captured.len(), + 3, + "prompt #1 must not re-trigger a handoff on the cleared usage" + ); + let users: Vec<&str> = captured[2]["messages"] + .as_array() + .unwrap() + .iter() + .filter(|m| m["role"] == "user") + .map(|m| m["content"].as_str().unwrap_or("")) + .collect(); + assert_eq!( + users.len(), + 2, + "expected [handoff, prompt #1], got {users:?}" ); - let stderr = h.stderr_text(); assert!( - stderr.contains("handoff #1 (history"), - "expected handoff log line in stderr, got: {stderr}" + users[0].starts_with("[Context Handoff]"), + "first user item must be the handoff block, got {:?}", + users[0] ); + assert_eq!(users[1], "hello 1"); assert!( - stderr.contains(" -> ") && stderr.contains(" tokens"), - "expected handoff log to include before/after token counts, got: {stderr}" + !users.iter().any(|u| u.contains("hello 0")), + "answered prompt must not be re-seated after end-of-turn compaction: {users:?}" ); h.shutdown().await; } -/// Regression for the stale-usage gap (caught in review): the exact token -/// count describes the PREVIOUS request, but history grows afterward (tool -/// results, next prompt). If the gate trusted only the stale `Some(tokens)` -/// and skipped the byte signal, a previously-under-threshold session could -/// still 400 once a large tool result lands. The fix adds a conservative -/// token estimate of the bytes grown since the measurement. Here usage is -/// reported UNDER threshold, then a large tool result grows history enough -/// that the projection crosses — so the handoff must fire. +/// Proactive compaction never fires mid-turn, however much history grows +/// between rounds. The old round-start gate projected growth at 1 byte/token, +/// so a single large tool result tripped it and wiped the working set the +/// model was about to use. Here usage is UNDER threshold and a ~6 KB tool +/// result lands mid-turn: the turn must complete with no summarize() call, +/// and the end-of-turn gate — reading the FINAL request's usage, not the +/// turn's cumulative input — must also stay quiet. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn stale_usage_plus_history_growth_triggers_handoff() { - // window 10_000, output 1_000 -> threshold = min(9_000, 9_000) = 9_000. - // req1 reports usage 8_500 (UNDER 9_000). Its response is a tool_call; - // the fake MCP returns a ~6 KB result, appended to history. At the - // conservative 1 byte/token estimate that's ~6_000 projected tokens, so - // projected ~14_500 >= 9_000 -> the next loop iteration hands off before - // the follow-up complete(). - // req1: tool_call + usage(8500) - // (tool result ~6KB appended) - // req2: summarize() (handoff) - // req3: final text +async fn history_growth_mid_turn_does_not_trigger_handoff() { + // window 10_000 -> threshold 9_000. + // req1: tool_call + usage(8500) (tool result ~6KB appended) + // req2: final text + usage(8900) cumulative 17_400 > 9_000, final < 9_000 + // Expected: exactly 2 requests — no summarize(). let llm = spawn_capturing_llm(vec![ { let mut v = openai_tool_call("tc1", "fake__tool_0", json!({})); @@ -1483,8 +1526,7 @@ async fn stale_usage_plus_history_growth_triggers_handoff() { json!({"prompt_tokens": 8500, "completion_tokens": 1, "total_tokens": 8501}); v }, - openai_text("handoff summary text"), - openai_text("done"), + openai_text_with_usage("done", 8900), ]) .await; let mut h = Harness::spawn_with_env( @@ -1493,8 +1535,6 @@ async fn stale_usage_plus_history_growth_triggers_handoff() { ("BUZZ_AGENT_MAX_CONTEXT_TOKENS", "10000"), ("BUZZ_AGENT_MAX_OUTPUT_TOKENS", "1000"), ("BUZZ_AGENT_MAX_HANDOFFS", "3"), - // Huge byte budget so the None-path byte fallback can't be what - // fires — only the token-mode growth estimate can explain it. ( "BUZZ_AGENT_MAX_HISTORY_BYTES", &(16 * 1024 * 1024).to_string(), @@ -1518,17 +1558,141 @@ async fn stale_usage_plus_history_growth_triggers_handoff() { ) .await; let _ = h.recv_until(|v| v["id"] == json!(p)).await; - // req1 (tool_call) + summarize (handoff) + req2 (done) = 3. Without the - // growth estimate we'd see only 2 (stale 8500 < 9000, no handoff). let captured = llm.captured.lock().await.len(); assert_eq!( - captured, 3, - "expected handoff after history grew past threshold (3 reqs), saw {captured} — \ - stale under-threshold usage skipped the growth estimate" + captured, 2, + "expected no handoff (2 reqs), saw {captured} — proactive compaction fired mid-turn \ + or the end-of-turn gate summed the turn's input instead of reading the final request" + ); + assert!( + !h.stderr_text().contains("handoff #"), + "no handoff log expected: {}", + h.stderr_text() ); h.shutdown().await; } +/// The end-of-turn gate compares the final request's MEASURED input usage +/// against the threshold — nothing else. The old round-start gate added a +/// 1-byte/token estimate of history appended since the measurement; at end of +/// turn the only such history is the final assistant reply, so keeping that +/// term would compact an under-threshold turn merely because its answer was +/// long. Here usage is 8500 < 9000 and the reply is ~2 KB: no handoff. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn large_final_reply_does_not_trigger_end_of_turn_handoff() { + let long_reply = "a".repeat(2048); + let llm = spawn_capturing_llm(vec![ + openai_text_with_usage(&long_reply, 8500), + openai_text("summary that must never be requested"), + ]) + .await; + let mut h = Harness::spawn_with_env( + &llm.url, + &[ + ("BUZZ_AGENT_MAX_CONTEXT_TOKENS", "10000"), + ("BUZZ_AGENT_MAX_OUTPUT_TOKENS", "1000"), + ("BUZZ_AGENT_MAX_HANDOFFS", "3"), + ( + "BUZZ_AGENT_MAX_HISTORY_BYTES", + &(16 * 1024 * 1024).to_string(), + ), + ], + ) + .await; + let sid = init_session(&mut h, json!([])).await; + let p = h + .send( + "session/prompt", + json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}), + ) + .await; + let _ = h.recv_until(|v| v["id"] == json!(p)).await; + assert_eq!( + llm.captured.lock().await.len(), + 1, + "gate must read measured usage (8500 < 9000) and ignore the reply's byte size" + ); + assert!(!h.stderr_text().contains("handoff #")); + h.shutdown().await; +} + +/// A cancelled turn must not compact. `run()` reports cancellation as +/// `Ok(StopReason::Cancelled)`, so a gate on `result.is_ok()` would spend a +/// summarize round trip the user just asked us to stop — and wipe history the +/// next prompt may want. +/// +/// The cancel must land while a PROVIDER call is in flight, not a tool call: +/// that is the path where `run()` itself consumes the cancel signal +/// (`cancel.changed()` in its `select!`), so nothing downstream would notice +/// a stray summarize. With a tool-call cancel the handoff's own cancel check +/// happens to fire first and an `is_ok()` gate passes by accident. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cancelled_turn_skips_end_of_turn_handoff() { + // req 1: tool_call + usage(950) — over threshold, seeds the gate + // req 2: HANG — cancel arrives here + // (req 3 would be the summarize an `is_ok()` gate issues) + let llm = spawn_capturing_llm_with_status(vec![ + (200, { + let mut v = openai_tool_call("tc1", "fake__tool_0", json!({})); + v["usage"] = json!({"prompt_tokens": 950, "completion_tokens": 1, "total_tokens": 951}); + v + }), + (200, json!(HANG)), + (200, openai_text("summary that must never be requested")), + ]) + .await; + let mut h = Harness::spawn_with_env( + &llm.url, + &[ + ("BUZZ_AGENT_MAX_CONTEXT_TOKENS", "1000"), + ("BUZZ_AGENT_MAX_OUTPUT_TOKENS", "100"), + ("BUZZ_AGENT_MAX_HANDOFFS", "3"), + ( + "BUZZ_AGENT_MAX_HISTORY_BYTES", + &(16 * 1024 * 1024).to_string(), + ), + ], + ) + .await; + let sid = init_session_with_fake_mcp(&mut h, &[("FAKE_MCP_TOOL_COUNT", "1")]).await; + + let p = h + .send( + "session/prompt", + json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}), + ) + .await; + // The tool result is reported before round 2's provider call is issued; + // wait for the hanging request to be captured so the cancel is guaranteed + // to land mid-call. + h.recv_until(|v| { + v.get("params") + .and_then(|p| p.get("update")) + .and_then(|u| u.get("status")) + .and_then(Value::as_str) + == Some("completed") + }) + .await; + let deadline = Instant::now() + Duration::from_secs(5); + while llm.captured.lock().await.len() < 2 { + assert!( + Instant::now() < deadline, + "round 2 provider call never issued" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + h.notify("session/cancel", json!({"sessionId": sid})).await; + let r = h.recv_until(|v| v["id"] == json!(p)).await; + assert_eq!(r["result"]["stopReason"], "cancelled", "got {r}"); + assert_eq!( + llm.captured.lock().await.len(), + 2, + "cancelled turn must not issue an end-of-turn summarize()" + ); + assert!(!h.stderr_text().contains("handoff #")); + h.shutdown().await; +} + /// `_Stop` hook that takes longer than `BUZZ_AGENT_HOOK_TIMEOUT_MS` /// must be treated as no-objection (fail-open). Agent stops normally. /// @@ -3241,10 +3405,10 @@ async fn recovery_shrinks_further_on_each_rung() { /// Gate 5, and the DIRECTION the clearing protects: not a spurious handoff, a /// MISSED one. After a reactive reset the stale `last_request_input_tokens` -/// describes history that no longer exists, and its paired byte baseline -/// describes the pre-reset (larger) history — so `grown` stays near zero and the -/// projection collapses to the stale sub-threshold token count. The gate goes -/// BLIND until history exceeds its pre-reset size. +/// describes history that no longer exists; it is sub-threshold by +/// construction (the request that overflowed reported no usage), so as long as +/// it survives the end-of-turn gate reads "under budget" and goes BLIND to an +/// oversized history that only the byte fallback would see. /// /// Constructing the divergence takes three turns, and two of the constraints are /// load-bearing — a first attempt with a simpler fixture produced traces @@ -3254,17 +3418,16 @@ async fn recovery_shrinks_further_on_each_rung() { /// usage is ever recorded, both variants sit at `None`, and the test measures /// nothing. /// * The post-recovery retry must report NO usage. A usage-bearing response -/// overwrites both fields with coherent values on the spot, which makes the -/// clear genuinely redundant and the mutant equivalent. The reachable window -/// is exactly when the retry omits usage and the stale pair survives. -/// Turn 3 then carries a large prompt: a cleared baseline falls through to the -/// byte signal and hands off, while the stale pair projects -/// `10 + (190KB - 100KB)` = ~90k tokens, under the 180k threshold, and does not. +/// overwrites the reading on the spot, which makes the clear genuinely +/// redundant and the mutant equivalent. The reachable window is exactly when +/// the retry omits usage and the stale reading survives. +/// Turn 3 then carries a large prompt: a cleared reading falls through to the +/// byte signal and hands off, while the stale `Some(10)` stays under the 180k +/// threshold and does not. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn reactive_reset_clears_usage_baseline_so_the_gate_is_not_blind() { // ~100 KB: under the 180 KB byte-fallback threshold, so turn 1 does NOT - // trip the proactive gate, but large enough to be the stale `measured_bytes` - // that suppresses `grown` later. + // trip the proactive gate and a usage reading gets recorded. let mut medium = String::with_capacity(100 * 1024); medium.push_str("turn-one-medium "); while medium.len() < 100 * 1024 { @@ -3278,14 +3441,14 @@ async fn reactive_reset_clears_usage_baseline_so_the_gate_is_not_blind() { } let llm = spawn_capturing_llm_with_status(vec![ - // Turn 1: succeeds, reporting a SMALL usage reading against a ~100 KB - // history. This is the pair that goes stale. + // Turn 1: succeeds, reporting a SMALL usage reading. This is the + // reading that goes stale. (200, openai_text_with_usage("ack-medium", 10)), // Turn 2: the overflow. (400, openai_context_length_error()), // Turn 2: the forced handoff's summarize. (200, openai_text("forced summary")), - // Turn 2: the retry — NO usage block, so the baseline is not refreshed. + // Turn 2: the retry — NO usage block, so the reading is not refreshed. (200, openai_text("recovered, no usage reported")), // Turn 3: with a cleared baseline a gated summarize comes first; with a // stale one this slot is the completion instead. Spares so an exhausted @@ -3313,7 +3476,7 @@ async fn reactive_reset_clears_usage_baseline_so_the_gate_is_not_blind() { .await; let sid = init_session(&mut h, json!([])).await; - // Turn 1: under threshold, records the usage pair. + // Turn 1: under threshold, records the usage reading. let p0 = h .send( "session/prompt", @@ -3329,7 +3492,7 @@ async fn reactive_reset_clears_usage_baseline_so_the_gate_is_not_blind() { assert!(r0.get("error").is_none(), "turn 1 should succeed: {r0}"); assert!( !h.stderr_text().contains("handoff #"), - "precondition: turn 1 must NOT hand off, or no usage pair is recorded and this test \ + "precondition: turn 1 must NOT hand off, or no usage reading is recorded and this test \ measures nothing. stderr={}", h.stderr_text() ); @@ -3356,8 +3519,8 @@ async fn reactive_reset_clears_usage_baseline_so_the_gate_is_not_blind() { ); let handoffs_after_turn2 = h.stderr_text().matches("handoff #").count(); - // Turn 3: large prompt. A cleared baseline sees it via the byte signal and - // hands off; a stale pair under-projects and stays blind. + // Turn 3: large prompt. A cleared reading sees it via the byte signal and + // hands off; a stale reading stays blind. let p2 = h .send( "session/prompt", @@ -3376,54 +3539,37 @@ async fn reactive_reset_clears_usage_baseline_so_the_gate_is_not_blind() { assert!( handoffs_after_turn3 > handoffs_after_turn2, "turn 3 must produce a GATED handoff ({handoffs_after_turn2} before, \ - {handoffs_after_turn3} after): the reactive reset must clear the usage baseline, or the \ - proactive gate under-projects and stays blind to an oversized history. stderr={stderr}" + {handoffs_after_turn3} after): the reactive reset must clear the usage reading, or the \ + end-of-turn gate stays blind to an oversized history. stderr={stderr}" ); h.shutdown().await; } -// ─── Tests: per-turn handoff cap semantics ─────────────────────────────────── +// ─── Tests: end-of-turn handoff semantics ──────────────────────────────────── -/// A session that has already performed N handoffs in previous turns must still -/// compact on subsequent turns — the per-session lifetime kill switch is gone. -/// -/// Mechanism: the gate fires at the start of each round, comparing -/// `last_request_input_tokens` (stored by the previous response) against the -/// token threshold. So: -/// - Turn 1 complete() returns usage=950 (> threshold=900). Turn ends; usage stored. -/// - Turn 2 round 0: 950 >= 900 → handoff. post-handoff complete() returns usage=950. -/// Session `handoff_count` is now 1; `turn_handoff_count` was just reset to 0 at -/// turn start and is now 1. -/// - Turn 3 round 0: `turn_handoff_count` resets to 0; session count is 1 but -/// the gate uses `turn_handoff_count` → cap not reached → handoff fires again. -/// -/// Without the fix (`handoff_count` compared against cap, never reset): -/// session count after turn 2 = 1 >= max_handoffs=1 → gate permanently blocked -/// for all subsequent turns → history grows until provider wall. +/// A session compacts on every turn that ends over threshold — there is no +/// per-session lifetime kill switch, and the magnitude of `max_handoffs` (here +/// 1) does not bound how many turns may compact. After a compaction the stale +/// usage is cleared, so a turn that ends under threshold does not fire. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn handoff_cap_resets_per_turn_not_per_session() { - // LLM call sequence: - // req 1: turn 1 complete() → usage=950 (over threshold) - // req 2: turn 2 pre-flight summarize → summary text - // req 3: turn 2 complete() → usage=950 (re-arms gate for turn 3) - // req 4: turn 3 pre-flight summarize → summary text ← cap reset proves this fires - // req 5: turn 3 complete() → done +async fn handoff_fires_on_each_over_threshold_turn() { + // threshold = 900. + // req 1: turn 1 complete() → usage=950 → req 2: summarize + // req 3: turn 2 complete() → usage=950 → req 4: summarize + // req 5: turn 3 complete() → usage=10 → (no summarize) let llm = spawn_capturing_llm(vec![ - openai_text_with_usage("ack-t1", 950), // turn 1: stores high usage - openai_text("summary-t2"), // turn 2: pre-flight summarize - openai_text_with_usage("done-t2", 950), // turn 2: post-handoff, re-arms gate - openai_text("summary-t3"), // turn 3: pre-flight summarize (cap reset) - openai_text_with_usage("done-t3", 10), // turn 3: post-handoff complete + openai_text_with_usage("ack-t1", 950), + openai_text("summary-t1"), + openai_text_with_usage("ack-t2", 950), + openai_text("summary-t2"), + openai_text_with_usage("ack-t3", 10), ]) .await; - let mut h = Harness::spawn_with_env( &llm.url, &[ ("BUZZ_AGENT_MAX_CONTEXT_TOKENS", "1000"), ("BUZZ_AGENT_MAX_OUTPUT_TOKENS", "100"), - // Cap of 1 per turn. Before the fix this permanently disables the - // gate once session handoff_count reaches 1. ("BUZZ_AGENT_MAX_HANDOFFS", "1"), ( "BUZZ_AGENT_MAX_HISTORY_BYTES", @@ -3434,117 +3580,42 @@ async fn handoff_cap_resets_per_turn_not_per_session() { .await; let sid = init_session(&mut h, json!([])).await; - // Turn 1: no prior usage; preflight skips (byte-fallback not triggered by - // tiny prompt). complete() stores usage=950. - let p1 = h - .send( - "session/prompt", - json!({"sessionId": sid, "prompt": [{"type":"text","text":"turn 1"}]}), - ) - .await; - let _ = h.recv_until(|v| v["id"] == json!(p1)).await; - assert_eq!( - llm.captured.lock().await.len(), - 1, - "turn 1 must produce exactly 1 LLM request" - ); - - // Turn 2: 950 >= threshold=900 → handoff fires. Session handoff_count: 1. - let p2 = h - .send( - "session/prompt", - json!({"sessionId": sid, "prompt": [{"type":"text","text":"turn 2"}]}), - ) - .await; - let _ = h.recv_until(|v| v["id"] == json!(p2)).await; - assert_eq!( - llm.captured.lock().await.len(), - 3, - "turn 2 must produce 2 LLM requests (summarize + complete), 3 total" - ); - let stderr = h.stderr_text(); - assert!( - stderr.contains("handoff #1"), - "expected first handoff log after turn 2; got: {stderr}" - ); - - // Turn 3: turn_handoff_count resets to 0 → gate fires again despite - // session handoff_count=1 == cap=1. - let p3 = h - .send( - "session/prompt", - json!({"sessionId": sid, "prompt": [{"type":"text","text":"turn 3"}]}), - ) - .await; - let _ = h.recv_until(|v| v["id"] == json!(p3)).await; - assert_eq!( - llm.captured.lock().await.len(), - 5, - "turn 3 must also produce 2 LLM requests (per-turn cap reset → handoff fires again), \ - 5 total" - ); + for (turn, expected_reqs) in [(1, 2usize), (2, 4), (3, 5)] { + let p = h + .send( + "session/prompt", + json!({"sessionId": sid, "prompt": [{"type":"text","text": format!("turn {turn}")}]}), + ) + .await; + let _ = h.recv_until(|v| v["id"] == json!(p)).await; + assert_eq!( + llm.captured.lock().await.len(), + expected_reqs, + "after turn {turn}" + ); + } let stderr = h.stderr_text(); - assert!( - stderr.contains("handoff #2"), - "expected second handoff log after turn 3 (cap reset); got: {stderr}" - ); - + assert!(stderr.contains("handoff #1"), "got: {stderr}"); + assert!(stderr.contains("handoff #2"), "got: {stderr}"); + assert!(!stderr.contains("handoff #3"), "got: {stderr}"); h.shutdown().await; } -/// Within a single turn, the per-turn cap still bounds the number of handoffs. -/// A turn that exceeds `max_handoffs` compaction attempts must emit a WARN and -/// fall back to truncation — it must NOT compact indefinitely. -/// -/// Mechanism: with cap=1 and a multi-round turn (tool call in round 1 → round 2), -/// the pre-flight handoff fires at the start of round 1 (usage from a *previous* -/// turn is high). After the compaction, the post-handoff complete() in round 1 -/// returns a tool call, causing a second round. Round 2's preflight sees that -/// turn_handoff_count=1 == max_handoffs=1, so it refuses and emits WARN. -/// -/// A steer is injected while the run is active to prove that the steer path -/// does NOT reset `handoff_attempts` — the cap must still fire on round 1 with -/// no second summarize call. -/// -/// This test requires a fake MCP server to produce a tool-call round. -/// It drives via `fake-mcp` — the same binary used in other multi-round tests. +/// `BUZZ_AGENT_MAX_HANDOFFS=0` disables proactive compaction: an over-threshold +/// turn ends with no summarize() call. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn handoff_cap_binds_within_a_single_turn() { - // LLM call sequence in turn 2 (turn 1 seeds the usage): - // req 1: turn 1 complete() → usage=950 (over threshold=900) - // req 2: turn 2 round 0 summarize() → summary (handoff_attempts: 0→1) - // req 3: turn 2 round 0 complete() → tool_call + usage=950 (re-arms gate) - // [fake-mcp tool executes; steer queued while run is active] - // req 4: turn 2 round 1 preflight → 950 >= 900 AND attempts=1 >= max=1 - // → WARN, skip (cap exhausted for this turn) - // req 5: turn 2 round 1 complete() → end_turn (steer text folded into messages) - let fake_mcp = env!("CARGO_BIN_EXE_fake-mcp"); - // Build a tool-call response that also carries usage so the gate re-arms - // on round 1's preflight (without usage, last_request_input_tokens is None - // after the handoff clears it, and the byte-fallback won't fire on tiny history). - let tool_call_with_usage = { - let mut v = openai_tool_call("tc-1", "test_tool", json!({})); - v["usage"] = json!({ - "prompt_tokens": 950u64, - "completion_tokens": 5, - "total_tokens": 955, - }); - v - }; +async fn max_handoffs_zero_disables_end_of_turn_handoff() { let llm = spawn_capturing_llm(vec![ - openai_text_with_usage("seed", 950), // turn 1: seed high usage - openai_text("handoff-summary"), // turn 2 round 0: summarize - tool_call_with_usage, // turn 2 round 0: tool call + usage (re-arms) - openai_text_with_usage("end_turn_text", 10), // turn 2 round 1: final answer + openai_text_with_usage("ack", 950), + openai_text("summary that must never be requested"), ]) .await; - let mut h = Harness::spawn_with_env( &llm.url, &[ ("BUZZ_AGENT_MAX_CONTEXT_TOKENS", "1000"), ("BUZZ_AGENT_MAX_OUTPUT_TOKENS", "100"), - ("BUZZ_AGENT_MAX_HANDOFFS", "1"), + ("BUZZ_AGENT_MAX_HANDOFFS", "0"), ( "BUZZ_AGENT_MAX_HISTORY_BYTES", &(16 * 1024 * 1024).to_string(), @@ -3552,190 +3623,35 @@ async fn handoff_cap_binds_within_a_single_turn() { ], ) .await; - - // Init with the fake MCP server so test_tool is available. - h.send( - "initialize", - json!({"protocolVersion":1,"clientCapabilities":{}}), - ) - .await; - let _ = h.recv().await; - h.send( - "session/new", - json!({ - "cwd": "/tmp", - "mcpServers": [{ - "name": "cap_test", - "command": fake_mcp, - "args": [], - "env": [{ "name": "FAKE_MCP_TOOL_COUNT", "value": "1" }], - }], - }), - ) - .await; - let r = h - .recv_until(|v| v.get("result").is_some() || v.get("error").is_some()) - .await; - let sid = r["result"]["sessionId"].as_str().unwrap().to_owned(); - - // Turn 1: seed high usage. - let p1 = h - .send( - "session/prompt", - json!({"sessionId": sid, "prompt": [{"type":"text","text":"seed"}]}), - ) - .await; - let _ = h.recv_until(|v| v["id"] == json!(p1)).await; - - // Turn 2: triggers a handoff at round 0, then a tool call, then round 1 - // where the cap is already exhausted. A steer is injected while the run - // is active to prove mid-turn steers cannot reset `handoff_attempts`. - let p2 = h + let sid = init_session(&mut h, json!([])).await; + let p = h .send( "session/prompt", - json!({"sessionId": sid, "prompt": [{"type":"text","text":"do work"}]}), + json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}), ) .await; - - // Drain until the final response, approving tool-permission requests, - // capturing the activeRunId once it is broadcast, sending one steer, - // and verifying that it is accepted in the live run. - let mut run_id: Option = None; - let mut steer_id: i64 = -1; - let mut steer_accepted = false; - loop { - let v = h.recv().await; - - // Capture the run id from the first session/update that carries it, - // then immediately queue a steer. This must happen before round 1 so - // the steer text is present but the cap check still fires — proving - // the counter is not reset by the steer path. - if run_id.is_none() { - if let Some(rid) = v["params"]["update"]["_meta"]["goose"]["activeRunId"].as_str() { - run_id = Some(rid.to_owned()); - steer_id = h - .send( - "_goose/unstable/session/steer", - json!({ - "sessionId": sid, - "expectedRunId": rid, - "prompt": [{"type":"text","text":"STEER-CANARY: also consider the edge case"}], - }), - ) - .await; - } - } - - // Steer response: assert it was accepted in the live run. - if steer_id >= 0 && v["id"] == json!(steer_id) { - assert!( - v.get("result").is_some(), - "steer must be accepted while the run is active; got: {v}" - ); - assert_eq!( - v["result"]["runId"].as_str(), - run_id.as_deref(), - "steer must reference the live run id" - ); - steer_accepted = true; - continue; - } - - if v.get("method") == Some(&json!("session/request_permission")) { - let id = v["id"].clone(); - h.write(json!({ - "jsonrpc": "2.0", - "id": id, - "result": { "outcome": { "outcome": "selected", "optionId": "allow" } }, - })) - .await; - continue; - } - if v["id"] == json!(p2) { - assert!( - v.get("result").is_some(), - "turn 2 must succeed even when cap blocks round-1 handoff; got: {v}" - ); - break; - } - } - - assert!( - steer_accepted, - "steer was never accepted during turn 2; the steer arm is missing coverage" - ); - - // 4 LLM requests: seed + summarize + tool-call-with-usage + final-complete. - let count = llm.captured.lock().await.len(); - assert_eq!( - count, 4, - "expected 4 LLM requests (seed + summarize + tool-call + final); got {count}" - ); - - let stderr = h.stderr_text(); - assert!( - stderr.contains("handoff cap reached"), - "expected cap-reached WARN in stderr; got: {stderr}" - ); - assert!( - stderr.contains("reason=\"preflight\""), - "expected reason=\"preflight\" field in cap WARN; got: {stderr}" - ); - assert!( - stderr.contains("handoff_attempts="), - "expected handoff_attempts field in cap WARN; got: {stderr}" - ); - assert!( - stderr.contains("max_handoffs="), - "expected max_handoffs field in cap WARN; got: {stderr}" - ); - + let _ = h.recv_until(|v| v["id"] == json!(p)).await; + assert_eq!(llm.captured.lock().await.len(), 1); + assert!(!h.stderr_text().contains("handoff #")); h.shutdown().await; } -/// A failing `summarize()` call must still consume one slot from the per-turn -/// handoff-attempt budget. Before the fix, `handoff_count` was incremented only -/// on a successful compaction; a flaky summarizer could be retried indefinitely -/// within a turn. The fix moves the increment to before `summarize()`. -/// -/// Proof: with `max_handoffs=1` and a multi-round turn: -/// - Round 0 preflight: threshold met, attempts: 0→1, summarize() fails → Skipped. -/// - Round 1 preflight: attempts=1 >= cap=1 → WARN (cap hit despite no successful -/// compaction). Without the pre-summarize increment, attempts would still be 0 -/// here and a second summarize() would be attempted — the bug. +/// A failing end-of-turn `summarize()` is logged and skipped: the turn still +/// succeeds and history is left intact for the next turn (truncation, not a +/// wipe, is the fallback). The next over-threshold turn simply tries again. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn failed_summarize_burns_handoff_attempt_budget() { - // We need the summarize() call to fail. The summarize path uses the same - // fake LLM server; we queue an HTTP error body for the summarize request. - // But our spawn_capturing_llm always returns 200, so we use a non-OpenAI- - // shaped response that the agent will treat as an error (missing `choices`). - // - // LLM call sequence: - // req 1: turn 1 complete() → usage=950 (seeds the gate) - // req 2: turn 2 round 0 summarize() → malformed response (treated as error) - // handoff_attempts incremented to 1 BEFORE this - // req 3: turn 2 round 0 complete() → tool_call + usage=950 (re-arms gate) - // req 4: turn 2 round 1 preflight → cap reached: WARN (attempts=1 >= max=1) - // req 5: turn 2 round 1 complete() → end_turn - let fake_mcp = env!("CARGO_BIN_EXE_fake-mcp"); - let bad_summary_response = json!({ "error": "upstream unavailable" }); // no `choices` - let tool_call_with_usage = { - let mut v = openai_tool_call("tc-2", "test_tool", json!({})); - v["usage"] = json!({ - "prompt_tokens": 950u64, - "completion_tokens": 5, - "total_tokens": 955, - }); - v - }; +async fn failed_end_of_turn_summarize_leaves_history_intact() { + // spawn_capturing_llm always returns 200, so a non-OpenAI-shaped body + // (missing `choices`) is what makes summarize() fail. + // req 1: turn 1 complete() → usage=950 + // req 2: turn 1 summarize() → malformed → "handoff failed" WARN + // req 3: turn 2 complete() → must still see turn 1's messages let llm = spawn_capturing_llm(vec![ - openai_text_with_usage("seed", 950), // turn 1: seed usage - bad_summary_response, // turn 2 round 0: summarize fails - tool_call_with_usage, // turn 2 round 0: complete → tool call - openai_text_with_usage("done", 10), // turn 2 round 1: final answer + openai_text_with_usage("ack-seed-marker", 950), + json!({ "error": "upstream unavailable" }), + openai_text_with_usage("done", 10), ]) .await; - let mut h = Harness::spawn_with_env( &llm.url, &[ @@ -3749,78 +3665,40 @@ async fn failed_summarize_burns_handoff_attempt_budget() { ], ) .await; + let sid = init_session(&mut h, json!([])).await; - h.send( - "initialize", - json!({"protocolVersion":1,"clientCapabilities":{}}), - ) - .await; - let _ = h.recv().await; - h.send( - "session/new", - json!({ - "cwd": "/tmp", - "mcpServers": [{ - "name": "budget_test", - "command": fake_mcp, - "args": [], - "env": [{ "name": "FAKE_MCP_TOOL_COUNT", "value": "1" }], - }], - }), - ) - .await; - let r = h - .recv_until(|v| v.get("result").is_some() || v.get("error").is_some()) - .await; - let sid = r["result"]["sessionId"].as_str().unwrap().to_owned(); - - // Turn 1: seed high usage. let p1 = h .send( "session/prompt", - json!({"sessionId": sid, "prompt": [{"type":"text","text":"seed"}]}), - ) - .await; - let _ = h.recv_until(|v| v["id"] == json!(p1)).await; - - // Turn 2: round 0 summarize fails, but attempts was already incremented. - // Round 1 preflight must see cap hit and emit WARN. - let p2 = h - .send( - "session/prompt", - json!({"sessionId": sid, "prompt": [{"type":"text","text":"work"}]}), + json!({"sessionId": sid, "prompt": [{"type":"text","text":"seed-prompt-marker"}]}), ) .await; - - loop { - let v = h.recv().await; - if v.get("method") == Some(&json!("session/request_permission")) { - let id = v["id"].clone(); - h.write(json!({ - "jsonrpc": "2.0", - "id": id, - "result": { "outcome": { "outcome": "selected", "optionId": "allow" } }, - })) - .await; - continue; - } - if v["id"] == json!(p2) { - assert!(v.get("result").is_some(), "turn 2 must succeed; got: {v}"); - break; - } - } - + let r1 = h.recv_until(|v| v["id"] == json!(p1)).await; + assert!(r1.get("result").is_some(), "turn 1 must succeed; got: {r1}"); + assert_eq!(llm.captured.lock().await.len(), 2); let stderr = h.stderr_text(); - // Round 0: the failed summarize should warn about the failure. assert!( stderr.contains("handoff failed") || stderr.contains("handoff returned empty"), "expected summarize-failure WARN; got: {stderr}" ); - // Round 1: cap must be hit (attempts=1 from the failed attempt). assert!( - stderr.contains("handoff cap reached"), - "expected cap-reached WARN after failed summarize burned the attempt; got: {stderr}" + !stderr.contains("handoff #"), + "no compaction must be logged: {stderr}" ); + let p2 = h + .send( + "session/prompt", + json!({"sessionId": sid, "prompt": [{"type":"text","text":"second"}]}), + ) + .await; + let _ = h.recv_until(|v| v["id"] == json!(p2)).await; + let captured = llm.captured.lock().await; + assert_eq!(captured.len(), 3); + let body = captured[2].to_string(); + assert!( + body.contains("seed-prompt-marker") && body.contains("ack-seed-marker"), + "turn 2 must run on intact history after a failed summarize: {body}" + ); h.shutdown().await; }