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

refactor(buzz-agent): keep the turn conversation in memory, not sqlite

Browse Source 
The turn loop round-tripped every message through goose's SessionManager
-- a sqlite database -- once per round. That was never required. None of
the goose APIs buzz calls reads the store:

- Provider::stream takes &[Message].
- Agent::dispatch_tool_call takes &Session but reads only working_dir
  and id.
- context_mgmt::{check_if_compaction_needed, compact_messages} read
  usage.total_tokens and model_config, and carry session_id only into a
  task-local that becomes an agent-session-id HTTP header for provider
  request attribution (goose/src/session_context.rs).

So the turn now owns a Session value directly (crate::turn_state) and
buzz-agent keeps the conversation across turns itself. goose still gets
the &Session its signatures ask for; nothing backs it with a database.

This is the right shape for buzz regardless of dependencies: a Buzz
agent's durable record is the relay, not a local sqlite file. Agents are
chatty and long-lived, and every one of them was growing a database
nothing ever read back.

One row remains, created at session/new: update_provider persists the
provider/model config against the session id, and that config is what
goose reads back to resolve the context limit for compaction. Declining
it would make auto-compaction fire at the wrong point. Zero message rows
are written for a conversation of any length.

Verified: a two-turn conversation recalls the first turn's content with
0 rows in the messages table; handtest.py --all 18/18; 84 unit + 21
integration green.

Also fixes the Desktop Core CI failure this branch introduced: the log
filter added earlier pushed runtime.rs past the 1000-line ratchet, so it
moves to its own module with its tests.

