jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-12 04:12:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

buzz-agent: gate end-of-turn handoff on measured usage alone

Browse Source 
Review finding (Wren): `should_handoff()` still routed its measured-usage
arm through `projected_handoff_input_tokens()`, which added a 1-byte/token
estimate of history appended since the usage was recorded. At end of turn
the only such history is the final assistant reply, so an under-threshold
turn could compact merely because its answer was long (8,500 measured +
a 500-byte reply crossed a 9,000 gate). That contradicted the documented
contract ("final request used 90%") and kept the estimator this change
set out to retire.

- handoff.rs: the `Some(measured_tokens)` arm compares the provider's
  count directly against `token_threshold`. `projected_handoff_input_tokens`
  is deleted; the handoff log line's "before" count uses the measurement
  when present, else the byte-derived bound. The `None` byte fallback is
  unchanged. Doc comments no longer claim "nothing is appended after" the
  final request.
- agent.rs / lib.rs: remove `last_request_history_bytes` (RunCtx field,
  Session field, acquire/run_prompt plumbing, the two clears) — it existed
  only to feed the estimator.
- tests/regressions.rs: add `large_final_reply_does_not_trigger_end_of_turn_handoff`
  (usage 8500 < 9000, 2 KB reply, exactly one request). Fails on the
  parent commit; passes here. Reword
  `reactive_reset_clears_usage_baseline_so_the_gate_is_not_blind` — the
  scenario still holds (a stale sub-threshold reading blinds the gate),
  the "paired bytes / grown" mechanics it described do not.

Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz>
Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz>
This commit is contained in:
Meli
2026-08-22 18:31:26 -04:00
parent 25d5a0ca66
commit 7f3b63ea7f
4 changed files with 88 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
crates/buzz-agent/src/agent.rs
View File

@@ -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<u64>,
/// 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<usize>,
/// 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,
@@ -448,7 +442,6 @@ impl RunCtx<'_> {
// 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),
@@ -463,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)
@@ -488,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).

65
crates/buzz-agent/src/handoff.rs
View File

@@ -65,9 +65,9 @@ enum Reseat {
impl RunCtx<'_> {
/// Proactive compaction, run once after a turn completes. Gates on the
/// provider's measured input usage for the turn's final request — no
/// growth estimate is needed because nothing is appended after it — and
/// compacts so the NEXT turn starts on a fresh context.
/// 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
@@ -83,7 +83,6 @@ impl RunCtx<'_> {
// 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;
*self.last_request_history_bytes = None;
}
outcome
}
@@ -190,7 +189,11 @@ impl RunCtx<'_> {
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,
@@ -259,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
@@ -280,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.
@@ -491,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 {

9
crates/buzz-agent/src/lib.rs
View File

@@ -87,9 +87,6 @@ struct Session {
/// resets the context). Drives the end-of-turn handoff gate; see
/// [`RunCtx::end_of_turn_handoff`].
last_request_input_tokens: Option<u64>,
/// 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<usize>,
effective_system_prompt: Arc<str>,
/// 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<App>, 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<App>, 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<App>, 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,
@@ -769,7 +763,6 @@ async fn run_prompt(app: Arc<App>, 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
@@ -873,7 +866,6 @@ async fn acquire_session(
Option<String>,
usize,
Option<u64>,
Option<usize>,
watch::Receiver<bool>,
Arc<str>,
Option<String>,
@@ -916,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,

85
crates/buzz-agent/tests/regressions.rs
View File

@@ -1572,6 +1572,50 @@ async fn history_growth_mid_turn_does_not_trigger_handoff() {
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
@@ -3361,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
@@ -3374,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 {
@@ -3398,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
@@ -3433,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",
@@ -3449,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()
);
@@ -3476,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",
@@ -3496,8 +3539,8 @@ 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;
}