diff --git a/crates/buzz-acp/src/filter.rs b/crates/buzz-acp/src/filter.rs index 43edd969dd..fae1a57fc8 100644 --- a/crates/buzz-acp/src/filter.rs +++ b/crates/buzz-acp/src/filter.rs @@ -371,6 +371,25 @@ pub async fn match_event( rules: &[SubscriptionRule], agent_pubkey_hex: &str, ) -> Option { + // Recipient isolation is independent of a rule's `require_mention` setting. + // Owner-direct rules deliberately allow untagged owner messages, but must + // never receive an event explicitly addressed to a different agent. + // A message addressed to multiple agents is valid for each addressed agent. + let mut has_recipients = false; + let mut addresses_agent = false; + for tag in event.tags.iter() { + let values = tag.as_slice(); + if values.first().map(|kind| kind.as_str()) == Some("p") { + has_recipients = true; + if values.get(1).map(|pubkey| pubkey.as_str()) == Some(agent_pubkey_hex) { + addresses_agent = true; + } + } + } + if has_recipients && !addresses_agent { + return None; + } + let filter_ctx = FilterContext::from_event(event, channel_id); for (index, rule) in rules.iter().enumerate() { @@ -484,6 +503,17 @@ mod tests { .unwrap() } + fn make_event_with_p_tags(kind: u32, content: &str, p_hexes: &[&str]) -> nostr::Event { + let keys = Keys::generate(); + let tags = p_hexes + .iter() + .map(|p_hex| Tag::parse(["p", *p_hex]).expect("tag parse")); + EventBuilder::new(Kind::Custom(kind as u16), content) + .tags(tags) + .sign_with_keys(&keys) + .unwrap() + } + fn any_channel() -> Uuid { Uuid::new_v4() } @@ -663,6 +693,68 @@ mod tests { assert_eq!(matched.prompt_tag, "mentioned"); } + #[tokio::test] + async fn test_match_event_rejects_foreign_recipient_before_owner_direct_rule() { + let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; + let foreign_pubkey = "feedfacefeedfacefeedfacefeedfacefeedfacefeedfacefeedfacefeedface"; + let event = make_event_with_p_tag(9, "for another agent", foreign_pubkey); + let rules = vec![make_rule( + "owner-direct", + ChannelScope::All("all".into()), + vec![9], + false, + None, + None, + )]; + + assert!(match_event(&event, any_channel(), &rules, agent_pubkey) + .await + .is_none()); + } + + #[tokio::test] + async fn test_match_event_accepts_self_and_foreign_recipients() { + let agent_pubkey = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; + let foreign_pubkey = "feedfacefeedfacefeedfacefeedfacefeedfacefeedfacefeedfacefeedface"; + let event = make_event_with_p_tags(9, "for both agents", &[foreign_pubkey, agent_pubkey]); + let rules = vec![make_rule( + "owner-direct", + ChannelScope::All("all".into()), + vec![9], + false, + None, + Some("owner-direct"), + )]; + + let matched = match_event(&event, any_channel(), &rules, agent_pubkey) + .await + .expect("self-addressed event must remain eligible"); + assert_eq!(matched.prompt_tag, "owner-direct"); + } + + #[tokio::test] + async fn test_match_event_accepts_untagged_owner_direct_event() { + let event = make_event(9, "untagged owner-direct event"); + let rules = vec![make_rule( + "owner-direct", + ChannelScope::All("all".into()), + vec![9], + false, + None, + Some("owner-direct"), + )]; + + let matched = match_event( + &event, + any_channel(), + &rules, + "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef", + ) + .await + .expect("untagged owner-direct event must remain eligible"); + assert_eq!(matched.prompt_tag, "owner-direct"); + } + #[tokio::test] async fn test_match_event_no_match() { let event = make_event(1, "hello"); diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 811253e4ac..4caf825ddf 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -38,7 +38,7 @@ use futures_util::FutureExt; use nostr::{PublicKey, ToBech32}; use pool::{ AgentPool, ControlSignal, IdleSwitchResult, OwnedAgent, PromptContext, PromptOutcome, - PromptResult, PromptSource, SessionState, TimeoutKind, + PromptResult, PromptSource, SessionState, TimeoutKind, PUBLICATION_VERIFICATION_ERROR_PREFIX, }; use pool_lifecycle::PoolLifecycle; use queue::{CancelReason, EventQueue, FlushBatch, QueuedEvent, ThreadTags}; @@ -3129,6 +3129,13 @@ fn is_auth_error(error: &acp::AcpError) -> bool { message.contains("Re-authenticate") || message.contains("API Error: 401") } +/// Publication verification failures are terminal for the batch. Retrying a +/// turn after its agent may already have published would risk duplicate user +/// replies; instead, surface an explicit failure asking the caller to retry. +fn is_publication_verification_error(error: &acp::AcpError) -> bool { + matches!(error, acp::AcpError::Protocol(message) if message.starts_with(PUBLICATION_VERIFICATION_ERROR_PREFIX)) +} + /// 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 @@ -3249,6 +3256,16 @@ fn handle_prompt_result( } else { hard_timeout_fate_suffix = Some(" — requeued for retry (recently active)"); } + } else if matches!(&result.outcome, PromptOutcome::Error(e) if is_publication_verification_error(e)) + { + tracing::error!( + channel_id = %batch.channel_id, + events = batch.events.len(), + "dead-lettering batch immediately — reply publication could not be verified" + ); + let content = "⚠️ I couldn't verify that my reply was delivered, so I stopped to avoid sending a duplicate. Please re-send the request." + .to_string(); + spawn_failure_notice(rest_client, &batch, content); } else if matches!(&result.outcome, PromptOutcome::Error(e) if is_auth_error(e)) { // Auth errors are non-retryable: the token won't self-repair // between retries, so requeueing only wastes attempt slots and @@ -6341,6 +6358,20 @@ mod error_outcome_emission_tests { ); } + #[test] + fn publication_verification_error_is_terminal() { + let error = AcpError::Protocol(format!( + "{PUBLICATION_VERIFICATION_ERROR_PREFIX} no signed kind:9 reply" + )); + assert!(is_publication_verification_error(&error)); + } + + #[test] + fn unrelated_protocol_error_can_still_retry() { + let error = AcpError::Protocol("unexpected ACP response".to_string()); + assert!(!is_publication_verification_error(&error)); + } + #[test] fn is_auth_error_rejects_transport_errors() { let io = acp::AcpError::Io(std::io::Error::other("pipe broke")); diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index ddc0330d9f..e05f5f35df 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -46,6 +46,10 @@ use crate::relay::{ChannelInfo, RestClient}; /// the turn as "recently active" (eligible for requeue instead of dead-letter). const RECENT_ACTIVITY_WINDOW: Duration = Duration::from_secs(60); +/// Stable marker used by the main loop to dead-letter an unverified reply +/// without retrying and risking a duplicate publication. +pub const PUBLICATION_VERIFICATION_ERROR_PREFIX: &str = "publication verification failed:"; + // FlushBatch and BatchEvent derive Clone (added in queue.rs) so we can store // a recoverable copy in TaskMeta for panic recovery in Queue mode. @@ -2105,6 +2109,42 @@ pub async fn run_prompt_task( "control signal arrived but turn already completed — treating as success" ); } + // The ACP response was consumed by the race, but it is + // still a completed channel turn. Preserve the same + // publication contract as the normal success arm. + // Explicit Cancel is the sole exception: its contract + // deliberately drops the caller's batch. + if !matches!(control_signal, ControlSignal::Cancel) { + if let Some(batch) = batch.as_ref() { + if let Err(error) = verify_reply_publication(&ctx, batch).await { + tracing::error!( + target: "pool::publication", + channel_id = %batch.channel_id, + events = batch.events.len(), + "race-completed ACP turn has no observed signed reply: {error}" + ); + let usage = agent.acp.take_turn_usage(); + publish_agent_turn_metric( + &ctx, + usage, + observer_channel_id, + &session_id, + &turn_id, + Some(buzz_core::agent_turn_metric::StopReason::Error), + ) + .await; + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(AcpError::Protocol(error)), + requeue_batch_if_queue(&ctx, Some(batch.clone())), + ); + return; + } + } + } apply_completed_before_control_signal( &mut agent.state, &source, @@ -2139,6 +2179,42 @@ pub async fn run_prompt_task( Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); + // ACP only acknowledges that the agent finished its turn; its + // response carries no user-visible text. A channel turn is not + // complete until the relay contains the agent's signed reply to + // this batch. Without this gate an agent can return `ok` after + // silently skipping `buzz messages send`, and the queue will + // permanently drop the caller's request. + if let Some(batch) = batch.as_ref() { + if let Err(error) = verify_reply_publication(&ctx, batch).await { + tracing::error!( + target: "pool::publication", + channel_id = %batch.channel_id, + events = batch.events.len(), + "ACP turn ended without an observed signed reply: {error}" + ); + let usage = agent.acp.take_turn_usage(); + publish_agent_turn_metric( + &ctx, + usage, + observer_channel_id, + &session_id, + &turn_id, + Some(buzz_core::agent_turn_metric::StopReason::Error), + ) + .await; + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(AcpError::Protocol(error)), + requeue_batch_if_queue(&ctx, Some(batch.clone())), + ); + return; + } + } + let should_rotate = matches!( stop_reason, StopReason::MaxTokens | StopReason::MaxTurnRequests @@ -2359,6 +2435,137 @@ pub async fn run_prompt_task( // _reaction_guard drops here → spawns clear_reactions for all exit paths. } +/// Confirm that a successful channel turn actually produced a signed reply. +/// ACP only returns a stop reason, so this relay check prevents a silent `ok` +/// from acknowledging a caller whose request was never published. +async fn verify_reply_publication(ctx: &PromptContext, batch: &FlushBatch) -> Result<(), String> { + use nostr::{Alphabet, Kind, SingleLetterTag}; + + let channel_info = ctx.channel_info.resolve(batch.channel_id).await; + if !batch_requires_reply( + batch, + &ctx.agent_keys.public_key().to_hex(), + channel_info.as_ref().map(|info| info.channel_type.as_str()), + ) { + tracing::debug!( + target: "pool::publication", + channel_id = %batch.channel_id, + "skipping reply verification for an unaddressed stream turn" + ); + return Ok(()); + } + + let anchors = publication_anchors(batch); + if anchors.is_empty() { + return Err(format!( + "{PUBLICATION_VERIFICATION_ERROR_PREFIX} batch has no valid reply anchor" + )); + } + let h_tag = SingleLetterTag::lowercase(Alphabet::H); + let channel = batch.channel_id.to_string(); + let filter = nostr::Filter::new() + .kind(Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE as u16)) + .author(ctx.agent_keys.public_key()) + .custom_tags(h_tag, [channel.as_str()]) + .limit(32); + let response = timeout(Duration::from_secs(10), ctx.rest_client.query(&[filter])) + .await + .map_err(|_| { + format!("{PUBLICATION_VERIFICATION_ERROR_PREFIX} reply publication query timed out") + })? + .map_err(|e| { + format!("{PUBLICATION_VERIFICATION_ERROR_PREFIX} reply publication query failed: {e}") + })?; + + if publication_proof_from_query(&response, batch.channel_id, &ctx.agent_keys, &anchors) { + Ok(()) + } else { + Err(format!( + "{PUBLICATION_VERIFICATION_ERROR_PREFIX} no signed kind:9 reply for this batch was found on the relay" + )) + } +} + +/// A missing publication is actionable only when the turn was directed at this +/// agent. Stream fan-out deliberately allows silence; DMs always require a +/// response. Unknown channel metadata fails closed as DM so a transient lookup +/// failure cannot silently discard a private request. +fn batch_requires_reply( + batch: &FlushBatch, + agent_pubkey_hex: &str, + channel_type: Option<&str>, +) -> bool { + if channel_type != Some("stream") { + return true; + } + + batch + .events + .iter() + .chain(batch.cancelled_events.iter()) + .any(|event| { + event.event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() >= 2 + && values[0] == "p" + && values[1].eq_ignore_ascii_case(agent_pubkey_hex) + }) + }) +} + +/// Return every event ID a correctly threaded reply to this batch may cite. +fn publication_anchors(batch: &FlushBatch) -> HashSet { + batch + .events + .iter() + .chain(batch.cancelled_events.iter()) + .flat_map(|event| { + let tags = crate::queue::parse_thread_tags(&event.event); + [ + Some(event.event.id.to_hex()), + tags.root_event_id, + tags.parent_event_id, + ] + }) + .flatten() + .filter(|id| id.len() == 64 && id.chars().all(|c| c.is_ascii_hexdigit())) + .collect() +} + +/// Validate queried events instead of trusting relay-side filters alone. +fn publication_proof_from_query( + response: &serde_json::Value, + channel_id: Uuid, + agent_keys: &nostr::Keys, + anchors: &HashSet, +) -> bool { + let Some(events) = response.as_array() else { + return false; + }; + let agent_pubkey = agent_keys.public_key(); + let channel = channel_id.to_string(); + events.iter().any(|raw| { + let Ok(event) = serde_json::from_value::(raw.clone()) else { + return false; + }; + if event.verify().is_err() + || event.kind != nostr::Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE as u16) + || event.pubkey != agent_pubkey + { + return false; + } + let has_channel = event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() >= 2 && values[0] == "h" && values[1] == channel + }); + let cites_anchor = event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() >= 2 && values[0] == "e" && anchors.contains(&values[1]) + }); + has_channel && cites_anchor + }) +} + /// Retry wrapper for context fetches: one retry with `CONTEXT_FETCH_RETRY_DELAY` /// on any `None` result. The closure is called twice at most. /// @@ -4027,8 +4234,10 @@ async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec) #[cfg(test)] mod tests { use super::*; + use crate::queue::BatchEvent; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; use serde_json::json; + use std::time::Instant; fn test_mcp_server() -> McpServer { McpServer { @@ -5475,6 +5684,130 @@ mod tests { } } + #[test] + fn publication_proof_requires_signed_threaded_reply_from_this_agent() { + let channel_id = Uuid::new_v4(); + let agent = Keys::generate(); + let batch = one_event_batch(channel_id); + let anchor = batch.events[0].event.id.to_hex(); + let reply = EventBuilder::new(Kind::Custom(9), "published") + .tags([ + Tag::parse(["h", channel_id.to_string().as_str()]).unwrap(), + Tag::parse(["e", anchor.as_str(), "", "reply"]).unwrap(), + ]) + .sign_with_keys(&agent) + .unwrap(); + let response = json!([reply]); + + assert!(publication_proof_from_query( + &response, + channel_id, + &agent, + &publication_anchors(&batch), + )); + } + + #[test] + fn publication_proof_rejects_unsigned_or_unanchored_events() { + let channel_id = Uuid::new_v4(); + let agent = Keys::generate(); + let batch = one_event_batch(channel_id); + let anchor = batch.events[0].event.id.to_hex(); + let reply = EventBuilder::new(Kind::Custom(9), "published") + .tags([ + Tag::parse(["h", channel_id.to_string().as_str()]).unwrap(), + Tag::parse(["e", anchor.as_str(), "", "reply"]).unwrap(), + ]) + .sign_with_keys(&agent) + .unwrap(); + let mut tampered = serde_json::to_value(reply).unwrap(); + tampered["content"] = json!("tampered after signing"); + + assert!(!publication_proof_from_query( + &json!([tampered]), + channel_id, + &agent, + &publication_anchors(&batch), + )); + } + + #[test] + fn publication_requirement_distinguishes_directed_turns_from_stream_fanout() { + let channel_id = Uuid::new_v4(); + let agent = Keys::generate(); + let agent_hex = agent.public_key().to_hex(); + let foreign = Keys::generate().public_key().to_hex(); + + let unaddressed = one_event_batch(channel_id); + assert!(!batch_requires_reply( + &unaddressed, + &agent_hex, + Some("stream") + )); + + let foreign_event = EventBuilder::new(Kind::Custom(9), "for another agent") + .tags([Tag::parse(["p", foreign.as_str()]).unwrap()]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let foreign_batch = FlushBatch { + channel_id, + events: vec![BatchEvent { + event: foreign_event, + prompt_tag: "owner-direct".into(), + received_at: Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + assert!(!batch_requires_reply( + &foreign_batch, + &agent_hex, + Some("stream") + )); + + let directed_event = EventBuilder::new(Kind::Custom(9), "for this agent") + .tags([Tag::parse(["p", agent_hex.as_str()]).unwrap()]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let directed_batch = FlushBatch { + channel_id, + events: vec![BatchEvent { + event: directed_event, + prompt_tag: "mentioned".into(), + received_at: Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + assert!(batch_requires_reply( + &directed_batch, + &agent_hex, + Some("stream") + )); + + assert!(batch_requires_reply(&unaddressed, &agent_hex, Some("dm"))); + assert!(batch_requires_reply(&unaddressed, &agent_hex, None)); + } + + #[test] + fn publication_requirement_includes_addressed_cancelled_events() { + let channel_id = Uuid::new_v4(); + let agent = Keys::generate(); + let agent_hex = agent.public_key().to_hex(); + let directed_event = EventBuilder::new(Kind::Custom(9), "steered request") + .tags([Tag::parse(["p", agent_hex.as_str()]).unwrap()]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let mut batch = one_event_batch(channel_id); + batch.cancelled_events.push(BatchEvent { + event: directed_event, + prompt_tag: "mentioned".into(), + received_at: Instant::now(), + }); + + assert!(batch_requires_reply(&batch, &agent_hex, Some("stream"))); + } + #[test] fn test_requeue_cancelled_batch_maps_control_signal_to_cancel_reason() { let cases = [