Co-authored-by: Michael Neale <michael.neale@gmail.com>
Signed-off-by: Michael Neale <michael.neale@gmail.com>
This commit is contained in:
Michael Neale
2026-08-11 20:25:46 +10:00
parent 979eff99fd
commit ce844719f8
7 changed files with 411 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -1173,6 +1173,7 @@ dependencies = [
"getrandom 0.4.3",
"goose",
"goose-provider-types",
"goose-providers",
"hex",
"idna_adapter",
"nix 0.31.3",

3
crates/buzz-agent/Cargo.toml
View File

@@ -31,6 +31,9 @@ path = "tests/bin/fake_mcp.rs"
buzz-model-catalog = { path = "../buzz-model-catalog" }
goose = { git = "https://github.com/aaif-goose/goose", rev = "bf332b983756eaf80353e82e4bb052d589629bf4", default-features = false, features = ["rustls-tls"] }
goose-provider-types = { git = "https://github.com/aaif-goose/goose", rev = "bf332b983756eaf80353e82e4bb052d589629bf4" }
# ModelConfig: goose re-exports it from goose-providers; goose::model_config
# keeps its own copy private.
goose-providers = { git = "https://github.com/aaif-goose/goose", rev = "bf332b983756eaf80353e82e4bb052d589629bf4" }
# Resolution guard: goose pins icu_locale "=2.1.1" (needs icu_collections
# ~2.1.1) while url 2.5.x -> idna -> idna_adapter 1.2.2 pulls icu_normalizer
# 2.2.0 (needs icu_collections ~2.2.0). Only one 2.x icu_collections can be

46
crates/buzz-agent/src/lib.rs
View File

@@ -33,6 +33,7 @@ pub mod prompt;
pub mod session_store;
pub mod steer;
pub mod tools;
pub mod turn_state;
pub mod types;
pub mod wire;
@@ -64,8 +65,19 @@ use wire::{
/// One live ACP session, wrapping a Goose `Agent`.
struct Session {
agent: Arc<Agent>,
/// Goose's own session id (the SQLite-backed conversation).
/// Session id, shared with goose so tool dispatch and provider request
/// attribution agree on a name. No database sits behind it: the
/// conversation lives in `history` and in the turn's own
/// [`crate::turn_state::TurnState`].
goose_session_id: String,
/// The conversation so far, across turns.
///
/// buzz-agent owns this because a Buzz agent's durable record is the
/// relay, not a local store. Held here so a second `session/prompt` sees
/// what the first one said.
history: Vec<goose_provider_types::conversation::message::Message>,
/// Working directory for the session, as given at `session/new`.
working_dir: std::path::PathBuf,
busy: bool,
/// Set for the duration of a turn; advertised to steer-capable clients.
active_run_id: Option<String>,
@@ -136,6 +148,15 @@ async fn build_agent(
GoosePlatform::GooseCli,
));
// A session id, not a database row. `Agent::update_provider` and
// `model_config_for_session` want an id to key on, but the conversation
// itself never goes to goose's store -- see `crate::turn_state`.
//
// The row still gets created because `update_provider` persists the
// provider/model config against it, and that is the one write we cannot
// decline without losing the model config goose reads back for the
// context limit. Nothing else is written: a full conversation adds zero
// message rows.
let session = session_manager
.create_session(
std::path::PathBuf::from(cwd),
@@ -530,6 +551,8 @@ async fn session_new(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSen
Session {
agent,
goose_session_id,
history: Vec::new(),
working_dir: std::path::PathBuf::from(&p.cwd),
busy: false,
active_run_id: None,
cancel: None,
@@ -634,7 +657,16 @@ async fn session_prompt(app: &Arc<App>, id: Value, params: Value, wire_tx: &Wire
let cancel = CancellationToken::new();
// Single-flight per session, and capture the agent handle.
let (agent, goose_session_id, pending_model, hook_extension, prompt, steers) = {
let (
agent,
goose_session_id,
pending_model,
hook_extension,
prompt,
steers,
history,
working_dir,
) = {
let mut sessions = app.sessions.lock().await;
let Some(s) = sessions.get_mut(&p.session_id) else {
return wire::send(
@@ -666,6 +698,8 @@ async fn session_prompt(app: &Arc<App>, id: Value, params: Value, wire_tx: &Wire
s.hook_extension.clone(),
s.prompt.clone(),
s.steers.clone(),
s.history.clone(),
s.working_dir.clone(),
)
};
@@ -696,7 +730,7 @@ async fn session_prompt(app: &Arc<App>, id: Value, params: Value, wire_tx: &Wire
)
.await;
let (result, tokens) = loop_drive::run_turn(
let (result, tokens, conversation) = loop_drive::run_turn(
loop_drive::TurnContext {
agent: &agent,
session_id: &goose_session_id,
@@ -706,6 +740,8 @@ async fn session_prompt(app: &Arc<App>, id: Value, params: Value, wire_tx: &Wire
max_rounds: app.cfg.max_rounds,
prompt: &prompt,
steers: &steers,
working_dir,
history: &history,
},
goose_provider_types::conversation::message::Message::user()
.with_text(agent::prompt_to_text(&p.prompt)),
@@ -716,6 +752,10 @@ async fn session_prompt(app: &Arc<App>, id: Value, params: Value, wire_tx: &Wire
let accumulated = {
let mut sessions = app.sessions.lock().await;
sessions.get_mut(&p.session_id).map(|s| {
// Carry the turn's conversation forward. Without this the next
// prompt starts from an empty history and the agent forgets what
// it just said.
s.history = conversation;
s.busy = false;
s.active_run_id = None;
s.cancel = None;

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

@@ -14,7 +14,7 @@
//! | tool execution | `Agent::dispatch_tool_call` |
//! | system prompt | `Agent::build_turn_system_prompt` → `PromptManager` |
//! | compaction | `goose::context_mgmt::{check_if_compaction_needed, compact_messages}` |
//! | conversation store | `SessionManager` (sqlite), read/written by us |
//! | conversation store | buzz's own [`crate::turn_state::TurnState`], in memory |
//!
//! Everything that decides *turn shape* stays here: round structure, the
//! `_Stop` veto, `[Reflect]` on tool failure, `[PostCompact]` re-injection,
@@ -47,7 +47,7 @@ use futures::StreamExt;
use tokio_util::sync::CancellationToken;
use goose::agents::Agent;
use goose::session::session_manager::{Session, SessionManager};
use goose::session::session_manager::Session;
use goose_provider_types::conversation::message::{Message, MessageContent, ToolRequest};
use goose_provider_types::conversation::Conversation;
@@ -105,25 +105,48 @@ pub struct TurnContext<'a> {
/// Messages injected mid-turn by `session/steer`. Drained at the round
/// boundary; a pending steer also blocks the end of the turn.
pub steers: &'a crate::steer::SteerQueue,
/// The session's working directory, for tool dispatch and hint tracking.
pub working_dir: std::path::PathBuf,
/// Conversation carried over from earlier turns in this session.
///
/// buzz-agent holds this across turns itself now that the turn's
/// conversation is in memory rather than in goose's database.
pub history: &'a [Message],
}
/// Drive one `session/prompt` turn to completion.
///
/// Returns the ACP stop reason plus the turn's token counts. The caller emits
/// `usage_update` *before* the `session/prompt` response — that ordering is
/// load-bearing for kind-44200 metrics.
/// Returns the ACP stop reason, the turn's token counts, and the conversation
/// as it stands at the end of the turn — the caller keeps that for the next
/// prompt, since no database does it for us. The caller emits `usage_update`
/// *before* the `session/prompt` response: that ordering is load-bearing for
/// kind-44200 metrics.
pub async fn run_turn(
ctx: TurnContext<'_>,
prompt: Message,
) -> (Result<StopReason, AgentError>, super::agent::TurnTokens) {
) -> (
Result<StopReason, AgentError>,
super::agent::TurnTokens,
Vec<Message>,
) {
let mut tokens = super::agent::TurnTokens::default();
let sessions = crate::session_store::session_manager();
// The user's prompt is persisted before the first inference so a resumed
// or inspected session sees the same history the model saw.
if let Err(e) = sessions.add_message(ctx.session_id, &prompt).await {
return (Err(AgentError::Llm(format!("persist prompt: {e}"))), tokens);
// The turn's conversation lives here, not in a database. See
// `crate::turn_state` for why goose never needed one.
let model_config = ctx
.agent
.model_config_for_session(ctx.session_id)
.await
.ok();
let mut state = crate::turn_state::TurnState::new(
ctx.session_id.to_string(),
ctx.working_dir.clone(),
model_config,
);
for message in ctx.history.iter().cloned() {
state.push(message);
}
state.push(prompt);
let max_rounds = ctx.max_rounds.unwrap_or(DEFAULT_MAX_ROUNDS);
let mut stop_blocks = 0u32;
@@ -142,16 +165,20 @@ pub async fn run_turn(
loop {
if ctx.cancel.is_cancelled() {
return (Ok(StopReason::Cancelled), tokens);
return (
Ok(StopReason::Cancelled),
tokens,
state.conversation().messages().to_vec(),
);
}
// 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, &sessions, ctx.session_id, &emitter).await {
Err(e) => return (Err(e), tokens),
match gate(&machine, state.session(), &emitter).await {
Err(e) => return (Err(e), tokens, state.conversation().messages().to_vec()),
Ok(true) => {
let reason = outcome.take().unwrap_or(StopReason::EndTurn);
return (Ok(reason), tokens);
return (Ok(reason), tokens, state.conversation().messages().to_vec());
}
Ok(false) => {}
}
@@ -159,20 +186,20 @@ pub async fn run_turn(
// 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 {
if let Err(e) = sessions.add_message(ctx.session_id, &message).await {
return (Err(AgentError::Llm(format!("persist steer: {e}"))), tokens);
}
state.push(message);
}
// Compaction at the round boundary, never mid-stream.
if let Err(e) = maybe_compact(&ctx, &sessions).await {
if let Err(e) = maybe_compact(&ctx, &mut state).await {
tracing::warn!(error = %e, "compaction failed; continuing uncompacted");
}
match round(&ctx, &sessions, &mut tokens, &mut reflections).await {
Err(e) => return (Err(e), tokens),
match round(&ctx, &mut state, &mut tokens, &mut reflections).await {
Err(e) => return (Err(e), tokens, state.conversation().messages().to_vec()),
Ok(Round::Continued) => continue,
Ok(Round::Stopped(reason)) => return (Ok(reason), tokens),
Ok(Round::Stopped(reason)) => {
return (Ok(reason), tokens, state.conversation().messages().to_vec())
}
Ok(Round::WantsToEnd) => {
// A steer that arrived while the model was finishing must be
// answered, not dropped — so it also blocks the end of turn.
@@ -181,20 +208,23 @@ pub async fn run_turn(
}
// The turn wants to end. `_Stop` gets a veto.
match stop_veto(&ctx, &sessions, &mut stop_blocks).await {
match stop_veto(&ctx, state.session(), &mut stop_blocks).await {
Some(objection) => {
let message = Message::user()
.with_text(format!("[Stop] {objection}"))
.with_visibility(false, true);
if let Err(e) = sessions.add_message(ctx.session_id, &message).await {
return (Err(AgentError::Llm(format!("persist stop: {e}"))), tokens);
}
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);
return (
Ok(StopReason::EndTurn),
tokens,
state.conversation().messages().to_vec(),
);
}
}
}
@@ -204,17 +234,13 @@ pub async fn run_turn(
/// Run the operation gate for one round.
///
/// Returns `true` when an operation ended the turn. The session is reloaded
/// each time because operations read the conversation, and the previous round
/// appended to it.
/// Returns `true` when an operation ended the turn.
async fn gate(
machine: &goose::agents::state_machine::StateMachine<'_>,
sessions: &SessionManager,
session_id: &str,
session: &Session,
emitter: &Emitter,
) -> Result<bool, AgentError> {
let session = load_session(sessions, session_id).await?;
match machine.step(&session, emitter).await {
match machine.step(session, emitter).await {
Ok(Some(result)) => Ok(result.yield_to_client),
Ok(None) => Ok(false),
// A failing gate must not take the turn down with it: the operations
@@ -230,23 +256,25 @@ async fn gate(
/// One inference plus, if the model asked for them, one batch of tool calls.
async fn round(
ctx: &TurnContext<'_>,
sessions: &SessionManager,
state: &mut crate::turn_state::TurnState,
tokens: &mut super::agent::TurnTokens,
reflections: &mut usize,
) -> Result<Round, AgentError> {
let session = load_session(sessions, ctx.session_id).await?;
let conversation = session.conversation.clone().unwrap_or_default();
let conversation = state.conversation();
let assistant = match infer(ctx, &session, &conversation, tokens).await? {
let assistant = match infer(ctx, state.session(), &conversation, tokens).await? {
Some(message) => message,
// A provider that returns nothing is not a turn we can continue.
None => return Ok(Round::Stopped(StopReason::EndTurn)),
};
sessions
.add_message(ctx.session_id, &assistant)
.await
.map_err(|e| AgentError::Llm(format!("persist assistant: {e}")))?;
state.push(assistant.clone());
// Keep the running total current: goose's compaction check prefers it
// over re-estimating the conversation.
// i32 because that is the width goose's `Usage` uses; a total beyond
// i32::MAX means something has gone very wrong upstream, and saturating is
// better than wrapping into a negative that reads as "no usage".
state.set_total_tokens(tokens.total.map(|t| i32::try_from(t).unwrap_or(i32::MAX)));
let requests: Vec<ToolRequest> = assistant
.content
@@ -269,7 +297,7 @@ async fn round(
crate::agent::emit_tool_call_update(ctx.wire_tx, ctx.session_id, &request.id, true)
.await;
}
persist_tool_results(ctx, sessions, results).await?;
push_tool_results(state, results);
return Ok(Round::Stopped(StopReason::Cancelled));
}
@@ -278,21 +306,21 @@ async fn round(
for request in &requests {
if let Ok(call) = &request.tool_call {
ctx.prompt
.record_tool_arguments(&call.arguments, &session.working_dir)
.record_tool_arguments(&call.arguments, &state.session().working_dir)
.await;
}
}
let results = crate::tools::execute(
ctx.agent,
&session,
state.session(),
ctx.wire_tx,
ctx.cancel,
&requests,
reflections,
)
.await;
persist_tool_results(ctx, sessions, results).await?;
push_tool_results(state, results);
Ok(Round::Continued)
}
@@ -413,43 +441,39 @@ fn accumulate_usage(
}
}
/// Persist tool results as a single user message, matching the shape every
/// Append tool results as a single user message, matching the shape every
/// provider wire format expects (one result per outstanding request).
async fn persist_tool_results(
ctx: &TurnContext<'_>,
sessions: &SessionManager,
fn push_tool_results(
state: &mut crate::turn_state::TurnState,
results: Vec<(
String,
goose_provider_types::conversation::message::ToolResult<rmcp::model::CallToolResult>,
)>,
) -> Result<(), AgentError> {
) {
if results.is_empty() {
return Ok(());
return;
}
let mut message = Message::user();
for (id, result) in results {
message = message.with_tool_response(id, result);
}
sessions
.add_message(ctx.session_id, &message)
.await
.map_err(|e| AgentError::Llm(format!("persist tool results: {e}")))
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<'_>, sessions: &SessionManager) -> anyhow::Result<()> {
let session = sessions.get_session(ctx.session_id, true).await?;
let Some(conversation) = session.conversation.clone() else {
return Ok(());
};
async fn maybe_compact(
ctx: &TurnContext<'_>,
state: &mut crate::turn_state::TurnState,
) -> anyhow::Result<()> {
let conversation = state.conversation();
let provider = ctx.agent.provider().await?;
if !goose::context_mgmt::check_if_compaction_needed(
provider.as_ref(),
&conversation,
None,
&session,
state.session(),
)
.await?
{
@@ -466,20 +490,23 @@ async fn maybe_compact(ctx: &TurnContext<'_>, sessions: &SessionManager) -> anyh
)
.await?;
sessions
.replace_conversation(ctx.session_id, &result.conversation)
.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 {
let session = sessions.get_session(ctx.session_id, true).await?;
if let Some(state) = crate::hooks::post_compact_state(ctx.agent, &session, extension).await
if let Some(reported) =
crate::hooks::post_compact_state(ctx.agent, state.session(), extension).await
{
let message = Message::user()
.with_text(format!("[PostCompact] {state}"))
.with_visibility(false, true);
sessions.add_message(ctx.session_id, &message).await?;
state.push(
Message::user()
.with_text(format!("[PostCompact] {reported}"))
.with_visibility(false, true),
);
}
}
@@ -489,7 +516,7 @@ async fn maybe_compact(ctx: &TurnContext<'_>, sessions: &SessionManager) -> anyh
/// Ask `_Stop` whether the turn may end. `Some(objection)` means keep working.
async fn stop_veto(
ctx: &TurnContext<'_>,
sessions: &SessionManager,
session: &Session,
stop_blocks: &mut u32,
) -> Option<String> {
if ctx.cancel.is_cancelled() {
@@ -502,21 +529,13 @@ async fn stop_veto(
return None;
}
let session = sessions.get_session(ctx.session_id, false).await.ok()?;
let objection = crate::hooks::stop_objection(ctx.agent, &session, extension).await?;
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)
}
async fn load_session(sessions: &SessionManager, id: &str) -> Result<Session, AgentError> {
sessions
.get_session(id, true)
.await
.map_err(|e| AgentError::Llm(format!("load session: {e}")))
}
/// Preserve buzz-agent's error taxonomy so the harness's JSON-RPC code mapping
/// stays meaningful.
fn classify_provider_error(msg: &str) -> AgentError {

167
crates/buzz-agent/src/turn_state.rs Normal file
View File

@@ -0,0 +1,167 @@
//! The conversation for one turn, held in memory.
//!
//! buzz-agent used to round-trip every message through goose's
//! `SessionManager` — a sqlite database — once per round. That was never
//! required: none of the goose APIs buzz calls reads the store.
//!
//! * `Provider::stream` takes `&[Message]`, not a session.
//! * `Agent::dispatch_tool_call` takes `&Session` but reads only
//! `working_dir` (for the subdirectory-hint tracker) and `id` (for the
//! `PreToolUse` hook context).
//! * `context_mgmt::{check_if_compaction_needed, compact_messages}` take a
//! `&Session` / `session_id`, but the id is only carried into a tokio
//! task-local that becomes an `agent-session-id` HTTP header for provider
//! request attribution (`goose/src/session_context.rs`), and the session is
//! read for `usage.total_tokens` and `model_config` — both plain fields.
//!
//! So the turn owns a [`Session`] value directly. goose still gets the
//! `&Session` its signatures ask for; it just isn't backed by a database.
//!
//! # Why this matters beyond dependency hygiene
//!
//! A `SessionManager` per agent process meant every buzz agent grew a sqlite
//! file it never read back. Worse, before the store was isolated they all
//! shared the user's own goose CLI history. An agent turn is ephemeral — the
//! durable record of a Buzz conversation is the relay, not a local database.
use goose::conversation::Conversation;
use goose::session::session_manager::Session;
use goose_provider_types::conversation::message::Message;
/// One turn's conversation and the session metadata goose needs alongside it.
pub struct TurnState {
session: Session,
}
impl TurnState {
/// Start a turn from the session's identity and working directory.
///
/// `model_config` is carried because `check_if_compaction_needed` reads it
/// from the session to resolve the context limit; without it goose falls
/// back to a default limit and compaction fires at the wrong point.
pub fn new(
id: String,
working_dir: std::path::PathBuf,
model_config: Option<goose_providers::model::ModelConfig>,
) -> Self {
let mut session = Session {
id,
working_dir,
model_config,
..Session::default()
};
session.conversation = Some(Conversation::new_unvalidated(Vec::new()));
Self { session }
}
/// The session value goose's APIs take.
pub fn session(&self) -> &Session {
&self.session
}
/// This turn's conversation.
pub fn conversation(&self) -> Conversation {
self.session.conversation.clone().unwrap_or_default()
}
/// Append a message to the turn.
pub fn push(&mut self, message: Message) {
let mut messages = self
.session
.conversation
.take()
.map(|conversation| conversation.messages().to_vec())
.unwrap_or_default();
messages.push(message);
self.session.message_count = messages.len();
self.session.conversation = Some(Conversation::new_unvalidated(messages));
}
/// Replace the conversation wholesale, as compaction does.
pub fn replace(&mut self, conversation: Conversation) {
self.session.message_count = conversation.messages().len();
self.session.conversation = Some(conversation);
}
/// Record the running token total.
///
/// `check_if_compaction_needed` prefers `session.usage.total_tokens` over
/// re-counting the conversation, so keeping this current is what makes
/// compaction fire on the provider's numbers rather than an estimate.
pub fn set_total_tokens(&mut self, total: Option<i32>) {
self.session.usage.total_tokens = total;
}
}
#[cfg(test)]
mod tests {
use super::*;
fn state() -> TurnState {
TurnState::new("s1".to_string(), std::path::PathBuf::from("/tmp"), None)
}
#[test]
fn a_new_turn_starts_empty_but_has_a_conversation() {
// `Some(empty)` rather than `None`: goose's compaction check bails on
// a session with no conversation at all.
let state = state();
assert!(state.session().conversation.is_some());
assert!(state.conversation().messages().is_empty());
}
#[test]
fn pushed_messages_are_visible_to_goose() {
let mut state = state();
state.push(Message::user().with_text("hello"));
state.push(Message::assistant().with_text("hi"));
let conversation = state.conversation();
assert_eq!(conversation.messages().len(), 2);
// goose reads the conversation through the session, so the two views
// must not drift.
assert_eq!(
state
.session()
.conversation
.as_ref()
.map(|c| c.messages().len()),
Some(2)
);
assert_eq!(state.session().message_count, 2);
}
#[test]
fn replace_swaps_the_whole_conversation() {
let mut state = state();
state.push(Message::user().with_text("one"));
state.push(Message::user().with_text("two"));
state.replace(Conversation::new_unvalidated(vec![
Message::user().with_text("summary")
]));
assert_eq!(state.conversation().messages().len(), 1);
assert_eq!(state.session().message_count, 1);
}
#[test]
fn total_tokens_reach_the_session_goose_reads() {
let mut state = state();
state.set_total_tokens(Some(1234));
assert_eq!(state.session().usage.total_tokens, Some(1234));
}
#[test]
fn the_working_dir_and_id_goose_reads_are_preserved() {
// These two fields are the entirety of what `dispatch_tool_call` uses
// the session for; getting them wrong breaks subdirectory hints and
// the PreToolUse hook context.
let state = TurnState::new("abc".to_string(), std::path::PathBuf::from("/work"), None);
assert_eq!(state.session().id, "abc");
assert_eq!(
state.session().working_dir,
std::path::PathBuf::from("/work")
);
}
}

90
desktop/src-tauri/src/managed_agents/runtime.rs
View File

@@ -911,93 +911,6 @@ pub fn spawn_agent_child(
})
}
/// Log filter for the spawned harness and, through it, the agent.
///
/// `buzz-acp` inherits its stderr to the agent subprocess (`acp.rs:463`), so
/// whatever this filter admits is what lands in the per-agent log file. That
/// makes `buzz_agent` as load-bearing as `buzz_acp`: without it the agent's own
/// turn diagnostics — compaction, the `_Stop` veto, max-rounds — are filtered
/// out before they reach disk, and an agent misbehaving in the desktop leaves
/// no trace to debug it with.
fn child_rust_log_filter() -> String {
const DEFAULTS: &str = "buzz_acp=info,buzz_agent=info";
match std::env::var("RUST_LOG") {
// An operator filter that already names both crates is taken verbatim.
Ok(existing) if existing.contains("buzz_acp") && existing.contains("buzz_agent") => {
existing
}
// Otherwise append only the directive that is missing, so an explicit
// `buzz_agent=debug` is not overridden by our default.
Ok(existing) if !existing.trim().is_empty() => {
let mut filter = existing;
if !filter.contains("buzz_acp") {
filter.push_str(",buzz_acp=info");
}
if !filter.contains("buzz_agent") {
filter.push_str(",buzz_agent=info");
}
filter
}
_ => DEFAULTS.to_string(),
}
}
#[cfg(test)]
mod rust_log_filter_tests {
use super::child_rust_log_filter;
/// Serialize: `RUST_LOG` is process-global and these tests mutate it.
fn with_rust_log<T>(value: Option<&str>, body: impl FnOnce() -> T) -> T {
static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
let _guard = LOCK.lock().unwrap_or_else(|e| e.into_inner());
let previous = std::env::var("RUST_LOG").ok();
match value {
Some(v) => std::env::set_var("RUST_LOG", v),
None => std::env::remove_var("RUST_LOG"),
}
let result = body();
match previous {
Some(v) => std::env::set_var("RUST_LOG", v),
None => std::env::remove_var("RUST_LOG"),
}
result
}
#[test]
fn default_admits_both_the_harness_and_the_agent() {
// buzz_agent was missing here, which silently discarded every agent
// turn diagnostic from the desktop's per-agent log files.
let filter = with_rust_log(None, child_rust_log_filter);
assert!(filter.contains("buzz_acp=info"));
assert!(filter.contains("buzz_agent=info"));
}
#[test]
fn an_unrelated_operator_filter_is_extended_not_replaced() {
let filter = with_rust_log(Some("warn"), child_rust_log_filter);
assert!(filter.starts_with("warn"), "operator filter must survive");
assert!(filter.contains("buzz_acp=info"));
assert!(filter.contains("buzz_agent=info"));
}
#[test]
fn an_explicit_agent_directive_is_not_downgraded() {
let filter = with_rust_log(Some("buzz_agent=debug"), child_rust_log_filter);
assert!(filter.contains("buzz_agent=debug"));
assert!(!filter.contains("buzz_agent=info"), "must not override");
assert!(filter.contains("buzz_acp=info"), "still needs the harness");
}
#[test]
fn a_fully_specified_filter_is_taken_verbatim() {
let filter = with_rust_log(
Some("buzz_acp=warn,buzz_agent=trace"),
child_rust_log_filter,
);
assert_eq!(filter, "buzz_acp=warn,buzz_agent=trace");
}
}
pub fn start_managed_agent_process(
app: &AppHandle,
record: &mut ManagedAgentRecord,
@@ -1055,6 +968,9 @@ pub fn start_managed_agent_process(
Ok(())
}
mod child_log_filter;
use child_log_filter::child_rust_log_filter;
#[cfg(test)]
mod test_fixtures;

91
desktop/src-tauri/src/managed_agents/runtime/child_log_filter.rs Normal file
View File

@@ -0,0 +1,91 @@
//! Log filter for spawned agent processes.
//!
//! Extracted from `runtime.rs` to keep that file under the desktop file-size
//! ratchet; it is a self-contained decision with its own tests.
/// Log filter for the spawned harness and, through it, the agent.
///
/// `buzz-acp` inherits its stderr to the agent subprocess (`acp.rs:463`), so
/// whatever this filter admits is what lands in the per-agent log file. That
/// makes `buzz_agent` as load-bearing as `buzz_acp`: without it the agent's own
/// turn diagnostics — compaction, the `_Stop` veto, max-rounds — are filtered
/// out before they reach disk, and an agent misbehaving in the desktop leaves
/// no trace to debug it with.
pub(super) fn child_rust_log_filter() -> String {
const DEFAULTS: &str = "buzz_acp=info,buzz_agent=info";
match std::env::var("RUST_LOG") {
// An operator filter that already names both crates is taken verbatim.
Ok(existing) if existing.contains("buzz_acp") && existing.contains("buzz_agent") => {
existing
}
// Otherwise append only the directive that is missing, so an explicit
// `buzz_agent=debug` is not overridden by our default.
Ok(existing) if !existing.trim().is_empty() => {
let mut filter = existing;
if !filter.contains("buzz_acp") {
filter.push_str(",buzz_acp=info");
}
if !filter.contains("buzz_agent") {
filter.push_str(",buzz_agent=info");
}
filter
}
_ => DEFAULTS.to_string(),
}
}
#[cfg(test)]
mod tests {
use super::child_rust_log_filter;
/// Serialize: `RUST_LOG` is process-global and these tests mutate it.
fn with_rust_log<T>(value: Option<&str>, body: impl FnOnce() -> T) -> T {
static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
let _guard = LOCK.lock().unwrap_or_else(|e| e.into_inner());
let previous = std::env::var("RUST_LOG").ok();
match value {
Some(v) => std::env::set_var("RUST_LOG", v),
None => std::env::remove_var("RUST_LOG"),
}
let result = body();
match previous {
Some(v) => std::env::set_var("RUST_LOG", v),
None => std::env::remove_var("RUST_LOG"),
}
result
}
#[test]
fn default_admits_both_the_harness_and_the_agent() {
// buzz_agent was missing here, which silently discarded every agent
// turn diagnostic from the desktop's per-agent log files.
let filter = with_rust_log(None, child_rust_log_filter);
assert!(filter.contains("buzz_acp=info"));
assert!(filter.contains("buzz_agent=info"));
}
#[test]
fn an_unrelated_operator_filter_is_extended_not_replaced() {
let filter = with_rust_log(Some("warn"), child_rust_log_filter);
assert!(filter.starts_with("warn"), "operator filter must survive");
assert!(filter.contains("buzz_acp=info"));
assert!(filter.contains("buzz_agent=info"));
}
#[test]
fn an_explicit_agent_directive_is_not_downgraded() {
let filter = with_rust_log(Some("buzz_agent=debug"), child_rust_log_filter);
assert!(filter.contains("buzz_agent=debug"));
assert!(!filter.contains("buzz_agent=info"), "must not override");
assert!(filter.contains("buzz_acp=info"), "still needs the harness");
}
#[test]
fn a_fully_specified_filter_is_taken_verbatim() {
let filter = with_rust_log(
Some("buzz_acp=warn,buzz_agent=trace"),
child_rust_log_filter,
);
assert_eq!(filter, "buzz_acp=warn,buzz_agent=trace");
}
}