mirror of
https://github.com/block/buzz.git
synced 2026-10-30 05:51:46 +00:00
9869f24319
Inverts the previous design. buzz-agent no longer calls
goose::agents::Agent::reply -- it drives its own round loop and calls
goose for the four heavy components:
model call Provider::stream (via Agent::provider)
tool surface Agent::list_tools
tool execution Agent::dispatch_tool_call
system prompt PromptManager (our own instance)
compaction context_mgmt::{check_if_compaction_needed, compact_messages}
Under Agent::reply every buzz-specific behaviour had to be smuggled in
around goose's turn policy: the _Stop veto needed an outer loop purely to
re-enter reply(), and [Reflect] had to be delivered as a *steer* because
the tool result itself was out of reach. Owning the loop removes the
smuggling -- the veto is a branch, and [Reflect] goes back on the tool
result where buzz-agent originally put it and where the model reads it in
context.
New modules:
loop_drive.rs the round loop: inference, tools, compaction, _Stop veto,
steer drain, cancellation, max-rounds bound
tools.rs parallel dispatch, announce->terminal wire invariant,
[Reflect] on failure
prompt.rs our PromptManager (goose's is pub(super) to Agent::reply)
steer.rs our steer queue (goose's drain is pub(crate) to reply)
agent.rs shrinks to what goose knows nothing about: ACP session/update
emission and the keepalive ticker. Tool lifecycle now emits from tools.rs,
which is the only place that knows when a call starts and ends -- emitting
from streamed content as well would double-announce every tool.
Behaviour preserved and covered by the existing suite: Fizz persona,
_Stop veto (cap 3), [Reflect] (cap 8), _PostCompact re-injection,
steering without cancellation, cancel leaving no unresolved tool call,
usage accounting, and the model catalog.
81 unit + 21 integration tests pass; clippy clean (the one hints.rs
warning predates this branch).
Co-authored-by: Michael Neale <michael.neale@gmail.com>
Signed-off-by: Michael Neale <michael.neale@gmail.com>
87 lines
2.8 KiB
Rust
87 lines
2.8 KiB
Rust
//! Mid-turn steering, owned by buzz-agent.
|
|
//!
|
|
//! Steering is a message injected into a turn already in flight, without
|
|
//! cancelling it. goose has an equivalent queue on its `Agent`, but
|
|
//! `drain_pending_steers` / `has_pending_steers` are `pub(crate)` — only
|
|
//! `Agent::reply` can consume them. Since buzz-agent owns the loop, it owns
|
|
//! the queue too.
|
|
//!
|
|
//! The semantics match what buzz-acp expects and what goose does internally:
|
|
//!
|
|
//! * the message is queued, never applied mid-inference — it lands at the
|
|
//! **round boundary**, so it cannot corrupt a partially streamed response;
|
|
//! * a pending steer **prevents the turn from ending**, so a user who steers
|
|
//! just as the model finishes is not silently ignored;
|
|
//! * steering never restarts or cancels the turn.
|
|
|
|
use std::collections::VecDeque;
|
|
use std::sync::Arc;
|
|
|
|
use goose_provider_types::conversation::message::Message;
|
|
use tokio::sync::Mutex;
|
|
|
|
/// A session's pending steer messages.
|
|
///
|
|
/// Cloneable and shared: `session/steer` pushes from the JSON-RPC dispatch
|
|
/// task while the turn drains from the loop task.
|
|
#[derive(Clone, Default)]
|
|
pub struct SteerQueue {
|
|
inner: Arc<Mutex<VecDeque<Message>>>,
|
|
}
|
|
|
|
impl SteerQueue {
|
|
pub fn new() -> Self {
|
|
Self::default()
|
|
}
|
|
|
|
pub async fn push(&self, message: Message) {
|
|
self.inner.lock().await.push_back(message);
|
|
}
|
|
|
|
pub async fn is_pending(&self) -> bool {
|
|
!self.inner.lock().await.is_empty()
|
|
}
|
|
|
|
/// Take everything queued so far.
|
|
///
|
|
/// Draining and appending is a single logical step for the loop: a steer
|
|
/// that arrives *during* the drain stays queued for the next boundary
|
|
/// rather than being lost.
|
|
pub async fn drain(&self) -> Vec<Message> {
|
|
self.inner.lock().await.drain(..).collect()
|
|
}
|
|
|
|
/// Drop anything queued. Used when a turn ends so a late steer cannot leak
|
|
/// into the next one.
|
|
pub async fn clear(&self) {
|
|
self.inner.lock().await.clear();
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[tokio::test]
|
|
async fn drain_takes_everything_in_order() {
|
|
let queue = SteerQueue::new();
|
|
queue.push(Message::user().with_text("first")).await;
|
|
queue.push(Message::user().with_text("second")).await;
|
|
assert!(queue.is_pending().await);
|
|
|
|
let drained = queue.drain().await;
|
|
assert_eq!(drained.len(), 2);
|
|
assert_eq!(drained[0].as_concat_text(), "first");
|
|
assert_eq!(drained[1].as_concat_text(), "second");
|
|
assert!(!queue.is_pending().await, "drain must empty the queue");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn clear_discards_a_late_steer() {
|
|
let queue = SteerQueue::new();
|
|
queue.push(Message::user().with_text("too late")).await;
|
|
queue.clear().await;
|
|
assert!(!queue.is_pending().await);
|
|
}
|
|
}
|