fix(acp): verify replies only for addressed turns

crates/buzz-acp/src/filter.rs

@@ -371,6 +371,25 @@ pub async fn match_event(
     rules: &[SubscriptionRule],
     agent_pubkey_hex: &str,
 ) -> Option<MatchedRule> {
+    // 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");

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"));