feat(cli): read-only unread state (NIP-RS) (#8789)

## Summary

This change adds `buzz channels unread`. The command is read-only. It
shows how many unread messages each channel has. It uses the read state
that desktop and mobile already sync (NIP-RS). It never publishes an
event. It never marks anything read.

Output: `[{channel_id, name, last_read_at, unread_count, capped}]`. The
most unread channel is first.

## Design

The command has two parts.

**Part 1: read state** (`commands/read_state.rs`)

- The command queries your own `kind:30078` events with the `read-state`
tag.
- It verifies each signature. It rejects events from other authors.
- It keeps the newest version of each `d` tag.
- It decrypts each slot with NIP-44 to yourself. It applies desktop's
validation rules.
- It merges all slots with `max`.
- It ignores `ov_*` keys. Desktop and mobile do not read them either.

**Part 2: counting** (`commands/channel_unread.rs`)

The rules follow desktop's `communityUnreadObserver.ts`.

- Without `--channel`, the command uses your non-archived channel
memberships.
- It reads message kinds 9, 40002, 45001, and 45003. For a DM it also
reads huddle starts (48100).
- It reads only events after the channel marker (`since = marker + 1`).
- Your own messages do not count.
- A message is read when its time is at or below the greatest of three
markers: the channel marker, its `msg:<id>` marker, and its
`thread:<root>` marker. A broadcast reply ignores the thread marker.

**Bounds and signals**

- At most 200 channels per run. `--limit` sets the number. The maximum
is 200.
- At most 1000 events per channel. A channel at that cap shows
`"capped": true`. Its count is a lower bound.
- If you have more channels than `--limit`, the extra channels are not
counted. Stderr shows
`{"warning":"channels_truncated","counted":N,"dropped":M}`.
- More than `--limit` `--channel` values is an error. The command never
drops a named channel.
- An unknown named channel exits with code 1.

**Failure rules**

- A slot that cannot be verified, decrypted, or decoded fails the
command with exit 4. Desktop skips such a slot. A skip would make the
count too high, with no sign of it.
- A slot with a newer `v` is skipped. Stderr shows
`read_state_slots_skipped`.

**Deviations from desktop**

- The command has no 7-day fetch horizon. It reads all slots. This can
only add markers.
- It does not apply desktop's notification gate. That gate uses
desktop-local data (followed threads, muted channels). So a count can be
higher than the desktop badge.
- It does not read mobile's `activity:` markers. Desktop does not read
them.

The README describes the output and the gotchas. The stable-inventory
test lists the new subcommand. This PR does not change `desktop/`
(deprecated; see #8840).

## Testing

**What ran**

- `cargo test -p buzz-cli`: 510 passed.
- `cargo clippy -p buzz-cli --all-targets -- -D warnings`: clean.
- `cargo fmt --all --check`: clean.
- `just file-size-check`: passed.

**Unit tests**

The tests run the production functions against a fake axum relay. The
fake relay applies the real filters (`#h`, `#d`, `#p`, `#t`, `since`,
`limit`).

- Merge: max across slots, newest version wins, bad values dropped,
`ov_*` ignored.
- Failures: bad JSON, wrong shape, bad `client_id`, wrong author, bad
signature, and undecryptable content all fail.
- Counting: own messages, `msg:` markers, `thread:` markers, broadcast
replies, archived channels, DM kinds, and sort order.
- Bounds: channel limit, event cap, `capped`, and explicit-channel
limits.
- Query shape: each channel query has `#h`, a bounded `limit`, and
`since = marker + 1`.
- A bad slot stops the command before any channel query.

**Interop evidence**

A node script runs the real `buzz` binary against a fake relay. Events
are made and encrypted by `nostr-tools`. This is an independent NIP-44
code base. The script compares the CLI with desktop's own TypeScript:

- `mergeReadStateEvents` gives the same frontier as `last_read_at` for
six channels.
- `fetchCommunityUnread` gives the same read/unread result as
`unread_count > 0` for each channel.

The six channels cover: read, unread, never read, own message only, two
slots with max-merge, and a `msg:` marker. The script prints `INTEROP
OK`. The script is not part of the repository.

**Mutation checks**

I broke each guard once and ran the tests. Each break made a test fail:

- Own-message exclusion.
- `msg:` marker, `thread:` marker, and broadcast rule.
- `since = marker + 1`.
- `capped` signal and channel limit.
- Explicit channels over the limit.
- Archived skip and DM kinds.
- Max-merge, newest version wins, and the `t` tag requirement.
- Signature check, author check, unknown `v`, and the 256-byte key
limit.

A mutation of the decrypt error path did not fail a test. The JSON step
fails the same way, so the break changes nothing. I removed that check
from the list.

## Manual test steps

You need a relay and an identity that has used desktop or mobile on that
relay.

1. Build the CLI:

   ```bash
   . ./bin/activate-hermit && cargo build -p buzz-cli
   ```

2. Set the identity and the relay:

   ```bash
   export BUZZ_RELAY_URL=<relay-url>
   export BUZZ_PRIVATE_KEY=<your-key>
   ```

3. Run the command:

   ```bash
   ./target/debug/buzz channels unread | jq .
   ```

Success: you see one object per channel. Each object has `channel_id`,
`name`, `last_read_at`, `unread_count`, and `capped`.

4. Compare with the app. Open desktop or mobile on the same identity. A
channel with a badge must show `unread_count` above 0. A channel with no
badge must show 0. Counts can be higher than the badge for muted
channels and followed threads. This is expected.

5. Read a channel in the app. Wait a few seconds. Run the command again.
Success: the count for that channel is 0 and `last_read_at` is higher.

6. Test the channel limit:

   ```bash
   ./target/debug/buzz channels unread --limit 1 | jq length
   ```

Success: the command prints `1`. If you have more than one channel,
stderr shows `channels_truncated`.

7. Test an unknown channel:

   ```bash
./target/debug/buzz channels unread --channel
00000000-0000-4000-8000-000000000000; echo $?
   ```

Success: the command prints a not-found error and the exit code is `1`.

8. Check that nothing was written. The command sends no events. The app
badges stay the same after the run.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: bmcanally <bmcanally@squareup.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

crates/buzz-cli/README.md

@@ -47,6 +47,7 @@ buzz messages send-diff --channel <uuid> --diff - --repo https://github.com/org/
 
 # Channels
 buzz channels list
+buzz channels unread                      # unread counts from your synced read state (read-only)
 buzz channels create --name "my-channel" --type stream --visibility open
 buzz channels join --channel <uuid>
 buzz channels topic --channel <uuid> --topic "New topic"
@@ -137,6 +138,7 @@ The table below mirrors that tree for readers who are not at a terminal.
 | | `vote` | Vote on a forum post |
 | `channels` | `list` | List channels |
 | | `get` | Get channel details |
+| | `unread` | Unread counts per channel (read-only) |
 | | `create` | Create a channel |
 | | `update` | Update channel name/description |
 | | `topic` | Set channel topic |

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

@@ -0,0 +1,716 @@
+//! `buzz channels unread` — per-channel unread counts from the synced NIP-RS
+//! read state. Read-only: it never publishes.
+//!
+//! Counting follows desktop's `communityUnreadObserver.ts`
+//! (`isUnreadExternalEvent`): an event is unread when it is not yours and is
+//! newer than the greatest of the channel frontier, its `msg:<id>` marker and,
+//! for a non-broadcast reply, its `thread:<root>` marker. Thread replies count.
+//! Desktop's notification gate (participated/followed threads, muted channels,
+//! hidden DMs) lives in desktop-local storage and is not applied here.
+
+use std::collections::HashMap;
+
+use buzz_core::kind::{
+    KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_HUDDLE_STARTED, KIND_STREAM_MESSAGE,
+    KIND_STREAM_MESSAGE_V2,
+};
+use serde_json::{json, Value};
+
+use super::read_state::{self, ReadState};
+use crate::client::{extract_d_tag, extract_tag_value, BuzzClient};
+use crate::error::CliError;
+use crate::validate::validate_uuid;
+
+const KIND_GROUP_METADATA: u32 = 39000;
+const KIND_GROUP_MEMBERS: u32 = 39002;
+const MEMBER_CHANNEL_LIMIT: u32 = 1000;
+
+/// Resource bounds of one run.
+pub(crate) struct Bounds {
+    /// Most channels counted in one run.
+    pub max_channels: usize,
+    /// Most events read per channel; a channel that reaches it reports `capped`.
+    pub events_per_channel: u32,
+}
+
+pub(crate) const BOUNDS: Bounds = Bounds {
+    max_channels: 200,
+    events_per_channel: 1000,
+};
+
+#[derive(Debug, PartialEq, Eq, serde::Serialize)]
+pub(crate) struct ChannelUnread {
+    pub channel_id: String,
+    pub name: String,
+    pub last_read_at: Option<u64>,
+    pub unread_count: u32,
+    /// `unread_count` is a lower bound: the per-channel event cap was reached.
+    pub capped: bool,
+}
+
+#[derive(Debug)]
+pub(crate) struct Outcome {
+    pub channels: Vec<ChannelUnread>,
+    /// Channels found but not counted because of `--limit`.
+    pub dropped: usize,
+    pub skipped_slots: Vec<read_state::SkippedSlot>,
+}
+
+struct Channel {
+    id: String,
+    name: String,
+    is_dm: bool,
+}
+
+fn message_kinds(is_dm: bool) -> Vec<u32> {
+    let mut kinds = vec![
+        KIND_STREAM_MESSAGE,
+        KIND_STREAM_MESSAGE_V2,
+        KIND_FORUM_POST,
+        KIND_FORUM_COMMENT,
+    ];
+    if is_dm {
+        kinds.push(KIND_HUDDLE_STARTED);
+    }
+    kinds
+}
+
+fn tags(event: &Value) -> impl Iterator<Item = Vec<&str>> {
+    event
+        .get("tags")
+        .and_then(Value::as_array)
+        .into_iter()
+        .flatten()
+        .filter_map(|tag| tag.as_array())
+        .map(|tag| tag.iter().map(|v| v.as_str().unwrap_or("")).collect())
+}
+
+/// The thread root a reply counts against, as desktop's `getThreadReference`
+/// and `isBroadcastReply` resolve it. A broadcast reply has none.
+fn thread_root(event: &Value) -> Option<String> {
+    let mut broadcast = false;
+    let mut root = None;
+    let mut reply = None;
+    for tag in tags(event) {
+        match tag.as_slice() {
+            ["broadcast", "1", ..] => broadcast = true,
+            ["e", id, _, "root", ..] => {
+                root.get_or_insert(*id);
+            }
+            ["e", id, _, "reply", ..] => reply = Some(*id),
+            _ => {}
+        }
+    }
+    if broadcast {
+        return None;
+    }
+    let parent = reply?;
+    Some(root.unwrap_or(parent).to_string())
+}
+
+fn is_unread(event: &Value, me: &str, channel_read_at: Option<u64>, state: &ReadState) -> bool {
+    if event
+        .get("pubkey")
+        .and_then(Value::as_str)
+        .is_some_and(|p| p.eq_ignore_ascii_case(me))
+    {
+        return false;
+    }
+    let id = event.get("id").and_then(Value::as_str).unwrap_or("");
+    let created_at = event.get("created_at").and_then(Value::as_u64).unwrap_or(0);
+    let read_at = [
+        channel_read_at,
+        state.frontier(&format!("msg:{id}")),
+        thread_root(event).and_then(|root| state.frontier(&format!("thread:{root}"))),
+    ]
+    .into_iter()
+    .flatten()
+    .max();
+    read_at.is_none_or(|read_at| created_at > read_at)
+}
+
+fn newest_by_d(events: Vec<Value>) -> HashMap<String, Value> {
+    let mut newest: HashMap<String, Value> = HashMap::new();
+    for event in events {
+        let d = extract_d_tag(&event);
+        if d.is_empty() {
+            continue;
+        }
+        let at = |e: &Value| e.get("created_at").and_then(Value::as_u64).unwrap_or(0);
+        if newest.get(&d).is_none_or(|head| at(&event) > at(head)) {
+            newest.insert(d, event);
+        }
+    }
+    newest
+}
+
+async fn member_channel_ids(client: &BuzzClient) -> Result<Vec<String>, CliError> {
+    let filter = json!({
+        "kinds": [KIND_GROUP_MEMBERS],
+        "#p": [client.keys().public_key().to_hex()],
+    });
+    let events = client
+        .query_all_bounded(filter, MEMBER_CHANNEL_LIMIT)
+        .await
+        .map_err(|e| match e {
+            CliError::Other(msg) => {
+                CliError::Other(format!("{msg}; name the channels with --channel instead"))
+            }
+            other => other,
+        })?;
+    let mut ids: Vec<String> = events
+        .iter()
+        .map(extract_d_tag)
+        .filter(|id| !id.is_empty())
+        .collect();
+    ids.sort();
+    ids.dedup();
+    Ok(ids)
+}
+
+async fn channel_metadata(
+    client: &BuzzClient,
+    ids: &[String],
+) -> Result<HashMap<String, Value>, CliError> {
+    if ids.is_empty() {
+        return Ok(HashMap::new());
+    }
+    let filter = json!({"kinds": [KIND_GROUP_METADATA], "#d": ids});
+    let events = client
+        .query_all_bounded(filter, MEMBER_CHANNEL_LIMIT)
+        .await?;
+    Ok(newest_by_d(events))
+}
+
+fn is_archived(metadata: &Value) -> bool {
+    tags(metadata).any(|t| t.as_slice() == ["archived", "true"])
+}
+
+async fn resolve_channels(
+    client: &BuzzClient,
+    requested: &[String],
+    limit: usize,
+) -> Result<(Vec<Channel>, usize), CliError> {
+    let explicit = !requested.is_empty();
+    let ids = if explicit {
+        requested.to_vec()
+    } else {
+        member_channel_ids(client).await?
+    };
+    let metadata = channel_metadata(client, &ids).await?;
+    let mut channels = Vec::new();
+    for id in ids {
+        match metadata.get(&id) {
+            Some(meta) if !explicit && is_archived(meta) => {}
+            Some(meta) => channels.push(Channel {
+                name: extract_tag_value(meta, "name"),
+                is_dm: extract_tag_value(meta, "t") == "dm",
+                id,
+            }),
+            None if explicit => {
+                return Err(CliError::NotFound(format!("channel {id} not found")));
+            }
+            None => channels.push(Channel {
+                name: String::new(),
+                is_dm: false,
+                id,
+            }),
+        }
+    }
+    channels.sort_by(|a, b| a.id.cmp(&b.id));
+    let dropped = channels.len().saturating_sub(limit);
+    channels.truncate(limit);
+    Ok((channels, dropped))
+}
+
+async fn count_channel(
+    client: &BuzzClient,
+    channel: &Channel,
+    state: &ReadState,
+    bounds: &Bounds,
+) -> Result<ChannelUnread, CliError> {
+    let last_read_at = state.frontier(&channel.id);
+    let mut filter = json!({
+        "kinds": message_kinds(channel.is_dm),
+        "#h": [channel.id],
+    });
+    if let Some(read_at) = last_read_at {
+        filter["since"] = json!(read_at + 1);
+    }
+    let events = client
+        .query_paginated(filter, bounds.events_per_channel)
+        .await?;
+    let capped = events.len() >= bounds.events_per_channel as usize;
+    let me = client.keys().public_key().to_hex();
+    let unread = events
+        .iter()
+        .filter(|e| is_unread(e, &me, last_read_at, state))
+        .count();
+    Ok(ChannelUnread {
+        channel_id: channel.id.clone(),
+        name: channel.name.clone(),
+        last_read_at,
+        unread_count: u32::try_from(unread).unwrap_or(u32::MAX),
+        capped,
+    })
+}
+
+/// Compute unread counts for the user's channels, or for `requested` ones.
+pub(crate) async fn unread(
+    client: &BuzzClient,
+    requested: &[String],
+    limit: Option<u32>,
+    bounds: &Bounds,
+) -> Result<Outcome, CliError> {
+    let limit = match limit {
+        None => bounds.max_channels,
+        Some(0) => return Err(CliError::Usage("--limit must be at least 1".into())),
+        Some(n) if n as usize > bounds.max_channels => {
+            return Err(CliError::Usage(format!(
+                "--limit must be at most {}",
+                bounds.max_channels
+            )));
+        }
+        Some(n) => n as usize,
+    };
+    let mut requested: Vec<String> = requested
+        .iter()
+        .map(|c| validate_uuid(c).map(|()| c.to_lowercase()))
+        .collect::<Result<_, _>>()?;
+    requested.sort();
+    requested.dedup();
+    if requested.len() > limit {
+        return Err(CliError::Usage(format!(
+            "{} channels given but at most {limit} can be counted per run",
+            requested.len()
+        )));
+    }
+
+    let state = read_state::fetch(client).await?;
+    let (channels, dropped) = resolve_channels(client, &requested, limit).await?;
+    let mut counted = Vec::with_capacity(channels.len());
+    for channel in &channels {
+        counted.push(count_channel(client, channel, &state, bounds).await?);
+    }
+    counted.sort_by(|a, b| {
+        b.unread_count
+            .cmp(&a.unread_count)
+            .then_with(|| a.name.cmp(&b.name))
+            .then_with(|| a.channel_id.cmp(&b.channel_id))
+    });
+    Ok(Outcome {
+        channels: counted,
+        dropped,
+        skipped_slots: state.skipped,
+    })
+}
+
+pub async fn cmd_unread(
+    client: &BuzzClient,
+    requested: &[String],
+    limit: Option<u32>,
+) -> Result<(), CliError> {
+    let outcome = unread(client, requested, limit, &BOUNDS).await?;
+    if outcome.dropped > 0 {
+        eprintln!(
+            "{}",
+            json!({
+                "warning": "channels_truncated",
+                "counted": outcome.channels.len(),
+                "dropped": outcome.dropped,
+                "message": "more channels than --limit; rerun with --channel to count the rest",
+            })
+        );
+    }
+    if !outcome.skipped_slots.is_empty() {
+        let slots: Vec<Value> = outcome
+            .skipped_slots
+            .iter()
+            .map(|s| json!({"d": s.d_tag, "v": s.version}))
+            .collect();
+        eprintln!(
+            "{}",
+            json!({
+                "warning": "read_state_slots_skipped",
+                "slots": slots,
+                "message": "written by a newer schema; counts may overstate unread",
+            })
+        );
+    }
+    let out = serde_json::to_string(&outcome.channels)
+        .map_err(|e| CliError::Other(format!("failed to encode unread counts: {e}")))?;
+    println!("{out}");
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::commands::read_state::tests::{slot_event, snapshot_descriptor, snapshot_envelope};
+    use axum::{body::Bytes, extract::State, routing::get, routing::post, Json, Router};
+    use nostr::Keys;
+    use std::sync::{Arc, Mutex};
+    use tokio::net::TcpListener;
+
+    #[derive(Default)]
+    struct Relay {
+        events: Vec<Value>,
+        filters: Vec<Value>,
+    }
+    type Shared = Arc<Mutex<Relay>>;
+
+    fn tag_values(filter: &Value, key: &str) -> Vec<String> {
+        filter[key]
+            .as_array()
+            .map(|a| {
+                a.iter()
+                    .filter_map(|v| v.as_str().map(str::to_string))
+                    .collect()
+            })
+            .unwrap_or_default()
+    }
+
+    fn matches(event: &Value, filter: &Value) -> bool {
+        let kinds: Vec<u64> = filter["kinds"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .filter_map(Value::as_u64)
+            .collect();
+        let has_tag = |name: &str, wanted: Vec<String>| {
+            wanted.is_empty()
+                || tags(event).any(|t| {
+                    t.first() == Some(&name) && wanted.iter().any(|w| t.get(1) == Some(&w.as_str()))
+                })
+        };
+        kinds.contains(&event["kind"].as_u64().unwrap())
+            && filter["authors"]
+                .as_array()
+                .is_none_or(|a| a.iter().any(|x| x == &event["pubkey"]))
+            && event["created_at"].as_u64().unwrap() >= filter["since"].as_u64().unwrap_or(0)
+            && has_tag("h", tag_values(filter, "#h"))
+            && has_tag("d", tag_values(filter, "#d"))
+            && has_tag("p", tag_values(filter, "#p"))
+            && has_tag("t", tag_values(filter, "#t"))
+    }
+
+    async fn fake_relay(relay: Shared) -> String {
+        let app = Router::new()
+            .route(
+                "/query",
+                post(|State(relay): State<Shared>, body: Bytes| async move {
+                    let filters: Vec<Value> = serde_json::from_slice(&body).unwrap();
+                    let [filter] = filters.as_slice() else {
+                        panic!("one filter expected")
+                    };
+                    let mut relay = relay.lock().unwrap();
+                    relay.filters.push(filter.clone());
+                    if filter.get("read_state_snapshot").is_some() {
+                        let own: Vec<Value> = relay
+                            .events
+                            .iter()
+                            .filter(|e| {
+                                e["kind"] == 30078 && filter["authors"] == json!([e["pubkey"]])
+                            })
+                            .cloned()
+                            .collect();
+                        let author = filter["authors"][0].as_str().unwrap();
+                        return Json(snapshot_envelope(author, own));
+                    }
+                    let mut hits: Vec<Value> = relay
+                        .events
+                        .iter()
+                        .filter(|e| matches(e, filter))
+                        .cloned()
+                        .collect();
+                    hits.sort_by_key(|e| std::cmp::Reverse(e["created_at"].as_u64().unwrap()));
+                    hits.truncate(filter["limit"].as_u64().unwrap() as usize);
+                    Json(json!(hits))
+                }),
+            )
+            .route(
+                "/info",
+                get(|| async { Json(json!({"read_state_snapshot": snapshot_descriptor()})) }),
+            )
+            .with_state(relay);
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        format!("http://{addr}")
+    }
+
+    struct World {
+        client: BuzzClient,
+        relay: Shared,
+        other: Keys,
+        counter: u64,
+    }
+
+    const GEN: &str = "11111111-1111-4111-8111-111111111111";
+    const DEV: &str = "22222222-2222-4222-8222-222222222222";
+    const OLD: &str = "33333333-3333-4333-8333-333333333333";
+    const NEW: &str = "44444444-4444-4444-8444-444444444444";
+
+    async fn world() -> World {
+        let relay = Shared::default();
+        let url = fake_relay(relay.clone()).await;
+        let client = BuzzClient::new(url, Keys::generate(), None, None).unwrap();
+        World {
+            client,
+            relay,
+            other: Keys::generate(),
+            counter: 0,
+        }
+    }
+
+    impl World {
+        fn push(&self, event: Value) {
+            self.relay.lock().unwrap().events.push(event);
+        }
+
+        fn signed(&self, keys: &Keys, kind: u16, tags: Vec<nostr::Tag>, at: u64) -> Value {
+            let event = nostr::EventBuilder::new(nostr::Kind::Custom(kind), "x")
+                .tags(tags)
+                .custom_created_at(nostr::Timestamp::from(at))
+                .sign_with_keys(keys)
+                .unwrap();
+            serde_json::to_value(event).unwrap()
+        }
+
+        fn member(&self, channel: &str) {
+            let me = self.client.keys().public_key();
+            let tags = vec![nostr::Tag::identifier(channel), nostr::Tag::public_key(me)];
+            self.push(self.signed(&Keys::generate(), 39002, tags, 1));
+        }
+
+        fn channel(&self, id: &str, name: &str, extra: &[(&str, &str)]) {
+            let mut tags = vec![
+                nostr::Tag::identifier(id),
+                nostr::Tag::parse(["name", name]).unwrap(),
+            ];
+            tags.extend(
+                extra
+                    .iter()
+                    .map(|(k, v)| nostr::Tag::parse([*k, *v]).unwrap()),
+            );
+            self.push(self.signed(&Keys::generate(), 39000, tags, 1));
+        }
+
+        fn say(&mut self, channel: &str, author: &Keys, at: u64, extra: &[Vec<&str>]) -> String {
+            self.counter += 1;
+            let mut tags = vec![nostr::Tag::parse(["h", channel]).unwrap()];
+            tags.extend(extra.iter().map(|t| nostr::Tag::parse(t.clone()).unwrap()));
+            tags.push(nostr::Tag::parse(["nonce", &self.counter.to_string()]).unwrap());
+            let event = self.signed(author, 9, tags, at);
+            let id = event["id"].as_str().unwrap().to_string();
+            self.push(event);
+            id
+        }
+
+        fn read_state(&self, slot: &str, contexts: Value, at: u64) {
+            let plaintext = json!({"v": 1, "client_id": slot, "contexts": contexts}).to_string();
+            self.push(slot_event(self.client.keys(), slot, &plaintext, at));
+        }
+
+        fn queries_for(&self, kind: u64) -> Vec<Value> {
+            self.relay
+                .lock()
+                .unwrap()
+                .filters
+                .iter()
+                .filter(|f| {
+                    f["kinds"]
+                        .as_array()
+                        .is_some_and(|k| k.contains(&json!(kind)))
+                })
+                .cloned()
+                .collect()
+        }
+    }
+
+    fn by_id<'a>(out: &'a Outcome, id: &str) -> &'a ChannelUnread {
+        out.channels.iter().find(|c| c.channel_id == id).unwrap()
+    }
+
+    #[tokio::test]
+    async fn counts_unread_with_desktop_rules() {
+        let mut w = world().await;
+        let me = w.client.keys().clone();
+        let other = w.other.clone();
+        for (id, name) in [(GEN, "general"), (DEV, "dev"), (OLD, "old")] {
+            w.member(id);
+            w.channel(id, name, &[]);
+        }
+        w.member(NEW);
+        w.channel(NEW, "new", &[("archived", "true")]);
+
+        // general: frontier 100 from slot a, raised to 150 by slot b (max-merge).
+        w.say(GEN, &other, 90, &[]);
+        w.say(GEN, &other, 120, &[]);
+        w.say(GEN, &other, 160, &[]);
+        w.say(GEN, &me, 170, &[]);
+        let marked = w.say(GEN, &other, 180, &[]);
+        let root = w.say(GEN, &other, 80, &[]);
+        let reply = w.say(
+            GEN,
+            &other,
+            200,
+            &[vec!["e", &root, "", "root"], vec!["e", &root, "", "reply"]],
+        );
+        let broadcast = w.say(
+            GEN,
+            &other,
+            203,
+            &[
+                vec!["e", &root, "", "root"],
+                vec!["e", &root, "", "reply"],
+                vec!["broadcast", "1"],
+            ],
+        );
+        w.read_state("a", json!({GEN: 100, DEV: 5}), 1);
+        w.read_state(
+            "b",
+            json!({GEN: 150, format!("msg:{marked}"): 180, format!("thread:{root}"): 205}),
+            2,
+        );
+        // dev: never read, two from others, one of mine.
+        w.say(DEV, &other, 10, &[]);
+        w.say(DEV, &other, 11, &[]);
+        w.say(DEV, &me, 12, &[]);
+        // old: fully read.
+        w.say(OLD, &other, 50, &[]);
+        w.read_state("c", json!({OLD: 50}), 3);
+        let _ = (reply, broadcast);
+
+        let out = unread(&w.client, &[], None, &BOUNDS).await.unwrap();
+
+        // 160 (> 150); 180 is covered by its msg marker; `reply` 200 is covered by
+        // thread:root 205; the broadcast reply 203 ignores the thread marker; 90/120
+        // predate the frontier; mine is excluded.
+        assert_eq!(by_id(&out, GEN).unread_count, 2);
+        assert_eq!(by_id(&out, GEN).last_read_at, Some(150));
+        assert_eq!(by_id(&out, DEV).unread_count, 2);
+        assert_eq!(by_id(&out, DEV).last_read_at, Some(5));
+        assert_eq!(by_id(&out, OLD).unread_count, 0);
+        assert!(
+            out.channels.iter().all(|c| c.channel_id != NEW),
+            "archived is skipped"
+        );
+        assert!(out.channels.iter().all(|c| !c.capped));
+        assert_eq!(out.dropped, 0);
+        let order: Vec<_> = out.channels.iter().map(|c| c.name.as_str()).collect();
+        assert_eq!(
+            order,
+            ["dev", "general", "old"],
+            "sorted by unread desc, then name"
+        );
+
+        // The production queries are bounded and scoped, not a firehose.
+        let reads = w.queries_for(9);
+        assert_eq!(reads.len(), 3);
+        for q in &reads {
+            assert!(q["limit"].as_u64().unwrap() <= u64::from(BOUNDS.events_per_channel));
+            assert!(q["#h"].as_array().is_some_and(|h| h.len() == 1));
+        }
+        let general_q = reads.iter().find(|q| q["#h"] == json!([GEN])).unwrap();
+        assert_eq!(
+            general_q["since"],
+            json!(151),
+            "reads start after the frontier"
+        );
+        let dev_q = reads.iter().find(|q| q["#h"] == json!([DEV])).unwrap();
+        assert_eq!(dev_q["since"], json!(6));
+        let state_qs = w.queries_for(30078);
+        assert_eq!(state_qs.len(), 1, "read state is one atomic snapshot");
+        assert_eq!(state_qs[0]["read_state_snapshot"], json!(1));
+        for paging in ["#t", "since", "until", "limit"] {
+            assert!(state_qs[0].get(paging).is_none(), "{paging}");
+        }
+    }
+
+    #[tokio::test]
+    async fn bounds_channels_and_events_and_says_so() {
+        let mut w = world().await;
+        let other = w.other.clone();
+        for id in [GEN, DEV, OLD] {
+            w.member(id);
+            w.channel(id, id, &[]);
+            for at in 1..=4 {
+                w.say(id, &other, at, &[]);
+            }
+        }
+        let bounds = Bounds {
+            max_channels: 2,
+            events_per_channel: 3,
+        };
+
+        let out = unread(&w.client, &[], Some(2), &bounds).await.unwrap();
+        assert_eq!((out.channels.len(), out.dropped), (2, 1));
+        assert!(out.channels.iter().all(|c| c.capped && c.unread_count == 3));
+        assert_eq!(
+            w.queries_for(9).len(),
+            2,
+            "the dropped channel is never queried"
+        );
+
+        let err = unread(&w.client, &[], Some(3), &bounds).await.unwrap_err();
+        assert!(matches!(err, CliError::Usage(_)));
+        let err = unread(&w.client, &[], Some(0), &bounds).await.unwrap_err();
+        assert!(matches!(err, CliError::Usage(_)));
+        let all = [GEN, DEV, OLD].map(String::from);
+        let err = unread(&w.client, &all, None, &bounds).await.unwrap_err();
+        assert!(
+            matches!(err, CliError::Usage(_)),
+            "explicit channels are never silently dropped"
+        );
+    }
+
+    #[tokio::test]
+    async fn explicit_channels_are_validated_and_must_exist() {
+        let mut w = world().await;
+        let other = w.other.clone();
+        w.channel(GEN, "general", &[("archived", "true"), ("t", "dm")]);
+        w.say(GEN, &other, 5, &[]);
+
+        let out = unread(&w.client, &[GEN.to_uppercase(), GEN.into()], None, &BOUNDS)
+            .await
+            .unwrap();
+        assert_eq!(
+            out.channels.len(),
+            1,
+            "case-insensitive dedupe; archived is allowed when named"
+        );
+        assert_eq!(out.channels[0].unread_count, 1);
+        let queried = w.queries_for(9);
+        assert!(
+            queried[0]["kinds"]
+                .as_array()
+                .unwrap()
+                .contains(&json!(48100)),
+            "dm counts huddle starts"
+        );
+
+        let err = unread(&w.client, &["nope".into()], None, &BOUNDS)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, CliError::Usage(_)));
+        let err = unread(&w.client, &[DEV.into()], None, &BOUNDS)
+            .await
+            .unwrap_err();
+        assert!(matches!(err, CliError::NotFound(_)));
+    }
+
+    #[tokio::test]
+    async fn an_unreadable_slot_fails_the_command_before_any_count() {
+        let w = world().await;
+        w.member(GEN);
+        w.channel(GEN, "general", &[]);
+        w.read_state("a", json!({GEN: 1}), 1);
+        w.push(slot_event(w.client.keys(), "b", "garbage", 2));
+
+        let err = unread(&w.client, &[], None, &BOUNDS).await.unwrap_err();
+        assert!(err.to_string().contains("unreadable"), "{err}");
+        assert!(w.queries_for(9).is_empty());
+        assert!(w.queries_for(39002).is_empty());
+    }
+}

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

