vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

matrix: wake and unread-guard on own read receipt, not push-rule count

Browse source
This commit is contained in:
damocles 2026-08-03 13:18:15 +02:00 • committed by mara
commit 1192c4e4c7
2 changed files with 105 additions and 81 deletions
Download patch file Download diff file Expand all files Collapse all files

176
hive-matrix-mcp/src/handlers.rs
View file

@ -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<DaemonResponse> {
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<DaemonResponse> {
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<crate::protocol::RoomUnread> {
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<String>, Option<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 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;

10
hive-matrix-mcp/src/protocol.rs
View file

@ -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<String>,
/// Sender of the last message (`@user:server`). Present when