From 099bacbe6ae39d6255dc602a267de075d4b89ef2 Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Wed, 12 Aug 2026 15:32:05 +1000 Subject: [PATCH] refactor(buzz-agent): express steer and compaction as goose Operations Steps 4 and 5 of PLANS/BUZZ_OPERATIONS_MIGRATION.md. Behaviour is unchanged: steer.rs, cancel.rs and the real-MCP integration tests all pass without modification. BuzzSteerOperation drains the steer queue; BuzzCompactionOperation wraps goose's check_if_compaction_needed/compact_messages plus the _PostCompact re-injection that buzz's old context handoff existed for. These need a second StateMachine (round_start), not more steps on the existing one: round_gate also runs when a turn wants to END, and draining steers or compacting there would change when they happen. Since step() stops at the first operation that applies, the start machine runs to exhaustion -- a turn that both steers and compacts needs two passes. The gate now carries StateEffect rather than Message, because compaction replaces the conversation instead of appending to it. apply_effects reports whether state actually changed, which is what stops a no-op-applied operation from spinning the loop. Not goose's CompactionOperation: it yields to the client and emits a user-facing notification. In buzz a yield ends the turn, and the notification would post 'compacting' into the channel -- so a long conversation would stop mid-work and narrate its own housekeeping. Nearly dropped the [PostCompact] prefix on the re-injected state while moving it. Caught by re-reading the old function before deleting it; the model uses that prefix to tell re-injected state from a user turn. 74 unit + 21 integration green, handtest --all 18/18. Signed-off-by: Michael Neale Co-authored-by: Michael Neale --- crates/buzz-agent/src/loop_drive.rs | 172 ++++++++++++---------------- crates/buzz-agent/src/ops.rs | 170 ++++++++++++++++++++++++++- 2 files changed, 245 insertions(+), 97 deletions(-) diff --git a/crates/buzz-agent/src/loop_drive.rs b/crates/buzz-agent/src/loop_drive.rs index 30f37219d8..3d446d3b19 100644 --- a/crates/buzz-agent/src/loop_drive.rs +++ b/crates/buzz-agent/src/loop_drive.rs @@ -155,9 +155,10 @@ pub async fn run_turn( let max_rounds = ctx.max_rounds.unwrap_or(DEFAULT_MAX_ROUNDS); let mut reflections = 0usize; - // First decision expressed as a goose `Operation` rather than inline - // control flow. The others follow one at a time; see - // `PLANS/BUZZ_OPERATIONS_MIGRATION.md` for the order and why. + // Every loop decision is a goose `Operation` -- see + // `PLANS/BUZZ_OPERATIONS_MIGRATION.md`. Two machines, because they run at + // different points: `start_machine` before inference, `machine` at the + // round gate and again when a turn wants to end. let outcome = crate::ops::Outcome::new(); let machine = crate::ops::round_gate( max_rounds, @@ -167,6 +168,16 @@ pub async fn run_turn( .map(|extension| (Arc::clone(ctx.agent), extension.to_string())), ctx.require_reply, ); + let start_machine = crate::ops::round_start( + ctx.steers.clone(), + crate::ops::BuzzCompactionOperation::new( + ctx.model.clone(), + Arc::clone(ctx.agent), + ctx.hook_extension.map(str::to_string), + ctx.session_id.to_string(), + ), + ctx.cancel.clone(), + ); let (event_tx, mut event_rx) = tokio::sync::mpsc::channel(16); // Operations may emit; nothing in this step does, but a dropped receiver // would silently swallow events from the ones that follow. @@ -189,23 +200,24 @@ pub async fn run_turn( let reason = outcome.take().unwrap_or(StopReason::EndTurn); return (Ok(reason), tokens, state.conversation().messages().to_vec()); } - Gate::Applied(messages) => { - for message in messages { - state.push(message); - } + Gate::Applied(effects) => { + apply_effects(&mut state, effects); } Gate::Open => {} } - // Steers land here, at the round boundary — never mid-inference, - // where they would race a partially streamed response. - for message in ctx.steers.drain().await { - state.push(message); - } - - // Compaction at the round boundary, never mid-stream. - if let Err(e) = maybe_compact(&ctx, &mut state).await { - tracing::warn!(error = %e, "compaction failed; continuing uncompacted"); + // Steering and compaction, both at the round boundary and never + // mid-inference where they would race a partially streamed response. + // Run to exhaustion: `step` stops at the first operation that applies, + // so a turn that both steers and compacts needs more than one pass. + loop { + let Gate::Applied(effects) = gate(&start_machine, state.session(), &emitter).await + else { + break; + }; + if !apply_effects(&mut state, effects) { + break; + } } match round(&ctx, &mut state, &mut tokens, &mut reflections).await { @@ -231,21 +243,20 @@ pub async fn run_turn( // cancel for no gain. if !ctx.cancel.is_cancelled() { match gate(&machine, state.session(), &emitter).await { - Gate::Applied(messages) if !messages.is_empty() => { - for message in messages { - state.push(message); + Gate::Applied(effects) => { + if apply_effects(&mut state, effects) { + continue; } - continue; + // Applied without changing state: the model has no + // new work, so the turn still ends rather than the + // loop spinning. } Gate::Ended => { let reason = outcome.take().unwrap_or(StopReason::EndTurn); ctx.steers.clear().await; return (Ok(reason), tokens, state.conversation().messages().to_vec()); } - // An operation that applied without appending anything - // has not given the model new work, so the turn still - // ends -- otherwise the loop would spin. - Gate::Applied(_) | Gate::Open => {} + Gate::Open => {} } } @@ -265,9 +276,9 @@ pub async fn run_turn( enum Gate { /// No operation applied; the loop proceeds. Open, - /// An operation applied and appended messages; the loop continues with - /// them in history rather than ending. - Applied(Vec), + /// An operation applied; the loop continues with its effects in state + /// rather than ending. + Applied(Vec), /// An operation ended the turn. Ended, } @@ -284,22 +295,7 @@ async fn gate( ) -> Gate { match machine.step(session, emitter).await { Ok(Some(result)) if result.yield_to_client => Gate::Ended, - Ok(Some(result)) => Gate::Applied( - result - .effects - .into_iter() - .filter_map(|effect| match effect { - goose::agents::state_machine::StateEffect::AppendMessage(message) => { - Some(message) - } - // buzz's operations only append. Other effects belong to - // goose's own operations (recipes, approval, extension - // data) and have no meaning in this loop; ignoring them - // is correct rather than lossy. - _ => None, - }) - .collect(), - ), + Ok(Some(result)) => Gate::Applied(result.effects), Ok(None) => Gate::Open, // A failing gate must not take the turn down with it: the operations // here are guards, and a guard that errors should let the round @@ -311,6 +307,43 @@ async fn gate( } } +/// Apply an operation's effects to the turn's state. +/// +/// buzz's operations produce two kinds: appended messages (objections, +/// reminders, steers) and a whole replaced conversation (compaction). Other +/// `StateEffect` variants belong to goose's own operations -- recipes, +/// approval, extension data -- and have no meaning in this loop, so ignoring +/// them is correct rather than lossy. +/// +/// Returns whether anything was actually applied: an operation that applied +/// without changing state has given the model no new work, and treating that +/// as progress would spin the loop. +fn apply_effects( + state: &mut crate::turn_state::TurnState, + effects: Vec, +) -> bool { + use goose::agents::state_machine::StateEffect; + + let mut changed = false; + for effect in effects { + match effect { + StateEffect::AppendMessage(message) => { + state.push(message); + changed = true; + } + StateEffect::ReplaceConversation { conversation, .. } => { + state.replace(conversation); + // The running total described a conversation that no longer + // exists; let goose re-estimate from the compacted messages. + state.set_total_tokens(None); + changed = true; + } + _ => {} + } + } + changed +} + /// One inference plus, if the model asked for them, one batch of tool calls. async fn round( ctx: &TurnContext<'_>, @@ -510,59 +543,6 @@ fn push_tool_results( state.push(message); } -/// Compact history when it is close to the context limit, then re-inject the -/// state `_PostCompact` reports so a todo list survives the truncation. -async fn maybe_compact( - ctx: &TurnContext<'_>, - state: &mut crate::turn_state::TurnState, -) -> anyhow::Result<()> { - let conversation = state.conversation(); - let provider = ctx.model.provider().await; - - if !goose::context_mgmt::check_if_compaction_needed( - provider.as_ref(), - &conversation, - None, - state.session(), - ) - .await? - { - return Ok(()); - } - - let model_config = ctx.model.config().await; - let result = goose::context_mgmt::compact_messages( - provider.as_ref(), - &model_config, - ctx.session_id, - &conversation, - false, - ) - .await?; - - state.replace(result.conversation); - // Compaction resets the context: the old running total describes a - // conversation that no longer exists, so drop it and let goose re-estimate - // from the compacted messages. - state.set_total_tokens(None); - tracing::info!(target: "buzz_agent::compaction", "history compacted"); - - // `_PostCompact` state is what buzz-agent's handoff existed to preserve. - if let Some(extension) = ctx.hook_extension { - if let Some(reported) = - crate::hooks::post_compact_state(ctx.agent, state.session(), extension).await - { - state.push( - Message::user() - .with_text(format!("[PostCompact] {reported}")) - .with_visibility(false, true), - ); - } - } - - Ok(()) -} - /// Preserve buzz-agent's error taxonomy so the harness's JSON-RPC code mapping /// stays meaningful. fn classify_provider_error(msg: &str) -> AgentError { diff --git a/crates/buzz-agent/src/ops.rs b/crates/buzz-agent/src/ops.rs index 5c24455afd..02a2a19948 100644 --- a/crates/buzz-agent/src/ops.rs +++ b/crates/buzz-agent/src/ops.rs @@ -23,7 +23,7 @@ use std::sync::{Arc, Mutex}; use anyhow::Result; use async_trait::async_trait; use goose::agents::state_machine::{ - applied, not_applicable, yielded, Emitter, Operation, OperationResult, + applied, not_applicable, yielded, Emitter, Operation, OperationResult, StateEffect, }; use goose::agents::Agent; use goose::conversation::Conversation; @@ -401,6 +401,148 @@ impl Operation for BuzzReplyGuardOperation { } } +/// Drains `session/steer` messages into the conversation at the round +/// boundary. +/// +/// Steering is buzz's own: goose has an equivalent queue on `Agent`, but +/// `drain_pending_steers` is `pub(crate)` — only `Agent::reply` can consume +/// it, and buzz owns the loop. Expressing the drain as an operation puts it +/// under the same gate as the other loop decisions instead of beside them. +/// +/// Semantics are unchanged from the inline drain: messages land at the round +/// boundary and never mid-inference, where they would race a partially +/// streamed response. +pub struct BuzzSteerOperation { + steers: crate::steer::SteerQueue, +} + +impl BuzzSteerOperation { + pub fn new(steers: crate::steer::SteerQueue) -> Self { + Self { steers } + } +} + +#[async_trait] +impl Operation for BuzzSteerOperation { + fn name(&self) -> &'static str { + "buzz_steer" + } + + async fn run( + &self, + _session: &Session, + _conversation: &Conversation, + _emit: &Emitter, + ) -> Result { + let messages = self.steers.drain().await; + if messages.is_empty() { + return not_applicable(); + } + tracing::info!( + count = messages.len(), + "steer messages drained into the turn" + ); + applied(messages.into_iter().map(StateEffect::AppendMessage)) + } +} + +/// Compacts the conversation when it approaches the context limit. +/// +/// The mechanism is entirely goose's (`check_if_compaction_needed` / +/// `compact_messages`); this operation is the buzz-specific part around it: +/// the `_PostCompact` hook that re-injects buzz-dev-mcp's todo state, which is +/// what buzz-agent's old context-handoff existed to preserve. +/// +/// Not goose's `CompactionOperation`: that one yields to the client and emits +/// its own user-facing notification. In buzz a yield ends the turn and the +/// notification would post into the channel, so a long conversation would stop +/// mid-work and announce its own housekeeping to the humans watching. +pub struct BuzzCompactionOperation { + model: crate::model::SessionModel, + agent: Arc, + hook_extension: Option, + session_id: String, +} + +impl BuzzCompactionOperation { + pub fn new( + model: crate::model::SessionModel, + agent: Arc, + hook_extension: Option, + session_id: String, + ) -> Self { + Self { + model, + agent, + hook_extension, + session_id, + } + } +} + +#[async_trait] +impl Operation for BuzzCompactionOperation { + fn name(&self) -> &'static str { + "buzz_compaction" + } + + async fn run( + &self, + session: &Session, + conversation: &Conversation, + _emit: &Emitter, + ) -> Result { + let provider = self.model.provider().await; + if !goose::context_mgmt::check_if_compaction_needed( + provider.as_ref(), + conversation, + None, + session, + ) + .await? + { + return not_applicable(); + } + + let model_config = self.model.config().await; + let result = goose::context_mgmt::compact_messages( + provider.as_ref(), + &model_config, + &self.session_id, + conversation, + false, + ) + .await?; + + tracing::info!(target: "buzz_agent::compaction", "history compacted"); + + let mut effects = vec![StateEffect::ReplaceConversation { + conversation: result.conversation, + // `None`, deliberately: the running total described a conversation + // that no longer exists, so goose re-estimates from the compacted + // messages rather than carrying a stale count forward. + usage: None, + }]; + + if let Some(extension) = &self.hook_extension { + if let Some(reported) = + crate::hooks::post_compact_state(&self.agent, session, extension).await + { + // `[PostCompact]` prefix preserved from the inline version: + // it is how the model tells re-injected state apart from a + // user turn, and dropping it would change what it reads. + effects.push(StateEffect::AppendMessage( + Message::user() + .with_text(format!("[PostCompact] {reported}")) + .with_visibility(false, true), + )); + } + } + + applied(effects) + } +} + /// The operations that gate the start of a round. /// /// One entry today. As `PLANS/BUZZ_OPERATIONS_MIGRATION.md` proceeds, the @@ -439,6 +581,32 @@ pub fn round_gate( goose::agents::state_machine::StateMachine::new(steps, cancel) } +/// The operations that run at the start of a round, before inference. +/// +/// A separate machine from [`round_gate`] because these two sets run at +/// different points and must not be confused: `round_gate` also runs when a +/// turn wants to *end*, and draining steers or compacting there would change +/// when they happen. `StateMachine::step` stops at the first operation that +/// applies, so one combined machine could not express "do both". +pub fn round_start( + steers: crate::steer::SteerQueue, + compaction: BuzzCompactionOperation, + cancel: tokio_util::sync::CancellationToken, +) -> goose::agents::state_machine::StateMachine<'static> { + use goose::agents::state_machine::Step; + + // Steer before compaction: a steer that arrives just as the window fills + // should be part of what gets summarised, not appended to a conversation + // that was compacted a moment earlier without it. + goose::agents::state_machine::StateMachine::new( + vec![ + Step::Operation(Arc::new(BuzzSteerOperation::new(steers))), + Step::Operation(Arc::new(compaction)), + ], + cancel, + ) +} + #[cfg(test)] mod tests { use super::*;