From ce844719f838233846f72dc86180e37f2e6b6c8a Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Tue, 11 Aug 2026 20:25:46 +1000 Subject: [PATCH] refactor(buzz-agent): keep the turn conversation in memory, not sqlite 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 Signed-off-by: Michael Neale --- Cargo.lock | 1 + crates/buzz-agent/Cargo.toml | 3 + crates/buzz-agent/src/lib.rs | 46 ++++- crates/buzz-agent/src/loop_drive.rs | 187 ++++++++++-------- crates/buzz-agent/src/turn_state.rs | 167 ++++++++++++++++ .../src-tauri/src/managed_agents/runtime.rs | 90 +-------- .../runtime/child_log_filter.rs | 91 +++++++++ 7 files changed, 411 insertions(+), 174 deletions(-) create mode 100644 crates/buzz-agent/src/turn_state.rs create mode 100644 desktop/src-tauri/src/managed_agents/runtime/child_log_filter.rs diff --git a/Cargo.lock b/Cargo.lock index dd6bc03d26..93a5ee6096 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1173,6 +1173,7 @@ dependencies = [ "getrandom 0.4.3", "goose", "goose-provider-types", + "goose-providers", "hex", "idna_adapter", "nix 0.31.3", diff --git a/crates/buzz-agent/Cargo.toml b/crates/buzz-agent/Cargo.toml index d764c81257..b76b757a67 100644 --- a/crates/buzz-agent/Cargo.toml +++ b/crates/buzz-agent/Cargo.toml @@ -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 diff --git a/crates/buzz-agent/src/lib.rs b/crates/buzz-agent/src/lib.rs index 6ec8fb9deb..c7f595de29 100644 --- a/crates/buzz-agent/src/lib.rs +++ b/crates/buzz-agent/src/lib.rs @@ -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, - /// 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, + /// 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, @@ -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, 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, 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, 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, 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, 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, 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; diff --git a/crates/buzz-agent/src/loop_drive.rs b/crates/buzz-agent/src/loop_drive.rs index 12494cad58..41905279b9 100644 --- a/crates/buzz-agent/src/loop_drive.rs +++ b/crates/buzz-agent/src/loop_drive.rs @@ -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, super::agent::TurnTokens) { +) -> ( + Result, + super::agent::TurnTokens, + Vec, +) { 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 { - 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 { - 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 = 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, )>, -) -> 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 { 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 { - 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 { diff --git a/crates/buzz-agent/src/turn_state.rs b/crates/buzz-agent/src/turn_state.rs new file mode 100644 index 0000000000..e6a2b495dc --- /dev/null +++ b/crates/buzz-agent/src/turn_state.rs @@ -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, + ) -> 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) { + 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") + ); + } +} diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index e01acc62d9..37e2318d40 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -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(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; diff --git a/desktop/src-tauri/src/managed_agents/runtime/child_log_filter.rs b/desktop/src-tauri/src/managed_agents/runtime/child_log_filter.rs new file mode 100644 index 0000000000..06a7196d98 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/runtime/child_log_filter.rs @@ -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(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"); + } +}