diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 3ff1dc39898..8f1c400af4d 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2584,6 +2584,13 @@ async fn tokio_main() -> Result<()> { tracing::info!("connected to relay at {}", config.relay_url); let relay_rest_client = relay.rest_client(); + let stale_working_reactions = pool::clear_stale_working_reactions(&relay_rest_client).await; + if stale_working_reactions > 0 { + tracing::info!( + count = stale_working_reactions, + "cleared stale working reactions from a previous harness process" + ); + } let mut author_gate_ctx = InboundAuthorGate::connect(&relay_rest_client, &pubkey_hex, "startup").await; @@ -4851,6 +4858,7 @@ fn handle_prompt_result( if let Some(ref observer) = observer { let mut payload = serde_json::json!({ "outcome": outcome_label, + "reason": classify_turn_error_reason(error_msg), "error": error_msg, }); if let Some(code) = error_code { @@ -5047,6 +5055,124 @@ fn handle_prompt_result( LoopAction::Continue } +/// Stable capability reason attached to every `turn_error` observer event. +#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Serialize)] +#[serde(rename_all = "snake_case")] +enum TurnErrorReason { + AuthRequired, + QuotaBlocked, + ModelUnavailable, + TransientProviderError, + Unknown, +} + +/// Classify an ACP failure into the stable operator-facing reason contract. +/// +/// JSON-RPC server-error codes are adapter-specific: Buzz-owned agents reserve +/// codes for authentication and missing models, but external adapters may reuse +/// those same values for unrelated failures. The raw code remains in the +/// observer payload; this stable reason requires narrow diagnostic text so a +/// generic resource error is never presented as a model or credential problem. +/// Unknown errors stay explicitly `unknown` instead of being guessed into a +/// misleading remediation. +fn classify_turn_error_reason(error: &str) -> TurnErrorReason { + let message = error.to_ascii_lowercase(); + if message.contains("llm auth:") + || message.contains("authentication required") + || message.contains("authorization required") + || message.contains("re-authenticate") + || message.contains("api error: 401") + || message.contains("http 401") + || message.contains("401 unauthorized") + || message.contains("oauth access token has expired") + { + return TurnErrorReason::AuthRequired; + } + + if message.contains("http 402") + || message.contains("402 payment required") + || message.contains("credits exhausted") + || message.contains("insufficient_quota") + || message.contains("quota exceeded") + || message.contains("spend limit") + { + return TurnErrorReason::QuotaBlocked; + } + + if message.contains("llm model not found:") + || message.contains("model not found") + || message.contains("model is not available") + || message.contains("model unavailable") + { + return TurnErrorReason::ModelUnavailable; + } + + if message.contains("http 429") + || message.contains("api error: 429") + || message.contains("429 too many requests") + || message.contains("rate limit") + || message.contains("provider overloaded") + || message.contains("service unavailable") + || message.contains("bad gateway") + || message.contains("gateway timeout") + || message.contains("connection reset") + || message.contains("request timeout") + { + return TurnErrorReason::TransientProviderError; + } + + TurnErrorReason::Unknown +} + +#[cfg(test)] +mod turn_error_reason_tests { + use super::{classify_turn_error_reason, TurnErrorReason}; + + #[test] + fn classifies_diagnostics_independently_of_adapter_codes() { + assert_eq!( + classify_turn_error_reason("model not found"), + TurnErrorReason::ModelUnavailable + ); + assert_eq!( + classify_turn_error_reason("quota exceeded"), + TurnErrorReason::QuotaBlocked + ); + } + + #[test] + fn classifies_actionable_provider_text_conservatively() { + assert_eq!( + classify_turn_error_reason("OpenRouter credits exhausted"), + TurnErrorReason::QuotaBlocked + ); + assert_eq!( + classify_turn_error_reason("429 Too Many Requests"), + TurnErrorReason::TransientProviderError + ); + assert_eq!( + classify_turn_error_reason("llm model not found: vendor/model"), + TurnErrorReason::ModelUnavailable + ); + assert_eq!( + classify_turn_error_reason("API Error: 401 OAuth access token has expired"), + TurnErrorReason::AuthRequired + ); + } + + #[test] + fn leaves_ambiguous_failures_unknown() { + assert_eq!( + classify_turn_error_reason("404 endpoint does not support tool use"), + TurnErrorReason::Unknown + ); + assert_eq!( + classify_turn_error_reason("Agent process exited unexpectedly"), + TurnErrorReason::Unknown + ); + } +} + #[allow(clippy::too_many_arguments)] fn recover_panicked_agent( pool: &mut AgentPool, @@ -9735,9 +9861,9 @@ mod error_outcome_emission_tests { .contains_key(&scope::SessionScope::Conversation { channel_id })); } - /// Drive one error outcome through `handle_prompt_result` and return how - /// many `turn_error` events it emitted to the observer feed. - async fn turn_errors_emitted_for(outcome: PromptOutcome) -> usize { + /// Drive one error outcome through `handle_prompt_result` and return the + /// `turn_error` events it emitted to the observer feed. + async fn turn_error_events_for(outcome: PromptOutcome) -> Vec { let agent = dummy_agent(0).await; let mut pool = AgentPool::from_slots(vec![None]); @@ -9808,7 +9934,68 @@ mod error_outcome_emission_tests { .all(|event| event.turn_id.as_deref() == Some("test-turn-id")), "turn_error must retain the completed turn id" ); - turn_errors.len() + turn_errors + } + + async fn turn_errors_emitted_for(outcome: PromptOutcome) -> usize { + turn_error_events_for(outcome).await.len() + } + + async fn turn_error_reason_emitted_for(outcome: PromptOutcome) -> String { + let events = turn_error_events_for(outcome).await; + events[0].payload["reason"] + .as_str() + .expect("turn_error reason is a string") + .to_string() + } + + #[tokio::test] + async fn turn_error_reasons_cross_observer_boundary() { + assert_eq!( + turn_error_reason_emitted_for(PromptOutcome::Error(acp::AcpError::AgentError { + code: -32001, + message: "authentication required".into(), + },)) + .await, + "auth_required" + ); + assert_eq!( + turn_error_reason_emitted_for(PromptOutcome::Error(acp::AcpError::AgentError { + code: -32002, + message: "llm model not found: configured model".into(), + },)) + .await, + "model_unavailable" + ); + assert_eq!( + turn_error_reason_emitted_for(PromptOutcome::Error(acp::AcpError::AgentError { + code: -32000, + message: "OpenRouter credits exhausted".into(), + },)) + .await, + "quota_blocked" + ); + assert_eq!( + turn_error_reason_emitted_for(PromptOutcome::Error(acp::AcpError::AgentError { + code: -32000, + message: "API Error: 429 Too Many Requests".into(), + },)) + .await, + "transient_provider_error" + ); + assert_eq!( + turn_error_reason_emitted_for(PromptOutcome::Error(acp::AcpError::AgentError { + code: -32002, + message: "Resource not found: session no longer exists".into(), + },)) + .await, + "unknown", + "adapter-specific server codes must not mislabel generic resources as models" + ); + assert_eq!( + turn_error_reason_emitted_for(PromptOutcome::AgentExited).await, + "unknown" + ); } #[tokio::test] diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index c188633bef7..24cd7c0bfc7 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -5266,6 +5266,12 @@ async fn publish_agent_turn_metric( const REACTION_SEEN: &str = "👀"; const REACTION_WORKING: &str = "💬"; +/// Bound startup cleanup so a long-lived identity with an unexpectedly large +/// reaction history cannot delay harness readiness without limit. Relay +/// queries exclude deleted events, so this normally returns zero or a handful +/// left behind by the immediately preceding process. +const STALE_WORKING_REACTION_LIMIT: usize = 100; + /// Best-effort timeout for a single reaction REST call. const REACTION_TIMEOUT: Duration = Duration::from_millis(500); @@ -5453,6 +5459,77 @@ pub(crate) async fn reaction_remove(rest: &crate::relay::RestClient, event_id: & } } +/// Best-effort startup cleanup for 💬 reactions left by a process that died +/// before its [`ReactionGuard`] could unwind. +/// +/// The query is scoped to this agent's signing key and deletion events target +/// only reaction events signed by that same key. Cleanup runs before channel +/// subscriptions are installed, so it cannot remove a reaction created by a +/// turn in this process. Failures are cosmetic and never block startup. +pub(crate) async fn clear_stale_working_reactions(rest: &crate::relay::RestClient) -> usize { + let filter = nostr::Filter::new() + .kind(nostr::Kind::Reaction) + .author(rest.keys.public_key()) + .limit(STALE_WORKING_REACTION_LIMIT); + let response = match tokio::time::timeout(REACTION_TIMEOUT, rest.query(&[filter])).await { + Ok(Ok(response)) => response, + Ok(Err(error)) => { + tracing::debug!("startup working-reaction query failed: {error}"); + return 0; + } + Err(_) => { + tracing::debug!("startup working-reaction query timed out"); + return 0; + } + }; + + let reaction_ids: Vec = response + .as_array() + .into_iter() + .flatten() + .filter(|event| { + event.get("content").and_then(serde_json::Value::as_str) == Some(REACTION_WORKING) + }) + .filter_map(|event| event.get("id").and_then(serde_json::Value::as_str)) + .filter_map(|id| nostr::EventId::from_hex(id).ok()) + .take(STALE_WORKING_REACTION_LIMIT) + .collect(); + + let mut cleared = 0; + for chunk in reaction_ids.chunks(REACTION_CONCURRENCY) { + let results = futures_util::future::join_all(chunk.iter().map(|reaction_id| async move { + let builder = match buzz_sdk::build_remove_reaction(*reaction_id) { + Ok(builder) => builder, + Err(error) => { + tracing::warn!(%reaction_id, "startup reaction cleanup build failed: {error}"); + return false; + } + }; + let event = match builder.sign_with_keys(&rest.keys) { + Ok(event) => event, + Err(error) => { + tracing::warn!(%reaction_id, "startup reaction cleanup sign failed: {error}"); + return false; + } + }; + match tokio::time::timeout(REACTION_TIMEOUT, rest.submit_event(&event)).await { + Ok(Ok(_)) => true, + Ok(Err(error)) => { + tracing::debug!(%reaction_id, "startup reaction cleanup failed: {error}"); + false + } + Err(_) => { + tracing::debug!(%reaction_id, "startup reaction cleanup timed out"); + false + } + } + })) + .await; + cleared += results.into_iter().filter(|removed| *removed).count(); + } + cleared +} + /// Maximum concurrent reaction HTTP requests per fan-out call. /// Prevents unbounded parallelism when a large batch of events arrives. const REACTION_CONCURRENCY: usize = 10; @@ -5508,6 +5585,124 @@ mod tests { } } + #[tokio::test(flavor = "multi_thread")] + async fn startup_cleanup_queries_and_deletes_only_working_reactions() { + use std::sync::{Arc, Mutex}; + use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; + + let agent_keys = Keys::generate(); + let target = EventBuilder::new(Kind::Custom(9), "trigger") + .sign_with_keys(&Keys::generate()) + .unwrap(); + let working = buzz_sdk::build_reaction(target.id, REACTION_WORKING) + .unwrap() + .sign_with_keys(&agent_keys) + .unwrap(); + let seen = buzz_sdk::build_reaction(target.id, REACTION_SEEN) + .unwrap() + .sign_with_keys(&agent_keys) + .unwrap(); + let query_response = json!([ + working, + seen, + { "content": REACTION_WORKING, "id": "not-a-valid-event-id" } + ]) + .to_string(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind startup cleanup test server"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let requests = Arc::new(Mutex::new(Vec::<(String, serde_json::Value)>::new())); + let server_requests = requests.clone(); + let server = tokio::spawn(async move { + for response_body in [query_response.as_str(), "{}"] { + let (socket, _) = listener.accept().await.expect("accept cleanup request"); + let mut reader = BufReader::new(socket); + let mut line = String::new(); + reader.read_line(&mut line).await.unwrap(); + let path = line + .split_whitespace() + .nth(1) + .expect("request path") + .to_string(); + + let mut content_length = None; + for _ in 0..64 { + line.clear(); + assert_ne!(reader.read_line(&mut line).await.unwrap(), 0); + if line == "\r\n" { + break; + } + if let Some(value) = line.to_ascii_lowercase().strip_prefix("content-length:") { + content_length = Some(value.trim().parse::().unwrap()); + } + } + let mut body = vec![0; content_length.expect("request Content-Length")]; + reader.read_exact(&mut body).await.unwrap(); + let body = serde_json::from_slice(&body).expect("request JSON"); + server_requests.lock().unwrap().push((path, body)); + + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + response_body.len(), + response_body + ); + reader + .into_inner() + .write_all(response.as_bytes()) + .await + .unwrap(); + } + }); + + let rest = crate::relay::RestClient { + http: reqwest::Client::new(), + base_url, + keys: agent_keys.clone(), + auth_tag_json: None, + }; + assert_eq!(clear_stale_working_reactions(&rest).await, 1); + tokio::time::timeout(Duration::from_secs(3), server) + .await + .expect("cleanup server completes") + .expect("cleanup server succeeds"); + + let requests = requests.lock().unwrap(); + assert_eq!(requests.len(), 2); + assert_eq!(requests[0].0, "/query"); + assert_eq!(requests[0].1[0]["kinds"], json!([7])); + assert_eq!( + requests[0].1[0]["authors"], + json!([agent_keys.public_key().to_hex()]) + ); + assert_eq!( + requests[0].1[0]["limit"], + json!(STALE_WORKING_REACTION_LIMIT) + ); + + assert_eq!(requests[1].0, "/events"); + let deletion = &requests[1].1; + assert_eq!(deletion["kind"], 5); + assert_eq!(deletion["pubkey"], agent_keys.public_key().to_hex()); + assert!( + deletion["tags"] + .as_array() + .unwrap() + .iter() + .any(|tag| tag[0] == "e" && tag[1] == working.id.to_hex()), + "deletion must target the working reaction" + ); + assert!( + deletion["tags"] + .as_array() + .unwrap() + .iter() + .all(|tag| tag[0] != "e" || tag[1] != seen.id.to_hex()), + "startup cleanup must leave non-working reactions untouched" + ); + } + #[test] fn delivery_receipt_line_sorts_event_ids() { let channel_id = Uuid::nil();