From 29884e433f54e51fb1836694e2eadd71fe802845 Mon Sep 17 00:00:00 2001 From: David Fitzsimmons Date: Sun, 26 Jul 2026 23:19:37 -0400 Subject: [PATCH 1/3] fix(acp): reduce runaway subscription usage Signed-off-by: David Fitzsimmons --- crates/buzz-acp/src/acp.rs | 58 ++++++++++--- crates/buzz-acp/src/filter.rs | 48 +++++++++++ crates/buzz-acp/src/lib.rs | 155 ++++++++++++++++++++++++++++++++-- crates/buzz-acp/src/pool.rs | 47 ++++++++--- crates/buzz-acp/src/queue.rs | 71 ++++++++++++++-- 5 files changed, 345 insertions(+), 34 deletions(-) diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 78db7ff718..081f40a1f3 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -10,12 +10,14 @@ use futures_util::StreamExt; use tokio::io::AsyncWriteExt; -use tokio::process::{Child, ChildStdin, ChildStdout}; +use tokio::process::{Child, ChildStdout}; use tokio_util::codec::{FramedRead, LinesCodec, LinesCodecError}; use crate::observer::{ObserverContext, ObserverHandle}; use crate::usage::{TurnUsage, UsageTracker}; +type StdinWriteRequest = (Vec, tokio::sync::oneshot::Sender>); + /// Maximum allowed size of a single NDJSON line from the agent's stdout. /// Lines exceeding this limit are rejected to prevent OOM from rogue agents. const MAX_LINE_SIZE: usize = 10_000_000; // 10 MB @@ -139,8 +141,13 @@ fn build_initialize_params() -> serde_json::Value { pub struct AcpClient { /// The agent child process (kept alive to prevent zombie). child: Child, - /// Write end of the agent's stdin pipe. - stdin: ChildStdin, + /// Cancellation-safe writer queue for the agent's stdin pipe. + /// + /// A dedicated task owns `ChildStdin` and finishes each complete NDJSON + /// frame after accepting it. This prevents a cancelled prompt future from + /// leaving a partial JSON payload that the next control frame would append + /// to on the same stream. + stdin_tx: tokio::sync::mpsc::Sender, /// Framed reader over the agent's stdout pipe (line-oriented, bounded). /// Uses `LinesCodec::new_with_max_length` to enforce MAX_LINE_SIZE at the /// read level — prevents OOM from rogue agents writing infinite non-newline bytes. @@ -481,9 +488,26 @@ impl AcpClient { .take() .ok_or_else(|| AcpError::Protocol("failed to open agent stdout".into()))?; + let (stdin_tx, mut stdin_rx) = tokio::sync::mpsc::channel::(8); + tokio::spawn(async move { + let mut stdin = stdin; + while let Some((frame, completion_tx)) = stdin_rx.recv().await { + let result = async { + stdin.write_all(&frame).await?; + stdin.flush().await + } + .await; + let write_failed = result.is_err(); + let _ = completion_tx.send(result); + if write_failed { + break; + } + } + }); + Ok(Self { child, - stdin, + stdin_tx, reader: FramedRead::new(stdout, LinesCodec::new_with_max_length(MAX_LINE_SIZE)), next_id: 0, pending_permission_id: None, @@ -944,18 +968,32 @@ impl AcpClient { self.parse_stop_reason(&result) } - /// Serialize `value` as a single NDJSON line and flush to the agent's stdin. + /// Serialize `value` as a single NDJSON frame and enqueue it for the + /// dedicated stdin writer. /// /// Bounded by a 30-second write timeout. If the agent stops reading stdin /// (e.g., it's stuck or dead), the write would otherwise block forever. async fn write_ndjson(&mut self, value: &serde_json::Value) -> Result<(), AcpError> { const WRITE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); - let line = serde_json::to_string(value)?; + let mut frame = serde_json::to_vec(value)?; + frame.push(b'\n'); + let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); tokio::time::timeout(WRITE_TIMEOUT, async { - self.stdin.write_all(line.as_bytes()).await?; - self.stdin.write_all(b"\n").await?; - self.stdin.flush().await?; - Ok::<(), std::io::Error>(()) + self.stdin_tx + .send((frame, completion_tx)) + .await + .map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "agent stdin writer stopped", + ) + })?; + completion_rx.await.map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "agent stdin writer dropped completion", + ) + })? }) .await .map_err(|_| AcpError::WriteTimeout(WRITE_TIMEOUT))? diff --git a/crates/buzz-acp/src/filter.rs b/crates/buzz-acp/src/filter.rs index 43edd969dd..eca67b5f15 100644 --- a/crates/buzz-acp/src/filter.rs +++ b/crates/buzz-acp/src/filter.rs @@ -371,6 +371,21 @@ pub async fn match_event( rules: &[SubscriptionRule], agent_pubkey_hex: &str, ) -> Option { + // UI activity signals are never actionable agent input. Keep this guard + // ahead of operator-authored rules so an accidental wildcard subscription + // cannot turn high-frequency typing/presence traffic into LLM prompts or + // mid-turn steer signals. Huddle reaction bursts are similarly visual-only. + // Durable message reactions (kind 7) are intentionally not included here: + // callers may explicitly subscribe to those for an agent workflow. + if matches!( + event.kind.as_u16() as u32, + buzz_core::kind::KIND_PRESENCE_UPDATE + | buzz_core::kind::KIND_TYPING_INDICATOR + | buzz_core::kind::KIND_HUDDLE_REACTION + ) { + return None; + } + let filter_ctx = FilterContext::from_event(event, channel_id); for (index, rule) in rules.iter().enumerate() { @@ -635,6 +650,39 @@ mod tests { assert_eq!(matched.prompt_tag, "matched"); } + #[tokio::test] + async fn test_match_event_rejects_ui_activity_signals_under_wildcard_rule() { + let channel_id = any_channel(); + let rules = vec![make_rule( + "wildcard", + ChannelScope::All("all".into()), + vec![], + false, + None, + Some("all"), + )]; + + for kind in [ + buzz_core::kind::KIND_PRESENCE_UPDATE, + buzz_core::kind::KIND_TYPING_INDICATOR, + buzz_core::kind::KIND_HUDDLE_REACTION, + ] { + let event = make_event(kind, ""); + assert!( + match_event(&event, channel_id, &rules, "").await.is_none(), + "UI activity kind {kind} must never become an agent prompt" + ); + } + + let message = make_event(buzz_core::kind::KIND_STREAM_MESSAGE, "hello"); + assert!( + match_event(&message, channel_id, &rules, "") + .await + .is_some(), + "the wildcard rule must still accept actionable messages" + ); + } + #[tokio::test] async fn test_match_event_require_mention() { let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 0230ea0875..e130a14d6a 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -3007,6 +3007,26 @@ fn is_auth_error(error: &acp::AcpError) -> bool { message.contains("Re-authenticate") || message.contains("API Error: 401") } +/// Returns `true` when the provider has rejected work until the user's quota +/// or subscription window resets. +/// +/// These failures cannot recover through immediate retry. Requeueing them can +/// repeatedly submit the same large prompt, consuming retry budget and making +/// an already-exhausted subscription worse. Match only provider-originated +/// `AgentError` messages and narrow, observed quota phrases to avoid treating +/// transient transport/rate errors as terminal. +fn is_usage_limit_error(error: &acp::AcpError) -> bool { + let acp::AcpError::AgentError { message, .. } = error else { + return false; + }; + let message = message.to_ascii_lowercase(); + message.contains("session limit") + || message.contains("usage credits required") + || message.contains("weekly limit") + || message.contains("monthly limit") + || message.contains("spending limit") +} + /// Spawn a task that posts a user-visible failure notice to the relay. /// /// Shared by the hard-cap immediate dead-letter path and the retries-exhausted @@ -3142,6 +3162,22 @@ fn handle_prompt_result( and then re-send." .to_string(); spawn_failure_notice(rest_client, &batch, content); + } else if matches!(&result.outcome, PromptOutcome::Error(e) if is_usage_limit_error(e)) + { + // Subscription/quota failures are also non-retryable until an + // external reset or account change. Do not resubmit the same + // prompt in a retry storm. + tracing::warn!( + channel_id = %batch.channel_id, + events = batch.events.len(), + "dead-lettering batch immediately — provider usage limit reached" + ); + let content = "⚠️ I couldn't process the last request because the AI provider's \ + usage or subscription limit was reached. Wait for the provider reset (or \ + change the configured model/account), then re-send. Buzz will not retry \ + automatically." + .to_string(); + spawn_failure_notice(rest_client, &batch, content); } else if let Some(dead) = queue.requeue(batch) { let reason = match &result.outcome { PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(), @@ -6146,6 +6182,38 @@ mod error_outcome_emission_tests { ); } + // ── is_usage_limit_error classification ─────────────────────────────── + + #[test] + fn is_usage_limit_error_matches_claude_session_limit() { + let e = acp::AcpError::AgentError { + code: -32000, + message: "You've hit your session limit · resets 9:50pm".to_string(), + }; + assert!(is_usage_limit_error(&e)); + } + + #[test] + fn is_usage_limit_error_matches_usage_credits_requirement() { + let e = acp::AcpError::AgentError { + code: -32000, + message: "Usage credits required for 1M context".to_string(), + }; + assert!(is_usage_limit_error(&e)); + } + + #[test] + fn is_usage_limit_error_rejects_transient_and_transport_errors() { + let transient = acp::AcpError::AgentError { + code: -32000, + message: "Service temporarily unavailable".to_string(), + }; + assert!(!is_usage_limit_error(&transient)); + assert!(!is_usage_limit_error(&acp::AcpError::Io( + std::io::Error::other("pipe broke") + ))); + } + // ── auth error dead-letter behavior ──────────────────────────────────── /// An auth-class `PromptOutcome::Error` must dead-letter immediately @@ -6234,10 +6302,10 @@ mod error_outcome_emission_tests { ); } - /// A non-auth application error (e.g. usage credits) must still follow the - /// standard requeue path so today's behavior is unchanged. + /// A provider quota error must dead-letter immediately instead of + /// repeatedly resubmitting the same prompt before the external reset. #[tokio::test] - async fn non_auth_application_error_is_requeued() { + async fn usage_limit_error_dead_letters_immediately_without_requeueing() { let keys = nostr::Keys::generate(); let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "test") .sign_with_keys(&keys) @@ -6306,17 +6374,88 @@ mod error_outcome_emission_tests { None, ); - // Non-auth application error: batch IS requeued (first attempt, retry budget > 0). + // Provider usage error: batch is not requeued. assert_eq!( queue.pending_channels(), - 1, - "non-auth application error must requeue the batch for retry" + 0, + "usage-limit error must not requeue the batch" ); assert_eq!( queue.queued_event_count(&channel_id), - 1, - "non-auth application error must preserve the event for retry" + 0, + "usage-limit error must not leave events pending" + ); + } + + /// Other application failures retain the bounded retry behavior. + #[tokio::test] + async fn transient_application_error_is_requeued() { + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "test") + .sign_with_keys(&keys) + .unwrap(); + let channel_id = uuid::Uuid::new_v4(); + let batch = FlushBatch { + channel_id, + events: vec![BatchEvent { + event, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + let error = acp::AcpError::AgentError { + code: -32000, + message: "Service temporarily unavailable".to_string(), + }; + let agent = dummy_agent(0).await; + let mut pool = AgentPool::from_slots(vec![None]); + let task_id = pool.join_set.spawn(async {}).id(); + pool.task_map_mut().insert( + task_id, + crate::pool::TaskMeta { + agent_index: 0, + channel_id: None, + turn_id: "test-turn-id".to_string(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + }, + ); + let mut queue = EventQueue::new(config::DedupMode::Queue); + let config = test_config(); + let mut heartbeat_in_flight = false; + let removed_channels = std::collections::HashSet::new(); + let mut crash_history = vec![SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: false, + }]; + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let result = PromptResult { + agent, + source: PromptSource::Channel(channel_id), + turn_id: "test-turn-id".to_string(), + outcome: PromptOutcome::Error(error), + batch: Some(batch), + }; + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, ); + assert_eq!(queue.pending_channels(), 1); + assert_eq!(queue.queued_event_count(&channel_id), 1); } } diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index cc537f8683..815233dbd6 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -1104,7 +1104,7 @@ pub(crate) fn prepend_base_for_legacy( /// /// Protocol-v2 agents already receive the canvas in `systemPrompt`; only /// legacy (protocol_version < 2) agents need it injected here so it arrives -/// before the first prompt — the same "every turn" semantics as per-turn core. +/// before the first prompt, alongside the other session-scoped preamble data. /// Heartbeats never have an initial_message, so the caller is responsible for /// not passing a canvas when `source` is `Heartbeat`. pub(crate) fn prepend_canvas_for_legacy( @@ -1589,7 +1589,7 @@ pub async fn run_prompt_task( // Legacy agents receive it via [Base] in the user message instead. // Canvas is also injected here for legacy agents: protocol-v2 agents // already have it in systemPrompt; legacy agents need it before the - // first prompt, matching the "every turn" per-turn delivery semantics. + // first prompt, alongside the other session-scoped preamble data. let init_msg = prepend_base_for_legacy( if agent.has_system_prompt_support() { 2 @@ -1728,15 +1728,19 @@ pub async fn run_prompt_task( let prompt_sections: Vec = if let Some(text) = prompt_text { // Heartbeats create their session before this point, so a Goose method-not-found // probe has already selected the correct framing for this process. - let text = prepend_base_for_legacy( - if agent.has_system_prompt_support() { - 2 - } else { - 1 - }, - ctx.base_prompt, - &text, - ); + let text = if is_new_session { + prepend_base_for_legacy( + if agent.has_system_prompt_support() { + 2 + } else { + 1 + }, + ctx.base_prompt, + &text, + ) + } else { + text + }; vec![text] } else if let Some(ref b) = batch { // Build prompt from batch with context enrichment. @@ -1780,6 +1784,7 @@ pub async fn run_prompt_task( system_prompt: ctx.system_prompt.as_deref(), team_instructions: ctx.team_instructions.as_deref(), agent_canvas: agent_canvas.as_deref(), + include_legacy_session_preamble: is_new_session, }, ) } else { @@ -1819,6 +1824,26 @@ pub async fn run_prompt_task( .collect(), None => prompt_sections.iter().map(String::as_str).collect(), }; + let prompt_bytes: usize = prompt_blocks.iter().map(|block| block.len()).sum(); + tracing::info!( + target: "pool::prompt", + prompt_bytes, + prompt_blocks = prompt_blocks.len(), + is_new_session, + system_prompt_transport = if agent.has_system_prompt_support() { + "system" + } else { + "legacy-user-prefix" + }, + "prompt prepared" + ); + if prompt_bytes > 50_000 { + tracing::warn!( + target: "pool::prompt", + prompt_bytes, + "large prompt prepared — inspect repeated base, system, memory, and conversation context" + ); + } // When control_rx is Some (channel tasks), wrap the prompt in select! so // the main loop can cancel, interrupt, or rotate it. Heartbeats diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 029bf86dbf..0b12a61375 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1349,7 +1349,6 @@ fn format_conversation_context( } /// Arguments for [`format_prompt`] beyond the required [`FlushBatch`]. -#[derive(Default)] pub struct FormatPromptArgs<'a> { pub agent_core: Option<&'a str>, pub channel_info: Option<&'a PromptChannelInfo>, @@ -1369,9 +1368,33 @@ pub struct FormatPromptArgs<'a> { /// /// For modern agents (protocol_version >= 2) the section is delivered via /// the system role in session/new; omit here to avoid duplication. - /// For legacy agents it rides in the user message on every turn of the - /// session, alongside `[Base]`/`[System]`/`[Agent Memory — core]`. + /// For legacy agents it rides in the first user message of the session, + /// alongside `[Base]`/`[System]`/`[Agent Memory — core]`. pub agent_canvas: Option<&'a str>, + /// Whether the stable legacy-only session preamble should be included. + /// + /// Legacy ACP agents have no system-prompt transport, so Buzz sends the + /// base, persona, team, core memory, and canvas sections in the first user + /// turn. The ACP session retains that turn; repeating the same sections on + /// every later turn wastes input tokens without adding information. + pub include_legacy_session_preamble: bool, +} + +impl Default for FormatPromptArgs<'_> { + fn default() -> Self { + Self { + agent_core: None, + channel_info: None, + conversation_context: None, + profile_lookup: None, + has_system_prompt_support: false, + base_prompt: None, + system_prompt: None, + team_instructions: None, + agent_canvas: None, + include_legacy_session_preamble: true, + } + } } /// Format the `[Base]` section for the base prompt. @@ -1426,7 +1449,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec) -> Vec Date: Mon, 27 Jul 2026 00:51:13 -0400 Subject: [PATCH 2/3] fix(acp): preserve legacy session instructions Signed-off-by: David Fitzsimmons --- crates/buzz-acp/src/pool.rs | 27 ++++++-------- crates/buzz-acp/src/queue.rs | 71 +++--------------------------------- 2 files changed, 16 insertions(+), 82 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 815233dbd6..fd8ea27c1d 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -1104,7 +1104,7 @@ pub(crate) fn prepend_base_for_legacy( /// /// Protocol-v2 agents already receive the canvas in `systemPrompt`; only /// legacy (protocol_version < 2) agents need it injected here so it arrives -/// before the first prompt, alongside the other session-scoped preamble data. +/// before the first prompt — the same "every turn" semantics as per-turn core. /// Heartbeats never have an initial_message, so the caller is responsible for /// not passing a canvas when `source` is `Heartbeat`. pub(crate) fn prepend_canvas_for_legacy( @@ -1589,7 +1589,7 @@ pub async fn run_prompt_task( // Legacy agents receive it via [Base] in the user message instead. // Canvas is also injected here for legacy agents: protocol-v2 agents // already have it in systemPrompt; legacy agents need it before the - // first prompt, alongside the other session-scoped preamble data. + // first prompt, matching the "every turn" per-turn delivery semantics. let init_msg = prepend_base_for_legacy( if agent.has_system_prompt_support() { 2 @@ -1728,19 +1728,15 @@ pub async fn run_prompt_task( let prompt_sections: Vec = if let Some(text) = prompt_text { // Heartbeats create their session before this point, so a Goose method-not-found // probe has already selected the correct framing for this process. - let text = if is_new_session { - prepend_base_for_legacy( - if agent.has_system_prompt_support() { - 2 - } else { - 1 - }, - ctx.base_prompt, - &text, - ) - } else { - text - }; + let text = prepend_base_for_legacy( + if agent.has_system_prompt_support() { + 2 + } else { + 1 + }, + ctx.base_prompt, + &text, + ); vec![text] } else if let Some(ref b) = batch { // Build prompt from batch with context enrichment. @@ -1784,7 +1780,6 @@ pub async fn run_prompt_task( system_prompt: ctx.system_prompt.as_deref(), team_instructions: ctx.team_instructions.as_deref(), agent_canvas: agent_canvas.as_deref(), - include_legacy_session_preamble: is_new_session, }, ) } else { diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 0b12a61375..029bf86dbf 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1349,6 +1349,7 @@ fn format_conversation_context( } /// Arguments for [`format_prompt`] beyond the required [`FlushBatch`]. +#[derive(Default)] pub struct FormatPromptArgs<'a> { pub agent_core: Option<&'a str>, pub channel_info: Option<&'a PromptChannelInfo>, @@ -1368,33 +1369,9 @@ pub struct FormatPromptArgs<'a> { /// /// For modern agents (protocol_version >= 2) the section is delivered via /// the system role in session/new; omit here to avoid duplication. - /// For legacy agents it rides in the first user message of the session, - /// alongside `[Base]`/`[System]`/`[Agent Memory — core]`. + /// For legacy agents it rides in the user message on every turn of the + /// session, alongside `[Base]`/`[System]`/`[Agent Memory — core]`. pub agent_canvas: Option<&'a str>, - /// Whether the stable legacy-only session preamble should be included. - /// - /// Legacy ACP agents have no system-prompt transport, so Buzz sends the - /// base, persona, team, core memory, and canvas sections in the first user - /// turn. The ACP session retains that turn; repeating the same sections on - /// every later turn wastes input tokens without adding information. - pub include_legacy_session_preamble: bool, -} - -impl Default for FormatPromptArgs<'_> { - fn default() -> Self { - Self { - agent_core: None, - channel_info: None, - conversation_context: None, - profile_lookup: None, - has_system_prompt_support: false, - base_prompt: None, - system_prompt: None, - team_instructions: None, - agent_canvas: None, - include_legacy_session_preamble: true, - } - } } /// Format the `[Base]` section for the base prompt. @@ -1449,7 +1426,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec) -> Vec Date: Mon, 27 Jul 2026 12:25:31 -0400 Subject: [PATCH 3/3] fix(acp): preserve prompt usage telemetry Signed-off-by: David Fitzsimmons --- crates/buzz-acp/src/pool.rs | 101 ++++++++++++++++++++++++++++++------ 1 file changed, 85 insertions(+), 16 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index fd8ea27c1d..1959299999 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -1250,6 +1250,36 @@ fn send_prompt_result( }); } +fn log_prompt_metrics( + prompt_bytes: usize, + prompt_blocks: usize, + is_new_session: bool, + supports_system_prompt: bool, +) { + // Keep this target under `buzz_acp`: Desktop intentionally runs managed + // harnesses with `warn,buzz_acp=info`, so a separate `pool::prompt` target + // silently drops ordinary usage measurements. + tracing::info!( + target: "buzz_acp::pool::prompt", + prompt_bytes, + prompt_blocks, + is_new_session, + system_prompt_transport = if supports_system_prompt { + "system" + } else { + "legacy-user-prefix" + }, + "prompt prepared" + ); + if prompt_bytes > 50_000 { + tracing::warn!( + target: "buzz_acp::pool::prompt", + prompt_bytes, + "large prompt prepared — inspect repeated base, system, memory, and conversation context" + ); + } +} + /// Core async function spawned for each prompt. /// /// Lifecycle: @@ -1820,25 +1850,12 @@ pub async fn run_prompt_task( None => prompt_sections.iter().map(String::as_str).collect(), }; let prompt_bytes: usize = prompt_blocks.iter().map(|block| block.len()).sum(); - tracing::info!( - target: "pool::prompt", + log_prompt_metrics( prompt_bytes, - prompt_blocks = prompt_blocks.len(), + prompt_blocks.len(), is_new_session, - system_prompt_transport = if agent.has_system_prompt_support() { - "system" - } else { - "legacy-user-prefix" - }, - "prompt prepared" + agent.has_system_prompt_support(), ); - if prompt_bytes > 50_000 { - tracing::warn!( - target: "pool::prompt", - prompt_bytes, - "large prompt prepared — inspect repeated base, system, memory, and conversation context" - ); - } // When control_rx is Some (channel tasks), wrap the prompt in select! so // the main loop can cancel, interrupt, or rotate it. Heartbeats @@ -3672,6 +3689,58 @@ mod tests { use super::*; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; + use std::io::Write; + + #[derive(Clone)] + struct CapturingMakeWriter { + buffer: Arc>>, + } + + struct CapturingWriter { + buffer: Arc>>, + } + + impl Write for CapturingWriter { + fn write(&mut self, data: &[u8]) -> std::io::Result { + self.buffer.lock().unwrap().extend_from_slice(data); + Ok(data.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturingMakeWriter { + type Writer = CapturingWriter; + + fn make_writer(&'a self) -> Self::Writer { + CapturingWriter { + buffer: Arc::clone(&self.buffer), + } + } + } + + #[test] + fn prompt_metrics_survive_desktop_child_log_filter() { + let buffer = Arc::new(Mutex::new(Vec::new())); + let subscriber = tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::new("warn,buzz_acp=info")) + .with_writer(CapturingMakeWriter { + buffer: Arc::clone(&buffer), + }) + .with_ansi(false) + .finish(); + + tracing::subscriber::with_default(subscriber, || { + log_prompt_metrics(321, 4, true, false); + }); + + let captured = String::from_utf8(buffer.lock().unwrap().clone()).unwrap(); + assert!(captured.contains("buzz_acp::pool::prompt"), "{captured}"); + assert!(captured.contains("prompt prepared"), "{captured}"); + assert!(captured.contains("prompt_bytes=321"), "{captured}"); + } // These pin the initial_message dispatch path (run_prompt_task, ~line 855): // a legacy agent WITH a base_prompt must get [Base] prepended to the user