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 steer and compaction as goose Operations

Browse Source 
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 <michael.neale@gmail.com>
Co-authored-by: Michael Neale <michael.neale@gmail.com>
This commit is contained in:
Michael Neale
2026-08-12 15:32:05 +10:00
parent aef3921c1e
commit 099bacbe6a
2 changed files with 245 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Message>),
/// An operation applied; the loop continues with its effects in state
/// rather than ending.
Applied(Vec<goose::agents::state_machine::StateEffect>),
/// 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<goose::agents::state_machine::StateEffect>,
) -> 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 {

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

@@ -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<OperationResult> {
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<Agent>,
hook_extension: Option<String>,
session_id: String,
}
impl BuzzCompactionOperation {
pub fn new(
model: crate::model::SessionModel,
agent: Arc<Agent>,
hook_extension: Option<String>,
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<OperationResult> {
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::*;