mirror of
https://github.com/block/buzz.git
synced 2026-11-01 06:51:17 +00:00
fix(buzz-agent): address expert review — shutdown, cap race, usage model
An adversarial review of the swap found several real defects, most of them comments asserting parity that did not survive checking. Fixed: B3 serve() dropped in-flight work on stdin EOF. Dropping a CancellationToken does not cancel it, so goose never sent notifications/cancelled to its MCP children and they outlived us as orphans -- exactly the failure run_turn goes to lengths to avoid on session/cancel. The detached writer task was also never awaited, so frames still queued (including a response from a turn that finished on the same tick) were discarded. main did both; restored. B4 max_sessions was TOCTOU-racy. session/new is dispatched on its own task and build_agent (MCP spawn + provider round-trip) sits between the check and the insert, so N concurrent calls all passed. main re-checked under the insert guard; that guard had been dropped. Restored. B5 usage_update reported an empty model whenever the model came from GOOSE_MODEL rather than BUZZ_AGENT_MODEL -- a supported path everywhere else (build_agent and session/new both fall back to it). Blanked kind-44200 attribution silently. Now uses the same resolution chain. B6 BUZZ_AGENT_LLM_TIMEOUT_SECS was parsed, documented, and never read. Deleted rather than left as a knob that does nothing. R2 A whitespace-only steer returned success instead of INVALID_PARAMS. buzz-acp maps success to SteerAck::Ok and treats the message as delivered, so it was swallowed and the cancel+merge fallback suppressed. Now rejected up front, before touching the session map, as main did. R6 Restored #![forbid(unsafe_code)], lost in the rewrite. B1/B2 are documentation corrections, and they matter more than the code fixes: the module table claimed builtin.rs was replaced by goose's skills extension and hints.rs by goose's hint loader. Neither holds. Agent::with_config loads zero extensions and build_agent only adds the harness's declared mcpServers, so the skills extension is never loaded and load_skill/SKILL.md discovery are simply gone. Goose's hint loader keys off .goosehints while the old code walked for AGENTS.md -- every repo here ships the latter and none the former, so no hints load at all. Both now documented as losses instead of substitutions. 53 tests green, fmt + clippy -D warnings clean workspace-wide. Signed-off-by: Michael Neale <michael.neale@gmail.com>
This commit is contained in:
2
Cargo.lock
generated
2
Cargo.lock
generated
@@ -3621,9 +3621,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
|
||||
dependencies = [
|
||||
"cfg-if 1.0.4",
|
||||
"js-sys",
|
||||
"libc",
|
||||
"r-efi 5.3.0",
|
||||
"wasip2",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -8,11 +8,24 @@
|
||||
//! | `mcp.rs` | 1139 | `goose::agents::extension_manager` |
|
||||
//! | `auth.rs` | 845 | `goose::providers` (incl. Databricks OAuth) |
|
||||
//! | `agent.rs` | 746 | `goose::agents::Agent::reply` |
|
||||
//! | `hints.rs` | 726 | `goose`'s hint loader (`prompt_manager::with_hints`) |
|
||||
//! | `hints.rs` | 726 | *partly* -- see the caveat below |
|
||||
//! | `catalog.rs` | 631 | `goose::providers::init` model discovery |
|
||||
//! | `builtin.rs` | 575 | `goose`'s `skills` platform extension |
|
||||
//! | `builtin.rs` | 575 | **dropped** -- see the caveat below |
|
||||
//! | `handoff.rs` | 430 | `goose::context_mgmt` (auto-compaction) |
|
||||
//!
|
||||
//! Two entries above are NOT clean substitutions, and saying otherwise would
|
||||
//! be worse than the loss itself:
|
||||
//!
|
||||
//! * **`builtin.rs` / `load_skill` is gone.** `Agent::with_config` loads zero
|
||||
//! extensions and `build_agent` only adds the `mcpServers` the harness
|
||||
//! declares, so goose's `skills` platform extension is never loaded. The
|
||||
//! `load_skill` tool and `SKILL.md` discovery (old `hints.rs:110-204`) have
|
||||
//! no replacement. Restoring them means explicitly adding the extension.
|
||||
//! * **`AGENTS.md` hints are not loaded.** Goose's loader keys off
|
||||
//! `GOOSE_HINTS_FILENAME` (`.goosehints`); the old code walked the directory
|
||||
//! chain for `AGENTS.md` plus `~/AGENTS.md`. Every repo here ships the
|
||||
//! former and none ship the latter, so in practice no hints load at all.
|
||||
//!
|
||||
//! What this file keeps is the part Goose does *not* know about: the mapping
|
||||
//! from `AgentEvent` onto the exact `session/update` notifications `buzz-acp`
|
||||
//! consumes (`acp.rs:1528-1627`), and the `keepalive` ticker that resets the
|
||||
|
||||
@@ -37,8 +37,6 @@ pub struct Config {
|
||||
pub max_sessions: usize,
|
||||
/// Default system prompt, used only when `session/new` omits one.
|
||||
pub system_prompt: Option<String>,
|
||||
/// Per-turn wall-clock budget for a single provider request.
|
||||
pub llm_timeout_secs: u64,
|
||||
/// Tool-call approval policy.
|
||||
///
|
||||
/// `GooseMode::default()` is **`Auto`** (`goose_mode.rs:23-25`) — every
|
||||
@@ -145,7 +143,6 @@ impl Config {
|
||||
max_rounds: env_parse::<u32>("BUZZ_AGENT_MAX_ROUNDS").filter(|n| *n > 0),
|
||||
max_sessions: env_parse("BUZZ_AGENT_MAX_SESSIONS").unwrap_or(8),
|
||||
system_prompt,
|
||||
llm_timeout_secs: env_parse("BUZZ_AGENT_LLM_TIMEOUT_SECS").unwrap_or(600),
|
||||
goose_mode: parse_approval(env_str("BUZZ_AGENT_APPROVAL").as_deref()),
|
||||
// Discovery-only; the agent loop resolves providers through goose.
|
||||
provider: Provider::default(),
|
||||
@@ -164,7 +161,6 @@ impl Config {
|
||||
max_rounds: None,
|
||||
max_sessions: 1,
|
||||
system_prompt: None,
|
||||
llm_timeout_secs: 30,
|
||||
goose_mode: GooseMode::default(),
|
||||
provider,
|
||||
api_key,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
#![forbid(unsafe_code)]
|
||||
//! buzz-agent's ACP server, with Goose as the agent loop.
|
||||
//!
|
||||
//! This is the layer `buzz-acp` actually talks to, and it is deliberately
|
||||
@@ -225,7 +226,10 @@ fn hook_tool_guidance() -> String {
|
||||
|
||||
pub async fn serve(cfg: Config) -> anyhow::Result<()> {
|
||||
let (wire_tx, wire_rx) = tokio::sync::mpsc::channel(256);
|
||||
tokio::spawn(wire::writer_task(wire_rx));
|
||||
// Keep the join handle: on EOF we must await it so queued frames -- including
|
||||
// the final response of a turn that finished on the same tick -- actually
|
||||
// reach stdout before the runtime is dropped.
|
||||
let writer = tokio::spawn(wire::writer_task(wire_rx));
|
||||
|
||||
let app = Arc::new(App {
|
||||
cfg,
|
||||
@@ -261,6 +265,25 @@ pub async fn serve(cfg: Config) -> anyhow::Result<()> {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Orderly shutdown. Both halves matter and both were missing:
|
||||
//
|
||||
// 1. Cancel every in-flight turn. Dropping a CancellationToken does NOT
|
||||
// cancel it, so goose would never send `notifications/cancelled` to its
|
||||
// MCP children and they would outlive us as orphans -- exactly the
|
||||
// failure `agent::run_turn` goes to lengths to avoid on session/cancel.
|
||||
// 2. Await the writer. `wire_tx` is a bounded mpsc drained by a detached
|
||||
// task; returning here would discard anything still queued.
|
||||
{
|
||||
let sessions = app.sessions.lock().await;
|
||||
for session in sessions.values() {
|
||||
if let Some(token) = &session.cancel {
|
||||
token.cancel();
|
||||
}
|
||||
}
|
||||
}
|
||||
drop(wire_tx);
|
||||
let _ = writer.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -388,6 +411,9 @@ async fn session_new(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSen
|
||||
}
|
||||
}
|
||||
|
||||
// Cheap early reject. The authoritative check is re-done under the insert
|
||||
// guard below -- `session/new` is dispatched on its own task, so N
|
||||
// concurrent calls would otherwise all pass this and all insert.
|
||||
if app.sessions.lock().await.len() >= app.cfg.max_sessions {
|
||||
return wire::send(
|
||||
wire_tx,
|
||||
@@ -415,21 +441,34 @@ async fn session_new(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSen
|
||||
.or_else(|| std::env::var("GOOSE_MODEL").ok())
|
||||
.unwrap_or_default();
|
||||
|
||||
app.sessions.lock().await.insert(
|
||||
sid.clone(),
|
||||
Session {
|
||||
agent,
|
||||
goose_session_id,
|
||||
busy: false,
|
||||
active_run_id: None,
|
||||
cancel: None,
|
||||
model_override: None,
|
||||
pending_model: None,
|
||||
hook_extension,
|
||||
accumulated_input_tokens: 0,
|
||||
accumulated_output_tokens: 0,
|
||||
},
|
||||
);
|
||||
{
|
||||
let mut sessions = app.sessions.lock().await;
|
||||
// Re-check under the guard we insert with: build_agent above spawns MCP
|
||||
// children and does a provider round-trip, so other session/new tasks
|
||||
// can land in that window.
|
||||
if sessions.len() >= app.cfg.max_sessions {
|
||||
return wire::send(
|
||||
wire_tx,
|
||||
wire::err(id, INVALID_PARAMS, "session/new: max sessions reached"),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
sessions.insert(
|
||||
sid.clone(),
|
||||
Session {
|
||||
agent,
|
||||
goose_session_id,
|
||||
busy: false,
|
||||
active_run_id: None,
|
||||
cancel: None,
|
||||
model_override: None,
|
||||
pending_model: None,
|
||||
hook_extension,
|
||||
accumulated_input_tokens: 0,
|
||||
accumulated_output_tokens: 0,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
// Advertise the model catalog. `buzz-acp` reads `models.availableModels`
|
||||
// off this response (`acp.rs:1866`, `:1900`) to drive the desktop
|
||||
@@ -629,6 +668,10 @@ async fn session_prompt(app: &Arc<App>, id: Value, params: Value, wire_tx: &Wire
|
||||
.get(&p.session_id)
|
||||
.and_then(|s| s.model_override.clone())
|
||||
.or_else(|| app.cfg.model.clone())
|
||||
// A session starts fine with only GOOSE_MODEL set
|
||||
// (build_agent falls back to it), so omitting it here
|
||||
// emitted `"model": ""` and blanked kind-44200 attribution.
|
||||
.or_else(|| std::env::var("GOOSE_MODEL").ok())
|
||||
.unwrap_or_default()
|
||||
};
|
||||
wire::send(
|
||||
@@ -716,6 +759,19 @@ async fn steer(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSender) {
|
||||
}
|
||||
};
|
||||
|
||||
// Reject an empty steer before touching the session map, matching main.
|
||||
// buzz-acp maps a successful steer to SteerAck::Ok and considers the user's
|
||||
// message delivered, so acknowledging a no-op would swallow it silently and
|
||||
// suppress the cancel+merge fallback (`buzz-acp/src/pool.rs:329-366`).
|
||||
let text = agent::prompt_to_text(&p.prompt);
|
||||
if text.trim().is_empty() {
|
||||
return wire::send(
|
||||
wire_tx,
|
||||
wire::err(id, INVALID_PARAMS, "steer: prompt must not be empty"),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
let (agent, goose_session_id) = {
|
||||
let sessions = app.sessions.lock().await;
|
||||
let Some(s) = sessions.get(&p.session_id) else {
|
||||
@@ -745,13 +801,6 @@ async fn steer(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSender) {
|
||||
(s.agent.clone(), s.goose_session_id.clone())
|
||||
};
|
||||
|
||||
let text = agent::prompt_to_text(&p.prompt);
|
||||
if text.trim().is_empty() {
|
||||
// Best-effort by contract: a whitespace-only steer is dropped, never
|
||||
// fatal to the turn (buzz-agent `agent.rs:266-278`).
|
||||
return wire::send(wire_tx, wire::ok(id, json!({ "runId": p.expected_run_id }))).await;
|
||||
}
|
||||
|
||||
let message_id = format!("steer_{}", uuid_like());
|
||||
agent
|
||||
.steer(
|
||||
|
||||
Reference in New Issue
Block a user