@@ -1953,6 +1953,9 @@ pub async fn dispatch(
             let vis_str = visibility.as_ref().map(|v| v.to_string());
             cmd_list_channels(client, vis_str.as_deref(), Some(member), limit, format).await
         }
+        ChannelsCmd::Unread { channel, limit } => {
+            crate::commands::channel_unread::cmd_unread(client, &channel, limit).await
+        }
         ChannelsCmd::Get { channel } => cmd_get_channel(client, &channel).await,
         ChannelsCmd::Search {
             query,

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

@@ -1,5 +1,6 @@
 pub mod agents;
 pub mod channel_templates;
+pub mod channel_unread;
 pub mod channels;
 pub mod dms;
 pub mod emoji;
@@ -20,6 +21,7 @@ pub mod pr;
 pub mod project_channel;
 pub mod projects;
 pub mod reactions;
+pub(crate) mod read_state;
 mod repo_default_branch;
 pub mod repos;
 pub mod social;

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

@@ -0,0 +1,520 @@
+//! Read-only view of the user's own NIP-RS read state (kind:30078,
+//! `d=read-state:<slot>`, NIP-44 encrypted to self).
+//!
+//! Frontiers are decoded and max-merged the way desktop's
+//! `readStateFormat.ts` / `readStateSnapshot.ts` do. Unlike desktop, which
+//! drops a slot it cannot decode and carries on, a slot this CLI cannot read
+//! is an error: an unread count built on a partial merge would overstate
+//! unread, never understate it, and the user could not tell.
+
+use std::collections::HashMap;
+
+use buzz_core::kind::KIND_READ_STATE;
+use nostr::nips::nip44;
+use nostr::Event;
+use serde_json::{json, Value};
+
+use crate::client::BuzzClient;
+use crate::error::CliError;
+
+const D_TAG_PREFIX: &str = "read-state:";
+const T_TAG: &str = "read-state";
+const MAX_CONTEXTS: usize = 10_000;
+const MAX_CONTEXT_KEY_BYTES: usize = 256;
+const MAX_SLOT_ID_LEN: usize = 64;
+const MAX_CLIENT_ID_UTF16_UNITS: usize = 64;
+const SNAPSHOT_VERSION: u64 = 1;
+
+/// A slot that was valid but written by a newer schema, so left out of the merge.
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) struct SkippedSlot {
+    pub d_tag: String,
+    pub version: i64,
+}
+
+/// Max-merged frontiers of every readable slot.
+#[derive(Debug, Default)]
+pub(crate) struct ReadState {
+    frontiers: HashMap<String, u64>,
+    pub skipped: Vec<SkippedSlot>,
+}
+
+impl ReadState {
+    /// The merged read marker of a context, in unix seconds.
+    pub(crate) fn frontier(&self, context: &str) -> Option<u64> {
+        self.frontiers.get(context).copied()
+    }
+}
+
+/// Load and merge every read-state slot of the signed-in identity.
+///
+/// Uses the relay's atomic snapshot (NIP-RS, Buzz Atomic Snapshot Extension):
+/// paginating by cursor cannot see a slot another client replaces mid-load, so
+/// a relay without the extension is an error rather than a partial read.
+pub(crate) async fn fetch(client: &BuzzClient) -> Result<ReadState, CliError> {
+    let community_id = discover_snapshot(client).await?;
+    let me = client.keys().public_key().to_hex();
+    let filter = json!({
+        "kinds": [KIND_READ_STATE],
+        "authors": [me],
+        "read_state_snapshot": SNAPSHOT_VERSION,
+    });
+    let raw = client.query(&filter).await?;
+    let events = snapshot_events(&raw, &community_id, &me)?;
+    merge(client.keys(), events)
+}
+
+fn cannot_prove_complete(detail: impl std::fmt::Display) -> CliError {
+    CliError::Other(format!(
+        "cannot prove your read state is complete: {detail}; refusing to report unread counts"
+    ))
+}
+
+/// The community id a relay advertises for its atomic read-state snapshot.
+async fn discover_snapshot(client: &BuzzClient) -> Result<String, CliError> {
+    let raw = client.get_public("/info").await?;
+    let info: Value = serde_json::from_str(&raw)
+        .map_err(|e| CliError::Other(format!("invalid NIP-11 response: {e}")))?;
+    let descriptor = info
+        .get("read_state_snapshot")
+        .filter(|d| d.is_object())
+        .ok_or_else(|| {
+            cannot_prove_complete("this relay does not offer the atomic read_state_snapshot")
+        })?;
+    if descriptor.get("version").and_then(Value::as_u64) != Some(SNAPSHOT_VERSION) {
+        return Err(cannot_prove_complete(
+            "this relay offers an unsupported read_state_snapshot version",
+        ));
+    }
+    descriptor
+        .get("community_id")
+        .and_then(Value::as_str)
+        .map(str::to_owned)
+        .ok_or_else(|| cannot_prove_complete("the relay advertised no community_id"))
+}
+
+/// Validate the snapshot envelope and return its events. Anything but a
+/// complete, version-1 envelope for this identity and community is rejected,
+/// including the plain event array an older relay returns when it ignores the
+/// extension.
+fn snapshot_events(raw: &str, community_id: &str, me: &str) -> Result<Vec<Value>, CliError> {
+    let mut envelope: Value = serde_json::from_str(raw)
+        .map_err(|e| CliError::Other(format!("failed to parse query response: {e}")))?;
+    let field = |name: &str| envelope.get(name);
+    if field("read_state_snapshot").and_then(Value::as_u64) != Some(SNAPSHOT_VERSION)
+        || field("complete") != Some(&Value::Bool(true))
+    {
+        return Err(cannot_prove_complete(
+            "the relay did not return a complete read_state_snapshot",
+        ));
+    }
+    if field("community_id").and_then(Value::as_str) != Some(community_id)
+        || field("pubkey").and_then(Value::as_str) != Some(me)
+    {
+        return Err(cannot_prove_complete(
+            "the snapshot belongs to a different community or identity",
+        ));
+    }
+    match envelope.get_mut("events").map(Value::take) {
+        Some(Value::Array(events)) => Ok(events),
+        _ => Err(cannot_prove_complete("the snapshot has no events array")),
+    }
+}
+
+fn unreadable(event: &Event, detail: impl std::fmt::Display) -> CliError {
+    CliError::Other(format!(
+        "your read-state slot {} is unreadable ({detail}); refusing to report unread counts from a partial read state",
+        event.id.to_hex()
+    ))
+}
+
+fn single_d_tag(event: &Event) -> Option<&str> {
+    let mut d = event
+        .tags
+        .iter()
+        .filter(|t| t.as_slice().first().map(String::as_str) == Some("d"));
+    let tag = d.next()?;
+    if d.next().is_some() {
+        return None;
+    }
+    tag.as_slice().get(1).map(String::as_str)
+}
+
+fn is_slot_coordinate(event: &Event) -> Option<&str> {
+    let d = single_d_tag(event)?;
+    let slot = d.strip_prefix(D_TAG_PREFIX)?;
+    if slot.is_empty() || slot.len() > MAX_SLOT_ID_LEN || !slot.is_ascii() {
+        return None;
+    }
+    let t_count = event
+        .tags
+        .iter()
+        .filter(|t| matches!(t.as_slice(), [k, v, ..] if k == "t" && v == T_TAG))
+        .count();
+    (t_count == 1).then_some(d)
+}
+
+/// JS `Number.isInteger` within the uint32 range.
+fn uint32(value: &Value) -> Option<u64> {
+    let n = value.as_u64().or_else(|| {
+        let f = value.as_f64()?;
+        (f >= 0.0 && f.fract() == 0.0 && f <= u32::MAX as f64).then_some(f as u64)
+    })?;
+    (n <= u64::from(u32::MAX)).then_some(n)
+}
+
+/// Decode one blob. `Ok(Err(version))` is a valid blob of an unknown schema.
+fn decode(event: &Event, plaintext: &str) -> Result<Result<HashMap<String, u64>, i64>, CliError> {
+    let blob: Value = serde_json::from_str(plaintext).map_err(|e| unreadable(event, e))?;
+    let Some(blob) = blob.as_object() else {
+        return Err(unreadable(event, "not a JSON object"));
+    };
+    let version = blob
+        .get("v")
+        .and_then(Value::as_f64)
+        .filter(|v| v.fract() == 0.0);
+    match version {
+        Some(1.0) => {}
+        Some(other) => return Ok(Err(other as i64)),
+        None => return Err(unreadable(event, "missing or non-integer schema version")),
+    }
+    let client_id_ok = blob
+        .get("client_id")
+        .and_then(Value::as_str)
+        .is_some_and(|id| (1..=MAX_CLIENT_ID_UTF16_UNITS).contains(&id.encode_utf16().count()));
+    if !client_id_ok {
+        return Err(unreadable(event, "invalid client_id"));
+    }
+    let Some(contexts) = blob.get("contexts").and_then(Value::as_object) else {
+        return Err(unreadable(event, "contexts is not an object"));
+    };
+    if contexts.len() > MAX_CONTEXTS {
+        return Err(unreadable(event, "too many contexts"));
+    }
+    Ok(Ok(contexts
+        .iter()
+        .filter(|(key, _)| key.len() <= MAX_CONTEXT_KEY_BYTES)
+        .filter_map(|(key, value)| Some((key.clone(), uint32(value)?)))
+        .collect()))
+}
+
+/// Verify, decrypt and max-merge raw relay events. Events that are not read-state
+/// coordinates are ignored; a coordinate that cannot be read is an error.
+pub(crate) fn merge(keys: &nostr::Keys, raw: Vec<Value>) -> Result<ReadState, CliError> {
+    let me = keys.public_key();
+    let mut heads: HashMap<String, Event> = HashMap::new();
+    for value in raw {
+        let event: Event = serde_json::from_value(value)
+            .map_err(|e| CliError::Other(format!("relay returned a malformed event: {e}")))?;
+        if event.pubkey != me || event.kind.as_u16() as u32 != KIND_READ_STATE {
+            return Err(CliError::Other(format!(
+                "relay returned event {} that is not your read state",
+                event.id.to_hex()
+            )));
+        }
+        event.verify().map_err(|_| {
+            CliError::Other(format!(
+                "relay returned event {} with an invalid signature",
+                event.id.to_hex()
+            ))
+        })?;
+        let Some(d) = is_slot_coordinate(&event) else {
+            continue;
+        };
+        let d = d.to_string();
+        let newer = heads
+            .get(&d)
+            .is_none_or(|head| (event.created_at, head.id) > (head.created_at, event.id));
+        if newer {
+            heads.insert(d, event);
+        }
+    }
+
+    let mut state = ReadState::default();
+    let mut ordered: Vec<_> = heads.into_iter().collect();
+    ordered.sort_by(|a, b| a.0.cmp(&b.0));
+    for (d_tag, event) in ordered {
+        let plaintext = nip44::decrypt(keys.secret_key(), &me, &event.content)
+            .map_err(|e| unreadable(&event, e))?;
+        match decode(&event, &plaintext)? {
+            Ok(contexts) => {
+                for (key, ts) in contexts {
+                    let slot = state.frontiers.entry(key).or_insert(0);
+                    *slot = (*slot).max(ts);
+                }
+            }
+            Err(version) => state.skipped.push(SkippedSlot { d_tag, version }),
+        }
+    }
+    Ok(state)
+}
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use super::*;
+    use axum::{body::Bytes, extract::State, routing::get, routing::post, Json, Router};
+    use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
+    use std::sync::{Arc, Mutex};
+    use tokio::net::TcpListener;
+
+    pub(crate) const COMMUNITY: &str = "5f0c1b9e-2d0b-4f4e-9a7c-0d6f3f3f3f3f";
+
+    pub(crate) fn snapshot_envelope(pubkey: &str, events: Vec<Value>) -> Value {
+        json!({
+            "read_state_snapshot": 1,
+            "complete": true,
+            "community_id": COMMUNITY,
+            "pubkey": pubkey,
+            "snapshot_id": "0".repeat(64),
+            "events": events,
+        })
+    }
+
+    pub(crate) fn snapshot_descriptor() -> Value {
+        json!({"version": 1, "community_id": COMMUNITY, "max_events": 4096})
+    }
+
+    pub(crate) fn slot_event(keys: &Keys, slot: &str, plaintext: &str, at: u64) -> Value {
+        let content = nip44::encrypt(
+            keys.secret_key(),
+            &keys.public_key(),
+            plaintext,
+            nip44::Version::V2,
+        )
+        .unwrap();
+        let event = EventBuilder::new(Kind::Custom(KIND_READ_STATE as u16), content)
+            .tags([
+                Tag::identifier(format!("{D_TAG_PREFIX}{slot}")),
+                Tag::hashtag(T_TAG),
+            ])
+            .custom_created_at(Timestamp::from(at))
+            .sign_with_keys(keys)
+            .unwrap();
+        serde_json::to_value(event).unwrap()
+    }
+
+    fn blob(contexts: Value) -> String {
+        json!({"v": 1, "client_id": "dev", "contexts": contexts}).to_string()
+    }
+
+    #[test]
+    fn max_merges_every_slot_and_ignores_override_and_foreign_entries() {
+        let keys = Keys::generate();
+        let state = merge(
+            &keys,
+            vec![
+                slot_event(&keys, "a", &blob(json!({"c1": 10, "c2": 50, "ov_s:c1": 3, "esc:ov_c:x": 9})), 1),
+                slot_event(&keys, "b", &blob(json!({"c1": 20, "c2": 5, "c3": 1.0e3, "neg": -1, "k".repeat(257): 9, "big": 4294967296u64, "frac": 1.5, "str": "7"})), 2),
+            ],
+        )
+        .unwrap();
+        assert_eq!(state.frontier("c1"), Some(20));
+        assert_eq!(state.frontier("c2"), Some(50));
+        assert_eq!(state.frontier("c3"), Some(1000));
+        assert_eq!(state.frontier(&"k".repeat(257)), None);
+        for dropped in ["neg", "big", "frac", "str", "missing"] {
+            assert_eq!(state.frontier(dropped), None, "{dropped}");
+        }
+        assert!(state.skipped.is_empty());
+    }
+
+    #[test]
+    fn newest_version_of_a_coordinate_wins() {
+        let keys = Keys::generate();
+        let old = slot_event(&keys, "a", &blob(json!({"c1": 99})), 1);
+        let new = slot_event(&keys, "a", &blob(json!({"c1": 5})), 2);
+        let state = merge(&keys, vec![new, old]).unwrap();
+        assert_eq!(state.frontier("c1"), Some(5));
+    }
+
+    #[test]
+    fn an_unreadable_slot_is_an_error_not_an_empty_state() {
+        let keys = Keys::generate();
+        let good = slot_event(&keys, "a", &blob(json!({"c1": 1})), 1);
+        for bad in [
+            "not json",
+            "[]",
+            r#"{"client_id":"d","contexts":{}}"#,
+            r#"{"v":1,"client_id":"","contexts":{}}"#,
+            r#"{"v":1,"client_id":"d","contexts":[]}"#,
+            r#"{"v":"1","client_id":"d","contexts":{}}"#,
+        ] {
+            let err =
+                merge(&keys, vec![good.clone(), slot_event(&keys, "b", bad, 2)]).expect_err(bad);
+            assert!(err.to_string().contains("unreadable"), "{bad}: {err}");
+        }
+
+        let other = Keys::generate();
+        let foreign = slot_event(&other, "b", &blob(json!({})), 2);
+        let err = merge(&keys, vec![foreign]).unwrap_err();
+        assert!(err.to_string().contains("not your read state"), "{err}");
+
+        let mut tampered = slot_event(&keys, "b", &blob(json!({"c1": 1})), 2);
+        tampered["created_at"] = json!(3);
+        assert!(merge(&keys, vec![tampered]).is_err());
+
+        let mut undecryptable = slot_event(&keys, "b", &blob(json!({})), 2);
+        let wrong = slot_event(&other, "b", &blob(json!({})), 2);
+        undecryptable["content"] = wrong["content"].clone();
+        assert!(merge(&keys, vec![undecryptable]).is_err());
+    }
+
+    #[test]
+    fn a_newer_schema_is_reported_not_merged() {
+        let keys = Keys::generate();
+        let future = json!({"v": 2, "client_id": "d", "contexts": {"c1": 99}}).to_string();
+        let state = merge(&keys, vec![slot_event(&keys, "a", &future, 1)]).unwrap();
+        assert_eq!(state.frontier("c1"), None);
+        assert_eq!(
+            state.skipped,
+            vec![SkippedSlot {
+                d_tag: "read-state:a".into(),
+                version: 2
+            }]
+        );
+    }
+
+    #[test]
+    fn non_coordinates_are_ignored() {
+        let keys = Keys::generate();
+        let unrelated = EventBuilder::new(Kind::Custom(KIND_READ_STATE as u16), "x")
+            .tags([Tag::identifier("channel-sections"), Tag::hashtag(T_TAG)])
+            .sign_with_keys(&keys)
+            .unwrap();
+        let no_t_event = EventBuilder::new(Kind::Custom(KIND_READ_STATE as u16), "x")
+            .tags([Tag::identifier("read-state:a")])
+            .sign_with_keys(&keys)
+            .unwrap();
+        let state = merge(
+            &keys,
+            vec![
+                serde_json::to_value(unrelated).unwrap(),
+                serde_json::to_value(no_t_event).unwrap(),
+            ],
+        )
+        .unwrap();
+        assert!(state.frontiers.is_empty());
+    }
+
+    type Bodies = Arc<Mutex<Vec<Value>>>;
+
+    #[derive(Clone)]
+    struct Canned {
+        info: Value,
+        query: String,
+        bodies: Bodies,
+    }
+
+    async fn relay(keys: &Keys, info: Value, query: Value) -> (BuzzClient, Bodies) {
+        let canned = Canned {
+            info,
+            query: query.to_string(),
+            bodies: Bodies::default(),
+        };
+        let bodies = canned.bodies.clone();
+        let app = Router::new()
+            .route(
+                "/info",
+                get(|State(c): State<Canned>| async move { Json(c.info) }),
+            )
+            .route(
+                "/query",
+                post(|State(c): State<Canned>, body: Bytes| async move {
+                    c.bodies
+                        .lock()
+                        .unwrap()
+                        .push(serde_json::from_slice(&body).unwrap());
+                    c.query
+                }),
+            )
+            .with_state(canned);
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("http://{}", listener.local_addr().unwrap());
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        (
+            BuzzClient::new(url, keys.clone(), None, None).unwrap(),
+            bodies,
+        )
+    }
+
+    fn advertised() -> Value {
+        json!({"read_state_snapshot": snapshot_descriptor()})
+    }
+
+    #[tokio::test]
+    async fn fetch_loads_one_atomic_snapshot() {
+        let keys = Keys::generate();
+        let slots = vec![
+            slot_event(&keys, "a", &blob(json!({"c1": 10})), 1),
+            slot_event(&keys, "b", &blob(json!({"c1": 20})), 2),
+        ];
+        let (client, bodies) = relay(
+            &keys,
+            advertised(),
+            snapshot_envelope(&keys.public_key().to_hex(), slots),
+        )
+        .await;
+
+        let state = fetch(&client).await.unwrap();
+
+        assert_eq!(state.frontier("c1"), Some(20));
+        assert_eq!(
+            *bodies.lock().unwrap(),
+            vec![json!([{
+                "kinds": [KIND_READ_STATE],
+                "authors": [keys.public_key().to_hex()],
+                "read_state_snapshot": 1,
+            }])],
+            "one exact snapshot request: no tag, limit or cursor that could page"
+        );
+    }
+
+    #[tokio::test]
+    async fn fetch_refuses_a_load_it_cannot_prove_complete() {
+        let keys = Keys::generate();
+        let slot = slot_event(&keys, "a", &blob(json!({"c1": 10})), 1);
+        let good = snapshot_envelope(&keys.public_key().to_hex(), vec![slot.clone()]);
+        let with = |field: &str, value: Value| {
+            let mut envelope = good.clone();
+            envelope[field] = value;
+            envelope
+        };
+        let cases: Vec<(&str, Value, Value)> = vec![
+            ("no descriptor", json!({}), good.clone()),
+            (
+                "descriptor version",
+                json!({"read_state_snapshot": {"version": 2, "community_id": COMMUNITY}}),
+                good.clone(),
+            ),
+            (
+                "descriptor community",
+                json!({"read_state_snapshot": {"version": 1}}),
+                good.clone(),
+            ),
+            ("relay ignored the extension", advertised(), json!([slot])),
+            ("incomplete", advertised(), with("complete", json!(false))),
+            ("unmarked", advertised(), with("complete", Value::Null)),
+            (
+                "version",
+                advertised(),
+                with("read_state_snapshot", json!(2)),
+            ),
+            (
+                "community",
+                advertised(),
+                with("community_id", json!("other")),
+            ),
+            (
+                "identity",
+                advertised(),
+                with("pubkey", json!("f".repeat(64))),
+            ),
+            ("events", advertised(), with("events", Value::Null)),
+        ];
+        for (name, info, response) in cases {
+            let (client, _) = relay(&keys, info, response).await;
+            let err = fetch(&client).await.expect_err(name);
+            assert!(err.to_string().contains("cannot prove"), "{name}: {err}");
+        }
+    }
+}

