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 <michael.neale@gmail.com>
Co-authored-by: Michael Neale <michael.neale@gmail.com>

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
+    // Every loop decision is a goose `Operation` -- see
-    // control flow. The others follow one at a time; see
+    // `PLANS/BUZZ_OPERATIONS_MIGRATION.md`. Two machines, because they run at
-    // `PLANS/BUZZ_OPERATIONS_MIGRATION.md` for the order and why.
+    // 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) => {
+            Gate::Applied(effects) => {
-                for message in messages {
+                apply_effects(&mut state, effects);
-                    state.push(message);
-                }
             }
             Gate::Open => {}
         }
 
-        // Steers land here, at the round boundary — never mid-inference,
+        // Steering and compaction, both at the round boundary and never
-        // where they would race a partially streamed response.
+        // mid-inference where they would race a partially streamed response.
-        for message in ctx.steers.drain().await {
+        // Run to exhaustion: `step` stops at the first operation that applies,
-            state.push(message);
+        // 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
-        // Compaction at the round boundary, never mid-stream.
+            else {
-        if let Err(e) = maybe_compact(&ctx, &mut state).await {
+                break;
-            tracing::warn!(error = %e, "compaction failed; continuing uncompacted");
+            };
+            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() => {
+                        Gate::Applied(effects) => {
-                            for message in messages {
+                            if apply_effects(&mut state, effects) {
-                                state.push(message);
+                                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
+                        Gate::Open => {}
-                        // has not given the model new work, so the turn still
-                        // ends -- otherwise the loop would spin.
-                        Gate::Applied(_) | 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
+    /// An operation applied; the loop continues with its effects in state
-    /// them in history rather than ending.
+    /// rather than ending.
-    Applied(Vec<Message>),
+    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(
+        Ok(Some(result)) => Gate::Applied(result.effects),
-            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
@@ -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 {

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<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::*;