feat(cli): add bounded message reply wait

crates/buzz-cli/Cargo.toml

@@ -93,3 +93,6 @@ tempfile = "3"
 axum = { workspace = true }
 # `test-util` enables paused-time control for deterministic timeout tests
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] }
+# Local authenticated WebSocket relay fixtures for subscription tests
+tokio-tungstenite = { workspace = true }
+futures-util = { workspace = true }

crates/buzz-cli/TESTING.md

@@ -222,6 +222,11 @@ buzz messages thread --channel "$CHANNEL_ID" --event "$REPLY_ID" | jq .
 buzz messages thread \
   --link "buzz://message?channel=$CHANNEL_ID&id=$REPLY_ID&thread=$EVENT_ID" | jq .
 
+# messages wait — returns the next matching reply immediately, or null on timeout
+SINCE=$(date +%s)
+buzz messages wait --channel "$CHANNEL_ID" --event "$EVENT_ID" \
+  --author "<REPLY_AUTHOR_PUBKEY>" --since "$SINCE" --timeout 5 | jq .
+
 # messages search
 buzz messages search --query "Hello" | jq .
 buzz messages search --query "CLI test" --limit 5 | jq .
@@ -571,6 +576,7 @@ buzz channels delete --channel "$FORUM_ID" | jq .
 | 6 | `messages thread` | ☐ | |
 | 7 | `messages search` | ☐ | With limit |
 | 8 | `messages vote` | ☐ | Up and down |
+| 8a | `messages wait` | ☐ | WebSocket reply and null timeout |
 | 9 | `channels list` | ☐ | With visibility, member |
 | 10 | `channels get` | ☐ | |
 | 11 | `channels create` | ☐ | Stream and forum |

crates/buzz-cli/src/client.rs

@@ -569,6 +569,67 @@ impl BuzzClient {
         &self.relay_url
     }
 
+    /// Wait for the first event matching a Nostr subscription filter.
+    ///
+    /// The full connect, NIP-42 authentication, subscription, and receive
+    /// sequence is bounded by `timeout_duration`. A normal timeout returns
+    /// `Ok(None)`; transport and relay subscription failures remain errors so
+    /// callers can reconcile through the HTTP query path.
+    pub async fn wait_for_event(
+        &self,
+        filter: &serde_json::Value,
+        timeout_duration: Duration,
+    ) -> Result<Option<nostr::Event>, CliError> {
+        if timeout_duration.is_zero() {
+            return Ok(None);
+        }
+
+        let ws_url = to_ws_url(&self.relay_url);
+        let subscription_id = format!("buzz-cli-wait-{}", uuid::Uuid::new_v4());
+        let wait = async {
+            let mut connection = buzz_ws_client::NostrWsConnection::connect_authenticated(
+                &ws_url,
+                &self.keys,
+                self.auth_tag.as_ref(),
+            )
+            .await
+            .map_err(|error| CliError::Other(error.to_string()))?;
+
+            connection
+                .send_raw(&serde_json::json!(["REQ", subscription_id, filter]))
+                .await
+                .map_err(|error| CliError::Other(error.to_string()))?;
+
+            loop {
+                match connection.next_event(timeout_duration).await {
+                    Ok(buzz_ws_client::RelayMessage::Event {
+                        subscription_id: received_subscription_id,
+                        event,
+                    }) if received_subscription_id == subscription_id => {
+                        let _ = connection.disconnect().await;
+                        return Ok(Some(*event));
+                    }
+                    Ok(buzz_ws_client::RelayMessage::Closed {
+                        subscription_id: closed_subscription_id,
+                        message,
+                    }) if closed_subscription_id == subscription_id => {
+                        return Err(CliError::Other(format!(
+                            "relay closed subscription {closed_subscription_id}: {message}"
+                        )));
+                    }
+                    Ok(_) => {}
+                    Err(buzz_ws_client::WsClientError::Timeout) => return Ok(None),
+                    Err(error) => return Err(CliError::Other(error.to_string())),
+                }
+            }
+        };
+
+        match tokio::time::timeout(timeout_duration, wait).await {
+            Ok(result) => result,
+            Err(_) => Ok(None),
+        }
+    }
+
     /// Return the owner pubkey carried by the NIP-OA auth tag, if any.
     ///
     /// The auth tag is `["auth", owner_pubkey, conditions, sig]`; the
@@ -2332,6 +2393,103 @@ mod tests {
         normalize_events, BuzzClient,
     };
     use nostr::{EventBuilder, Keys, Kind, Tag};