crates/buzz-cli/src/lib.rs

@@ -664,6 +664,19 @@ pub enum ChannelsCmd {
         #[arg(long)]
         limit: Option<u32>,
     },
+    /// Unread message counts from your synced read state (read-only)
+    #[command(
+        after_help = "Examples:\n  buzz channels unread\n  buzz channels unread --channel <uuid> --channel <uuid>\n  buzz channels unread --limit 20\n\nOutput: [{channel_id, name, last_read_at, unread_count, capped}]. `capped: true` means unread_count is a lower bound. When more channels exist than --limit (at most 200), the rest are listed in a stderr warning count and not counted. Fails if a read-state slot cannot be decrypted or decoded."
+    )]
+    Unread {
+        /// Count only these channels (repeatable, at most 200). Default: every
+        /// non-archived channel you are a member of
+        #[arg(long)]
+        channel: Vec<String>,
+        /// Most channels to count [default: 200, max: 200]
+        #[arg(long)]
+        limit: Option<u32>,
+    },
     /// Get details for a single channel
     Get {
         /// Channel UUID
@@ -2618,6 +2631,7 @@ mod tests {
                 "set-add-policy",
                 "topic",
                 "unarchive",
+                "unread",
                 "update"
             ]
         );
@@ -2729,7 +2743,7 @@ mod tests {
         let expected: Vec<(&str, usize)> = vec![
             ("agents", 5),
             ("canvas", 4),
-            ("channels", 16),
+            ("channels", 17),
             ("dms", 4),
             ("emoji", 5),
             ("feed", 1),