diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 7f03a87e..7b7d6df6 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -16,7 +16,7 @@ use matrix_sdk::{ api::client::receipt::create_receipt::v3::ReceiptType, events::{ reaction::ReactionEventContent, - receipt::{ReceiptThread, ReceiptType as LocalReceiptType}, + receipt::ReceiptThread, relation::Annotation, room::message::{AddMentions, MessageType, RoomMessageEventContent}, }, @@ -175,98 +175,33 @@ pub fn room_label(room: &matrix_sdk::Room) -> String { /// Refuse to post into a room the agent hasn't caught up on. Returns /// `Some(error)` with a helpful hint when the room still has unread -/// content (the agent must `read_room` then `mark_read` the latest event -/// first), or `None` when the send may proceed. +/// notifications (the agent must `read_room` then `mark_read` the +/// latest event first), or `None` when the send may proceed. /// -/// Read-state is [`room_unread_state`] — the same predicate the wake path -/// and `get_loose_ends` use, so "caught up" here means exactly what those -/// surfaces mean. Reactions and `mark_read` are not gated — only -/// message-posting tools (`send_message`, `send_reply`, `send_dm`) so an -/// agent can't talk over messages it hasn't seen. -async fn unread_guard(client: &Client, room: &matrix_sdk::Room) -> Option { - room_unread_state(client, room).await?; +/// Read-state is the matrix unread-notification count, the same signal +/// the wake path and `get_loose_ends` use, so "caught up" here means +/// exactly what those surfaces mean. Reactions and `mark_read` are not +/// gated — only message-posting tools (`send_message`, `send_reply`, +/// `send_dm`) so an agent can't talk over messages it hasn't seen. +fn unread_guard(room: &matrix_sdk::Room) -> Option { + let count = room.unread_notification_counts().notification_count; + if count == 0 { + return None; + } let label = room_label(room); Some(DaemonResponse::error(format!( - "refusing to send: unread message(s) in {label}. \ + "refusing to send: {count} unread message(s) in {label}. \ use read_room to view them, then mark_read the latest event before \ sending so you don't talk over messages you haven't seen." ))) } -/// Fetch the single most recent timeline event in `room`, of any type -/// (redactions/state/reactions included — identity is what -/// [`room_unread_state`] needs, not content). `None` for a genuinely -/// empty room or on any request failure. -async fn latest_event( - client: &Client, - room: &matrix_sdk::Room, -) -> Option<(OwnedEventId, OwnedUserId, String)> { - use matrix_sdk::ruma::api::Direction; - use matrix_sdk::ruma::api::client::message::get_message_events; - let mut req = - get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward); - req.limit = matrix_sdk::ruma::UInt::from(1u32); - let resp = client.send(req).await.ok()?; - let raw = resp.chunk.first()?; - let ev = raw.deserialize().ok()?; - let body = extract_body(&ev); - Some((ev.event_id().to_owned(), ev.sender().to_owned(), body)) -} - -/// Whether `room` carries content the agent hasn't caught up on. -/// `Some((event_id, sender, body))` for the latest event when it's -/// genuinely unread; `None` when the room is caught up, empty, or its -/// latest event is self-authored. -/// -/// Diffs the room's own local state against the agent's own `m.read` -/// receipt instead of `unread_notification_counts()`. That count is -/// server-computed from Matrix **push rules** — a DM or a plain message -/// that doesn't match any push rule (no keyword, no @mention) can leave a -/// genuinely-unread event reporting `notification_count: 0` forever, so -/// the agent never wakes for it (confirmed live on a prior investigation: -/// the daemon had the event in its local store the whole time and still -/// reported zero — the count was never the right signal to read). -/// Comparing the latest local event id against the own-account read -/// receipt needs no push-rule semantics at all. -/// -/// Self-authored latest events are never "unread" — after a daemon -/// rebuild the read receipt can lag behind the agent's own just-sent -/// message, which must not self-wake it. -async fn room_unread_state( - client: &Client, - room: &matrix_sdk::Room, -) -> Option<(OwnedEventId, OwnedUserId, String)> { - let own_user_id = client.user_id()?; - let (event_id, sender, body) = latest_event(client, room).await?; - if sender.as_str() == own_user_id.as_str() { - return None; - } - // `load_user_receipt` takes `ruma::events::receipt::ReceiptType` - // (imported here as `LocalReceiptType`), a distinct type from the - // `ReceiptType` this module already imports for `send_single_receipt` - // (`ruma::api::client::receipt::create_receipt::v3::ReceiptType`) — two - // same-named enums from different ruma crates, not interchangeable. - let read = room - .load_user_receipt( - LocalReceiptType::Read, - ReceiptThread::Unthreaded, - own_user_id, - ) - .await - .ok() - .flatten(); - if read.is_some_and(|(id, _)| id == event_id) { - return None; - } - Some((event_id, sender, body)) -} - pub async fn send_message(client: &Client, room_ref: &str, body: &str) -> DaemonResponse { let room = match resolve_room(client, room_ref).await { Ok(r) => r, Err(e) => return e, }; - if let Some(reject) = unread_guard(client, &room).await { + if let Some(reject) = unread_guard(&room) { return reject; } let content = RoomMessageEventContent::text_markdown(body); @@ -307,7 +242,7 @@ pub async fn send_dm(client: &Client, user_id: &str, body: &str) -> DaemonRespon Ok(r) => r, Err(e) => return e, }; - if let Some(reject) = unread_guard(client, &room).await { + if let Some(reject) = unread_guard(&room) { return reject; } let content = RoomMessageEventContent::text_markdown(body); @@ -393,7 +328,7 @@ pub async fn send_file( Ok(r) => r, Err(e) => return e, }; - if let Some(reject) = unread_guard(client, &room).await { + if let Some(reject) = unread_guard(&room) { return reject; } upload_attachment(&room, path, caption).await @@ -448,7 +383,7 @@ pub async fn send_reply( Ok(r) => r, Err(e) => return e, }; - if let Some(reject) = unread_guard(client, &room).await { + if let Some(reject) = unread_guard(&room) { return reject; } let eid: OwnedEventId = match event_id.parse() { @@ -901,18 +836,40 @@ pub async fn download_file( })) } +/// Return the number of joined rooms with at least one unread +/// notification according to the server-side push notification counts +/// cached by the matrix-sdk client. +/// +/// The harness uses [`unread_summary`] (richer, with per-room body +/// snippets) rather than this endpoint. `unread_count` is kept as a +/// lightweight public endpoint for callers that only need the count +/// and want to avoid the `/messages` network round-trips that +/// [`collect_unread`] issues for count==1 rooms. +#[must_use] +pub fn unread_count(client: &Client) -> DaemonResponse { + let rooms = u32::try_from( + client + .joined_rooms() + .into_iter() + .filter(|r| r.unread_notification_counts().notification_count > 0) + .count(), + ) + .unwrap_or(u32::MAX); + DaemonResponse::ok(&serde_json::json!({ "rooms": rooms })) +} + /// Collect all rooms with unread notifications and build per-room /// [`crate::protocol::RoomUnread`] entries. For rooms with exactly /// one unread notification the last message body is fetched via the /// `/messages` endpoint (best-effort; failures leave `last_body` as /// `None`). Rooms with zero unreads are omitted. /// -/// **Latency note**: each unread room triggers a live `/messages` -/// network request to the matrix homeserver to determine its latest -/// event (via [`room_unread_state`]). This adds per-room round-trip -/// latency to `get_loose_ends` and the wake-signal path. Acceptable in -/// practice (rooms with unread are few; request is best-effort), but -/// worth bearing in mind if latency becomes a concern. +/// **Latency note**: each count==1 room triggers a live `/messages` +/// network request to the matrix homeserver to retrieve the message +/// body. This adds per-room round-trip latency to `get_loose_ends` +/// and the wake-signal path. Acceptable in practice (rooms with +/// unread are few; request is best-effort), but worth bearing in +/// mind if latency becomes a concern. #[must_use] pub async fn collect_unread(client: &Client) -> Vec { collect_unread_with_ids(client) @@ -932,29 +889,74 @@ pub async fn collect_unread_with_ids( ) -> Vec<(matrix_sdk::ruma::OwnedRoomId, crate::protocol::RoomUnread)> { use crate::protocol::RoomUnread; let mut result = Vec::new(); + let own_user_id = client.user_id(); for room in client.joined_rooms() { - let Some((_, sender, body)) = room_unread_state(client, &room).await else { + let count = + u32::try_from(room.unread_notification_counts().notification_count).unwrap_or(u32::MAX); + if count == 0 { continue; - }; + } let label = room_label(&room); - let last_body = (!body.is_empty()).then(|| crate::wake::truncate_chars(&body, 100)); + let (last_body, last_sender) = if count == 1 { + fetch_last_message(client, &room).await + } else { + (None, None) + }; + // Skip a single-unread room whose newest event is the agent's own + // message: after a rebuild the read receipt may not have advanced + // past it yet, so `notification_count` can still report 1 — without + // this the agent self-wakes on its own message. Only reachable on the + // count==1 path, where `last_sender` is populated. + if let (Some(sender), Some(own_id)) = (&last_sender, own_user_id) + && sender.as_str() == own_id.as_str() + { + continue; + } result.push(( room.room_id().to_owned(), RoomUnread { label, - count: 1, + count, last_body, - last_sender: Some(sender.to_string()), + last_sender, }, )); } result } -/// Return per-room unread summaries. Each entry carries the sender + -/// truncated body of the room's latest (unread) event when it's a -/// text-like message; rooms whose latest unread event has no body -/// (a reaction, a state change, …) carry the room + sender only. +/// Fetch the body + sender of the most recent room message. Returns +/// `(None, None)` on any error or when the timeline contains no text +/// events. +async fn fetch_last_message( + client: &Client, + room: &matrix_sdk::Room, +) -> (Option, Option) { + use matrix_sdk::ruma::api::Direction; + use matrix_sdk::ruma::api::client::message::get_message_events; + let mut req = + get_message_events::v3::Request::new(room.room_id().to_owned(), Direction::Backward); + req.limit = matrix_sdk::ruma::UInt::from(1u32); + let Ok(resp) = client.send(req).await else { + return (None, None); + }; + for raw in &resp.chunk { + if let Ok(ev) = raw.deserialize() { + let body = extract_body(&ev); + if !body.is_empty() { + return ( + Some(crate::wake::truncate_chars(&body, 100)), + Some(ev.sender().to_string()), + ); + } + } + } + (None, None) +} + +/// Return per-room unread summaries. For rooms with exactly one +/// unread notification, attempts to include the sender + truncated +/// body; rooms with multiple unreads carry only the count. #[must_use] pub async fn unread_summary(client: &Client) -> DaemonResponse { let rooms = collect_unread(client).await; diff --git a/hive-matrix-mcp/src/protocol.rs b/hive-matrix-mcp/src/protocol.rs index a4bd41ce..26443051 100644 --- a/hive-matrix-mcp/src/protocol.rs +++ b/hive-matrix-mcp/src/protocol.rs @@ -27,14 +27,10 @@ pub enum InviteAction { pub struct RoomUnread { /// Canonical alias (`#name:server`) or room id (`!id:server`). pub label: String, - /// Always `1` — an exact unread-message count is no longer tracked - /// (see `handlers::room_unread_state`'s doc comment for why); the - /// agent reads the room via `read_room` for full context, so only - /// "there's something new" plus a one-message preview matters here. + /// Server-side push-notification count for this room. Always ≥ 1. pub count: u32, - /// Truncated body of the room's latest (unread) event. Present when - /// that event is a text-like message; absent for a reaction/state - /// event or when the fetch failed. + /// Truncated body of the last message in the room. Present only when + /// `count == 1` and the fetch succeeded; absent otherwise. #[serde(skip_serializing_if = "Option::is_none")] pub last_body: Option, /// Sender of the last message (`@user:server`). Present when