+    use std::time::Duration;
+
+    #[tokio::test]
+    async fn wait_for_event_authenticates_subscribes_and_returns_live_match() {
+        use futures_util::{SinkExt, StreamExt};
+        use tokio_tungstenite::{accept_async, tungstenite::Message};
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let channel_id = "123e4567-e89b-12d3-a456-426614174000";
+        let root_event_id = "a".repeat(64);
+        let event = EventBuilder::new(Kind::Custom(9), "live reply")
+            .tags([
+                Tag::parse(["h", channel_id]).unwrap(),
+                Tag::parse(["e", root_event_id.as_str(), "", "root"]).unwrap(),
+            ])
+            .sign_with_keys(&Keys::generate())
+            .unwrap();
+        let expected_event_id = event.id;
+        let expected_filter = serde_json::json!({
+            "kinds": [9],
+            "#h": [channel_id],
+            "#e": [root_event_id],
+        });
+        let server_filter = expected_filter.clone();
+
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut websocket = accept_async(stream).await.unwrap();
+            websocket
+                .send(Message::Text(
+                    serde_json::json!(["AUTH", "test-challenge"])
+                        .to_string()
+                        .into(),
+                ))
+                .await
+                .unwrap();
+
+            let auth_text = websocket
+                .next()
+                .await
+                .unwrap()
+                .unwrap()
+                .into_text()
+                .unwrap();
+            let auth: serde_json::Value = serde_json::from_str(&auth_text).unwrap();
+            let auth_event_id = auth[1]["id"].as_str().unwrap();
+            websocket
+                .send(Message::Text(
+                    serde_json::json!(["OK", auth_event_id, true, ""])
+                        .to_string()
+                        .into(),
+                ))
+                .await
+                .unwrap();
+
+            let request_text = websocket
+                .next()
+                .await
+                .unwrap()
+                .unwrap()
+                .into_text()
+                .unwrap();
+            let request: serde_json::Value = serde_json::from_str(&request_text).unwrap();
+            assert_eq!(request[0], "REQ");
+            assert_eq!(request[2], server_filter);
+            let subscription_id = request[1].as_str().unwrap();
+            websocket
+                .send(Message::Text(
+                    serde_json::json!(["EOSE", subscription_id])
+                        .to_string()
+                        .into(),
+                ))
+                .await
+                .unwrap();
+            websocket
+                .send(Message::Text(
+                    serde_json::json!(["EVENT", subscription_id, event])
+                        .to_string()
+                        .into(),
+                ))
+                .await
+                .unwrap();
+        });
+
+        let client =
+            BuzzClient::new(format!("http://{address}"), Keys::generate(), None, None).unwrap();
+        let received = client
+            .wait_for_event(&expected_filter, Duration::from_secs(2))
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(received.id, expected_event_id);
+        assert_eq!(received.content, "live reply");
+        server.await.unwrap();
+    }
 
     #[test]
     fn normalize_events_preserves_the_complete_signed_event_shape() {

crates/buzz-cli/src/commands/messages.rs

@@ -1,5 +1,6 @@
 use buzz_sdk::{DeleteMessageOptions, DiffMeta, ThreadRef, VoteDirection};
 use nostr::PublicKey;
+use std::time::Duration;
 use uuid::Uuid;
 
 use crate::client::{normalize_events, normalize_write_response, BuzzClient};
@@ -464,6 +465,68 @@ pub async fn cmd_get_thread(
     Ok(())
 }
 
+fn build_wait_reply_filter(
+    channel_id: &str,
+    root_event_id: &str,
+    author_hex: &str,
+    since: i64,
+) -> serde_json::Value {
+    serde_json::json!({
+        "kinds": [9, 40002],
+        "#h": [channel_id],
+        "#e": [root_event_id],
+        "authors": [author_hex],
+        "since": since,
+    })
+}
+
+fn format_wait_event(
+    event: nostr::Event,
+    format: &crate::OutputFormat,
+) -> Result<String, CliError> {
+    let raw = serde_json::to_value(event)
+        .map_err(|error| CliError::Other(format!("failed to serialize event: {error}")))?;
+    let normalized = normalize_events(&[raw]);
+    let formatted = format_events(&normalized, format);
+    let mut events: Vec<serde_json::Value> = serde_json::from_str(&formatted)
+        .map_err(|error| CliError::Other(format!("failed to format event: {error}")))?;
+    let event = events
+        .pop()
+        .ok_or_else(|| CliError::Other("relay returned an empty event".into()))?;
+
+    serde_json::to_string(&event)
+        .map_err(|error| CliError::Other(format!("failed to encode event output: {error}")))
+}
+
+pub async fn cmd_wait_for_reply(
+    client: &BuzzClient,
+    channel_id: &str,
+    root_event_id: &str,
+    author: &str,
+    since: i64,
+    timeout_seconds: u64,
+    format: &crate::OutputFormat,
+) -> Result<(), CliError> {
+    validate_uuid(channel_id)?;
+    validate_hex64(root_event_id)?;
+    if since < 0 {
+        return Err(CliError::Usage("--since must be zero or greater".into()));
+    }
+    let author_hex = PublicKey::parse(author)
+        .map_err(|_| CliError::Usage(format!("invalid --author pubkey: {author}")))?
+        .to_hex();
+    let filter = build_wait_reply_filter(channel_id, root_event_id, &author_hex, since);
+
+    match client
+        .wait_for_event(&filter, Duration::from_secs(timeout_seconds))
+        .await?
+    {
+        Some(event) => println!("{}", format_wait_event(event, format)?),
+        None => println!("null"),
+    }
+    Ok(())
+}
+
 pub async fn cmd_search(
     client: &BuzzClient,
     query: Option<&str>,
@@ -1031,6 +1094,13 @@ pub async fn dispatch(
             )
             .await
         }
+        MessagesCmd::Wait {
+            channel,
+            event,
+            author,
+            since,
+            timeout,
+        } => cmd_wait_for_reply(client, &channel, &event, &author, since, timeout, format).await,
         MessagesCmd::Search {
             query,
             author,
@@ -1056,16 +1126,16 @@ pub async fn dispatch(
 #[cfg(test)]
 mod tests {
     use super::{
-        channel_id_from_event, cmd_get_thread, event_mention_pubkeys, find_root_from_tags,
-        format_events, match_profiles_by_name, merge_message_mentions, missing_members,
-        normalize_explicit_mentions, parse_member_pubkeys, resolve_names_to_pubkeys,
-        resolve_thread_target, thread_ref_from_event, thread_ref_from_parent_tags, BuzzClient,
-        CliError, Uuid,
+        build_wait_reply_filter, channel_id_from_event, cmd_get_thread, event_mention_pubkeys,
+        find_root_from_tags, format_events, format_wait_event, match_profiles_by_name,
+        merge_message_mentions, missing_members, normalize_explicit_mentions, parse_member_pubkeys,
+        resolve_names_to_pubkeys, resolve_thread_target, thread_ref_from_event,
+        thread_ref_from_parent_tags, BuzzClient, CliError, Uuid,
     };
     use buzz_sdk::mentions::{
         extract_at_mentions_with_known, extract_at_names, match_names_to_profiles, MentionProfile,
     };
-    use nostr::Keys;
+    use nostr::{EventBuilder, Keys, Kind};
     use serde_json::json;
 
     const ID_A: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
@@ -1105,6 +1175,35 @@ mod tests {
         );
     }
 
+    #[test]
+    fn wait_reply_filter_is_scoped_to_thread_author_and_time() {
+        let channel = "123e4567-e89b-12d3-a456-426614174000";
+        let filter = build_wait_reply_filter(channel, ID_A, PK_VALID_A, 1_777_777_777);
+
+        assert_eq!(filter["kinds"], json!([9, 40002]));
+        assert_eq!(filter["#h"], json!([channel]));
+        assert_eq!(filter["#e"], json!([ID_A]));
+        assert_eq!(filter["authors"], json!([PK_VALID_A]));
+        assert_eq!(filter["since"], json!(1_777_777_777));
+    }
+
+    #[test]
+    fn wait_reply_uses_the_message_compact_contract() {
+        let event = EventBuilder::new(Kind::Custom(9), "live reply")
+            .sign_with_keys(&Keys::generate())
+            .unwrap();
+
+        let output: serde_json::Value =
+            serde_json::from_str(&format_wait_event(event, &crate::OutputFormat::Compact).unwrap())
+                .unwrap();
+
+        assert_eq!(output["content"], "live reply");
+        assert!(output.get("id").is_some());
+        assert!(output.get("created_at").is_some());
+        assert!(output.get("pubkey").is_none());
+        assert!(output.get("kind").is_none());
+    }
+
     #[tokio::test]
     async fn malformed_channel_is_rejected_before_thread_fetch() {
         let client =