jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-29 13:31:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(buzz-agent): express the _Stop veto as a goose Operation

Browse Source 
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 <michael.neale@gmail.com>
Co-authored-by: Michael Neale <michael.neale@gmail.com>
This commit is contained in:
Michael Neale
2026-08-12 14:06:03 +10:00
parent d4db850b3a
commit 15f6fd57f8
2 changed files with 311 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
crates/buzz-agent/src/loop_drive.rs
View File

@@ -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<Message>),
/// 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<bool, AgentError> {
) -> 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<String> {
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 {

233
crates/buzz-agent/src/ops.rs
View File

@@ -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<Agent>,
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<Agent>, 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<OperationResult> {
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<Agent>, 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<Step> = 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<Agent> {
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?".