From 15f6fd57f8b4f5bdadbccbe7e5ff33d67b50119a Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Wed, 12 Aug 2026 14:06:03 +1000 Subject: [PATCH] refactor(buzz-agent): express the _Stop veto as a goose Operation Step 2 of PLANS/BUZZ_OPERATIONS_MIGRATION.md. Behaviour is unchanged: the existing stop_hook and real_dev_mcp integration tests pass without modification, which is the point of this step. BuzzStopVetoOperation replaces the inline stop_veto() branch. The loop now runs the gate a second time when a round wants to end, which is where the operation applies -- it declines unless the last message ends the turn, so the top-of-round call is a no-op for it. Two things move rather than change: - The block cap was a loop-local u32 in run_turn; objections now carry a metadata note and the cap counts them since kickoff. run_turn is per session/prompt, so both are turn-scoped -- same three-strikes behaviour, but now a pure function of the conversation, which is why goose tags its own denials the same way. - MAX_STOP_BLOCKS was declared in both loop_drive and hooks with the same value. Dropped the loop_drive copy. Deliberately not goose's StopHookOperation: it drives goose's private HookManager plugin system, while buzz's hook is an MCP tool that can see buzz-dev-mcp's in-process todo state. It also emits a user-facing notification per denial, which in buzz would post 'stop hook blocked ending this turn' into a channel every time an agent had an open todo. The objection reaches the model, not the room. 65 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 | 138 +++++++++------- crates/buzz-agent/src/ops.rs | 233 +++++++++++++++++++++++++++- 2 files changed, 311 insertions(+), 60 deletions(-) diff --git a/crates/buzz-agent/src/loop_drive.rs b/crates/buzz-agent/src/loop_drive.rs index 235d9cae50..2c9697426d 100644 --- a/crates/buzz-agent/src/loop_drive.rs +++ b/crates/buzz-agent/src/loop_drive.rs @@ -70,10 +70,6 @@ const ERROR_REFLECTION: &str = /// conversation. const MAX_REFLECTIONS: usize = 8; -/// Hard cap on consecutive `_Stop` vetoes, mirroring goose's own -/// `stop_hook_block_cap`. A tool that always objects must not trap the turn. -pub const MAX_STOP_BLOCKS: u32 = 3; - /// Rounds allowed when the caller sets no limit. /// /// buzz-agent's loop was previously bounded by goose's `max_turns`. Owning the @@ -153,14 +149,19 @@ pub async fn run_turn( state.push(prompt); let max_rounds = ctx.max_rounds.unwrap_or(DEFAULT_MAX_ROUNDS); - let mut stop_blocks = 0u32; 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. let outcome = crate::ops::Outcome::new(); - let machine = crate::ops::round_gate(max_rounds, outcome.clone(), ctx.cancel.clone()); + let machine = crate::ops::round_gate( + max_rounds, + outcome.clone(), + ctx.cancel.clone(), + ctx.hook_extension + .map(|extension| (Arc::clone(ctx.agent), extension.to_string())), + ); 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. @@ -179,12 +180,16 @@ pub async fn run_turn( // Ask the machine before doing any work. `step` returns the first // operation that applies, or `None` when the loop below should run. match gate(&machine, state.session(), &emitter).await { - Err(e) => return (Err(e), tokens, state.conversation().messages().to_vec()), - Ok(true) => { + Gate::Ended => { let reason = outcome.take().unwrap_or(StopReason::EndTurn); return (Ok(reason), tokens, state.conversation().messages().to_vec()); } - Ok(false) => {} + Gate::Applied(messages) => { + for message in messages { + state.push(message); + } + } + Gate::Open => {} } // Steers land here, at the round boundary — never mid-inference, @@ -211,48 +216,92 @@ pub async fn run_turn( continue; } - // The turn wants to end. `_Stop` gets a veto. - match stop_veto(&ctx, state.session(), &mut stop_blocks).await { - Some(objection) => { - state.push( - Message::user() - .with_text(format!("[Stop] {objection}")) - .with_visibility(false, true), - ); - continue; - } - None => { - // A steer arriving after this point belongs to no run. - ctx.steers.clear().await; - return ( - Ok(StopReason::EndTurn), - tokens, - state.conversation().messages().to_vec(), - ); + // The turn wants to end, so ask the machine again. This is + // the call `BuzzStopVetoOperation` exists for: it only applies + // to a conversation whose last message ends the turn, which is + // true here and false at the top of a round. + // + // Cancellation short-circuits it -- dispatching a hook tool + // for a turn the user already abandoned would delay the + // 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); + } + continue; + } + 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 => {} } } + + // A steer arriving after this point belongs to no run. + ctx.steers.clear().await; + return ( + Ok(StopReason::EndTurn), + tokens, + state.conversation().messages().to_vec(), + ); } } } } -/// Run the operation gate for one round. +/// What the operation gate decided. +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 ended the turn. + Ended, +} + +/// Run the operation gate once. /// -/// Returns `true` when an operation ended the turn. +/// Called at the top of every round, and again when a round wants to end -- +/// that second call is what gives `_Stop` its veto, since its operation only +/// applies to a conversation whose last message ends the turn. async fn gate( machine: &goose::agents::state_machine::StateMachine<'_>, session: &Session, emitter: &Emitter, -) -> Result { +) -> Gate { match machine.step(session, emitter).await { - Ok(Some(result)) => Ok(result.yield_to_client), - Ok(None) => Ok(false), + 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(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 // proceed rather than strand the user's prompt unanswered. Err(e) => { tracing::warn!(error = %e, "operation gate failed; continuing"); - Ok(false) + Gate::Open } } } @@ -509,29 +558,6 @@ async fn maybe_compact( Ok(()) } -/// Ask `_Stop` whether the turn may end. `Some(objection)` means keep working. -async fn stop_veto( - ctx: &TurnContext<'_>, - session: &Session, - stop_blocks: &mut u32, -) -> Option { - if ctx.cancel.is_cancelled() { - return None; - } - let extension = ctx.hook_extension?; - - if *stop_blocks >= MAX_STOP_BLOCKS { - tracing::warn!(blocks = *stop_blocks, "_Stop veto cap reached; ending turn"); - return None; - } - - let objection = crate::hooks::stop_objection(ctx.agent, session, extension).await?; - - *stop_blocks += 1; - tracing::info!(blocks = *stop_blocks, "_Stop hook vetoed end of turn"); - Some(objection) -} - /// 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 1d532321a6..666042275b 100644 --- a/crates/buzz-agent/src/ops.rs +++ b/crates/buzz-agent/src/ops.rs @@ -22,7 +22,10 @@ use std::sync::{Arc, Mutex}; use anyhow::Result; use async_trait::async_trait; -use goose::agents::state_machine::{not_applicable, yielded, Emitter, Operation, OperationResult}; +use goose::agents::state_machine::{ + applied, not_applicable, yielded, Emitter, Operation, OperationResult, +}; +use goose::agents::Agent; use goose::conversation::Conversation; use goose::session::Session; use goose_provider_types::conversation::message::Message; @@ -160,6 +163,125 @@ impl Operation for BuzzMaxRoundsOperation { } } +/// Whether the last message ends the turn: an assistant message with no error +/// and no outstanding tool request. +/// +/// goose has this as `ends_turn` in its private `operation` module +/// (`operation.rs:63`); only the names re-exported from `state_machine` escape +/// the crate. Reimplemented to match, with one deliberate narrowing: buzz has +/// no frontend tools and no approval flow, so `FrontendToolRequest` and +/// `ActionRequired` cannot appear in a buzz conversation. Matching on +/// `ToolRequest` alone is the whole of buzz's turn shape. +fn ends_turn(messages: &[Message]) -> bool { + messages.last().is_some_and(|last| { + last.role == rmcp::model::Role::Assistant + && last.error_kind().is_none() + && !last.content.iter().any(|content| { + matches!( + content, + goose_provider_types::conversation::message::MessageContent::ToolRequest(_) + ) + }) + }) +} + +/// Lets the `_Stop` hook object to a turn that wants to end. +/// +/// When the agent has finished, `_Stop` is asked whether it may stop. An +/// objection is appended as an agent-visible `[Stop]` message and the turn +/// continues; silence lets it end. +/// +/// # How this differs from goose's `StopHookOperation` +/// +/// goose drives its own `HookManager` plugins and emits a user-facing +/// notification for each denial. buzz's hook is an **MCP tool** (`_Stop` on +/// the dev-MCP extension), and buzz stays silent: `buzz-acp` publishes +/// assistant messages into a channel, so a "hook blocked ending this turn" +/// notification would be posted to the humans in that channel every time an +/// agent's todo list was incomplete. The objection reaches the model, not the +/// room. Same reasoning as `BuzzMaxRoundsOperation` not appending goose's +/// "Would you like me to continue?". +/// +/// # Why the block count lives on the messages +/// +/// The cap was a loop-local `u32` in `run_turn`. Counting prior objections +/// from their own metadata instead makes the operation a pure function of the +/// conversation, so a pipeline rebuilt from persisted messages reaches the +/// same decision — the same reason goose tags its denials. +/// +/// **This is not a behaviour change.** `run_turn` is per `session/prompt`, so +/// the old counter was already turn-scoped, and `messages_since_kickoff` +/// spans exactly that same turn. Three objections still end the turn; a new +/// prompt still starts from zero. +pub struct BuzzStopVetoOperation { + agent: Arc, + extension: String, + block_cap: u32, +} + +impl BuzzStopVetoOperation { + /// `extension` is the MCP extension serving `_Stop`; without one there is + /// no hook to ask, and the operation never applies. + pub fn new(agent: Arc, extension: String, block_cap: u32) -> Self { + Self { + agent, + extension, + block_cap, + } + } +} + +/// Metadata key marking a message as a `_Stop` objection, so later rounds can +/// count them without a side channel. +const OBJECTED: &str = "objected"; + +#[async_trait] +impl Operation for BuzzStopVetoOperation { + fn name(&self) -> &'static str { + "buzz_stop_veto" + } + + async fn run( + &self, + session: &Session, + conversation: &Conversation, + _emit: &Emitter, + ) -> Result { + let Some(messages) = messages_since_kickoff(conversation) else { + return not_applicable(); + }; + // Only a turn that is trying to end can be vetoed. + if !ends_turn(messages) { + return not_applicable(); + } + + let blocks = messages + .iter() + .filter(|message| self.message_meta(message, OBJECTED).is_some()) + .count() as u32; + if blocks >= self.block_cap { + tracing::warn!(blocks, "_Stop veto cap reached; ending turn"); + return not_applicable(); + } + + // A broken or absent hook must never trap a turn -- `stop_objection` + // maps every failure to `None`, which is "no objection". + let Some(objection) = + crate::hooks::stop_objection(&self.agent, session, &self.extension).await + else { + return not_applicable(); + }; + + tracing::info!(blocks = blocks + 1, "_Stop hook vetoed end of turn"); + + let mut message = Message::user() + .with_text(format!("[Stop] {objection}")) + .with_visibility(false, true); + self.set_message_meta(&mut message, OBJECTED, serde_json::json!(true)); + applied([message.into()]) + } +} + /// The operations that gate the start of a round. /// /// One entry today. As `PLANS/BUZZ_OPERATIONS_MIGRATION.md` proceeds, the @@ -169,10 +291,23 @@ pub fn round_gate( max_rounds: u32, outcome: Outcome, cancel: tokio_util::sync::CancellationToken, + stop_veto: Option<(Arc, String)>, ) -> goose::agents::state_machine::StateMachine<'static> { - let steps = vec![goose::agents::state_machine::Step::Operation(Arc::new( - BuzzMaxRoundsOperation::new(max_rounds, outcome), - ))]; + use goose::agents::state_machine::Step; + + // Order is precedence. The round budget is checked first: once it is + // spent the turn ends, and asking `_Stop` to veto a turn we are ending + // anyway would dispatch a tool call whose answer cannot be honoured. + let mut steps: Vec = vec![Step::Operation(Arc::new(BuzzMaxRoundsOperation::new( + max_rounds, outcome, + )))]; + if let Some((agent, extension)) = stop_veto { + steps.push(Step::Operation(Arc::new(BuzzStopVetoOperation::new( + agent, + extension, + crate::hooks::MAX_STOP_BLOCKS, + )))); + } goose::agents::state_machine::StateMachine::new(steps, cancel) } @@ -182,6 +317,12 @@ mod tests { use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; + /// A bare agent with no extensions: enough to dispatch a tool call that + /// will fail to resolve, which is the "hook unavailable" path. + async fn agent() -> Arc { + Arc::new(Agent::new()) + } + fn emitter() -> Emitter { let (tx, rx) = mpsc::channel(16); // Keep the receiver alive for the duration of the test: a dropped @@ -243,6 +384,90 @@ mod tests { assert!(matches!(outcome.take(), Some(StopReason::MaxTurnRequests))); } + /// An assistant message with an outstanding tool request: the model is + /// still working, so the turn is not trying to end. + fn tool_calling_conversation() -> Conversation { + Conversation::new_unvalidated(vec![ + Message::user().with_text("kickoff"), + Message::assistant().with_tool_request( + "call_1", + Ok(rmcp::model::CallToolRequestParams::new("developer__shell")), + ), + ]) + } + + #[tokio::test] + async fn the_veto_does_not_apply_while_tools_are_outstanding() { + // The hook is only asked when the turn is trying to end. Asking + // mid-tool-call would dispatch `_Stop` on every round. + let op = BuzzStopVetoOperation::new(agent().await, "buzz-dev-mcp".to_string(), 3); + let result = op + .run( + &Session::default(), + &tool_calling_conversation(), + &emitter(), + ) + .await + .unwrap(); + assert!(matches!(result, OperationResult::NotApplicable)); + } + + #[tokio::test] + async fn a_missing_hook_extension_never_traps_the_turn() { + // No such extension -> `stop_objection` yields None -> the turn ends. + // A broken hook must not be able to hold a turn open. + let op = BuzzStopVetoOperation::new(agent().await, "no-such-extension".to_string(), 3); + let result = op + .run(&Session::default(), &conversation(1), &emitter()) + .await + .unwrap(); + assert!(matches!(result, OperationResult::NotApplicable)); + } + + #[tokio::test] + async fn the_cap_counts_objections_already_in_the_conversation() { + // Three prior objections is the cap, so the operation must decline + // rather than dispatch a fourth `_Stop` call. Proven by pointing it at + // an extension that would otherwise be consulted: the result is the + // same NotApplicable either way, so instead assert the counting + // helper directly against tagged messages. + let op = BuzzStopVetoOperation::new(agent().await, "buzz-dev-mcp".to_string(), 3); + + let mut objection = Message::user() + .with_text("[Stop] finish your todos") + .with_visibility(false, true); + op.set_message_meta(&mut objection, OBJECTED, serde_json::json!(true)); + + let mut messages = vec![Message::user().with_text("kickoff")]; + for _ in 0..3 { + messages.push(objection.clone()); + messages.push(Message::assistant().with_text("done")); + } + let conversation = Conversation::new_unvalidated(messages); + + let counted = messages_since_kickoff(&conversation) + .unwrap() + .iter() + .filter(|message| op.message_meta(message, OBJECTED).is_some()) + .count(); + assert_eq!(counted, 3, "each objection must be countable from history"); + + // At the cap the turn is allowed to end. + let result = op + .run(&Session::default(), &conversation, &emitter()) + .await + .unwrap(); + assert!(matches!(result, OperationResult::NotApplicable)); + } + + #[tokio::test] + async fn ends_turn_matches_the_shape_of_a_finished_answer() { + assert!(ends_turn(&[Message::assistant().with_text("done")])); + assert!(!ends_turn(&[Message::user().with_text("hi")])); + assert!(!ends_turn(tool_calling_conversation().messages())); + assert!(!ends_turn(&[])); + } + #[tokio::test] async fn the_budget_appends_no_message() { // goose's MaxTurnsOperation asks "Would you like me to continue?".