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..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: @@ -1819,6 +1849,13 @@ 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(); + log_prompt_metrics( + prompt_bytes, + prompt_blocks.len(), + is_new_session, + agent.has_system_prompt_support(), + ); // When control_rx is Some (channel tasks), wrap the prompt in select! so // the main loop can cancel, interrupt, or rotate it. Heartbeats @@ -3652,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