From 1192c4e4c70c478a5ffe176e35180681f4406bff Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 3 Aug 2026 13:18:15 +0200 Subject: [PATCH] matrix: wake and unread-guard on own read receipt, not push-rule count --- hive-matrix-mcp/src/handlers.rs | 176 ++++++++++++++++++-------------- hive-matrix-mcp/src/protocol.rs | 10 +- 2 files changed, 105 insertions(+), 81 deletions(-) diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 7b7d6df6..facefb5b 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, + receipt::{ReceiptThread, ReceiptType as LocalReceiptType}, relation::Annotation, room::message::{AddMentions, MessageType, RoomMessageEventContent}, }, @@ -175,33 +175,98 @@ 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 -/// notifications (the agent must `read_room` then `mark_read` the -/// latest event first), or `None` when the send may proceed. +/// content (the agent must `read_room` then `mark_read` the latest event +/// first), or `None` when the send may proceed. /// -/// 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; - } +/// 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?; let label = room_label(room); Some(DaemonResponse::error(format!( - "refusing to send: {count} unread message(s) in {label}. \ + "refusing to send: 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(&room) { + if let Some(reject) = unread_guard(client, &room).await { return reject; } let content = RoomMessageEventContent::text_markdown(body); @@ -242,7 +307,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(&room) { + if let Some(reject) = unread_guard(client, &room).await { return reject; } let content = RoomMessageEventContent::text_markdown(body); @@ -328,7 +393,7 @@ pub async fn send_file( Ok(r) => r, Err(e) => return e, }; - if let Some(reject) = unread_guard(&room) { + if let Some(reject) = unread_guard(client, &room).await { return reject; } upload_attachment(&room, path, caption).await @@ -383,7 +448,7 @@ pub async fn send_reply( Ok(r) => r, Err(e) => return e, }; - if let Some(reject) = unread_guard(&room) { + if let Some(reject) = unread_guard(client, &room).await { return reject; } let eid: OwnedEventId = match event_id.parse() { @@ -864,12 +929,12 @@ pub fn unread_count(client: &Client) -> DaemonResponse { /// `/messages` endpoint (best-effort; failures leave `last_body` as /// `None`). Rooms with zero unreads are omitted. /// -/// **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. +/// **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. #[must_use] pub async fn collect_unread(client: &Client) -> Vec { collect_unread_with_ids(client) @@ -889,74 +954,29 @@ 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 count = - u32::try_from(room.unread_notification_counts().notification_count).unwrap_or(u32::MAX); - if count == 0 { + let Some((_, sender, body)) = room_unread_state(client, &room).await else { continue; - } - let label = room_label(&room); - 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; - } + let label = room_label(&room); + let last_body = (!body.is_empty()).then(|| crate::wake::truncate_chars(&body, 100)); result.push(( room.room_id().to_owned(), RoomUnread { label, - count, + count: 1, last_body, - last_sender, + last_sender: Some(sender.to_string()), }, )); } result } -/// 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. +/// 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. #[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 26443051..a4bd41ce 100644 --- a/hive-matrix-mcp/src/protocol.rs +++ b/hive-matrix-mcp/src/protocol.rs @@ -27,10 +27,14 @@ pub enum InviteAction { pub struct RoomUnread { /// Canonical alias (`#name:server`) or room id (`!id:server`). pub label: String, - /// Server-side push-notification count for this room. Always ≥ 1. + /// 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. pub count: u32, - /// Truncated body of the last message in the room. Present only when - /// `count == 1` and the fetch succeeded; absent otherwise. + /// 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. #[serde(skip_serializing_if = "Option::is_none")] pub last_body: Option, /// Sender of the last message (`@user:server`). Present when