diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 9d9a0b61..49773875 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -290,21 +290,28 @@ async fn bring_up_account( // flagged "unverified" in other users' clients. Idempotent + swallows // failures (a UIAA-requiring homeserver) inside ensure_cross_signing. client::ensure_cross_signing(&client, &cfg.name).await; - timeline::install_message_handler(&client, hyperhive_socket.to_path_buf(), tag.clone()); let sync_client = client.clone(); let cb_client = client.clone(); - let invite_socket = Arc::new(hyperhive_socket.to_path_buf()); + let wake_socket = Arc::new(hyperhive_socket.to_path_buf()); + // Separate dedup sets: one tracks invites already woken about, the + // other unread-message rooms. Both are pruned to their current state + // each sweep (see the sweep fns) so re-invites / new messages re-wake. let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); + let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); let sync_loop: SyncLoop = Box::pin(async move { sync_client .sync_with_callback(SyncSettings::default(), move |_response| { let client = cb_client.clone(); - let socket = invite_socket.clone(); - let notified = invite_notified.clone(); + let socket = wake_socket.clone(); + let invite_notified = invite_notified.clone(); + let unread_notified = unread_notified.clone(); let tag = tag.clone(); async move { - timeline::sweep_invites(&client, &socket, ¬ified, tag.as_deref()).await; + timeline::sweep_invites(&client, &socket, &invite_notified, tag.as_deref()) + .await; + timeline::sweep_unread(&client, &socket, &unread_notified, tag.as_deref()) + .await; matrix_sdk::LoopCtrl::Continue } }) diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index a60fe479..9947866c 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -1,108 +1,100 @@ -//! Matrix event handlers: incoming room messages fire a hyperhive -//! wake signal so the agent's harness drives a new claude turn. +//! Matrix → hyperhive wake bridge. Both room messages and room invites +//! are surfaced to the agent by SWEEPING state on every post-sync +//! callback, not by one-shot `m.room.message` / `StrippedRoomMemberEvent` +//! handlers: a one-shot wake whose `send_wake` raced a hive-c0re / socket- +//! down window (a container rebuild) was dropped with no retry, leaving +//! the agent deaf to matrix activity until manually prompted. Sweeping the +//! reliable sync path re-checks each tick and self-heals a dropped wake on +//! the next one. //! -//! Per the operator's call: wake body is a SHORT TEASER, not the full message -//! (msg stays unread server-side; agent fetches via `read_room`). The -//! `wake::format_wake_body` truncates to ~100 chars. -//! -//! Self-events (events sent by this agent) are filtered out so an -//! agent posting a message doesn't wake itself. +//! Wake bodies stay short: `sweep_unread` sends the all-rooms unread +//! summary (`wake::format_unread_summary`); the message stays unread +//! server-side so the agent fetches detail via `read_room`. Self-sent +//! messages never raise an unread notification, so they never wake. use std::collections::HashSet; -use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::path::Path; -use matrix_sdk::{ - Client, Room, RoomState, - ruma::OwnedRoomId, - ruma::events::room::message::{MessageType, OriginalSyncRoomMessageEvent}, -}; +use matrix_sdk::{Client, ruma::OwnedRoomId}; use tokio::sync::Mutex; use crate::{handlers, wake}; -/// Install the room-message handler on `client`. Fires on every -/// `m.room.message` event in a joined room; non-self messages trigger -/// a wake signal to the hyperhive harness via the unix socket at -/// `hyperhive_socket`. The wake body summarises ALL rooms with unread -/// notifications at wake time (not just the triggering event) so the -/// agent receives a full picture in one prompt. +/// Wake the agent for any joined room carrying unread notifications it has +/// not yet been successfully woken about. Driven from the post-sync +/// callback — the same reliable path `sweep_invites` uses — rather than a +/// one-shot `m.room.message` event handler: a message-wake whose +/// `send_wake` raced a hive-c0re / socket-down window (a container rebuild) +/// was dropped with no retry, so the agent went deaf until manually +/// prompted. The sync callback fires on every sync response (so promptly on +/// new activity), so this re-checks unread each tick and retries a dropped +/// wake on the next one — self-healing once the socket is back. /// -/// `account_tag` is `Some(name)` only in multi-account mode (N>1 matrix -/// accounts on this daemon); when set it is prepended to the wake body -/// so the agent knows which account to `read_room` on. In the -/// single-account case it is `None` and the wake body is unchanged. -pub fn install_message_handler( +/// `notified` dedups so a standing unread wakes the agent once, not on +/// every sync tick; it is pruned to the current unread set each pass so a +/// room that has been read and then receives a new message wakes again. On +/// a wake-send failure the freshly-seen rooms are rolled back out of +/// `notified` so the next pass retries them. `account_tag` is `Some(name)` +/// only in multi-account mode, prepended so the agent knows which account +/// to `read_room` on. +/// +/// Self-sent messages don't raise an unread notification server-side, so +/// no explicit self-filter is needed here. +pub async fn sweep_unread( client: &Client, - hyperhive_socket: PathBuf, - account_tag: Option, + socket: &Path, + notified: &Mutex>, + account_tag: Option<&str>, ) { - let socket = Arc::new(hyperhive_socket); - let account_tag = Arc::new(account_tag); - let own_user = client.user_id().map(std::borrow::ToOwned::to_owned); - client.add_event_handler({ - let socket = socket.clone(); - move |event: OriginalSyncRoomMessageEvent, room: Room, client: Client| { - let socket = socket.clone(); - let account_tag = account_tag.clone(); - let own_user = own_user.clone(); - async move { - // INFO so the live host journal shows the handler actually - // firing on an incoming event — the daemon previously logged - // only handler *install*, which made a non-firing dispatch - // path (sync not advancing → events deduped, or a room-state - // mismatch) impossible to distinguish from a delivery failure. - tracing::info!( - room = %room.room_id(), - sender = %event.sender, - state = ?room.state(), - "matrix: message handler fired" - ); - if room.state() != RoomState::Joined { - tracing::info!( - room = %room.room_id(), - state = ?room.state(), - "matrix: skipping message — room not in Joined state" - ); - return; - } - // Self-event filter so the agent's own outgoing messages - // don't wake it. matches forge_notify's self-skip pattern. - if own_user.as_ref().is_some_and(|u| u == &event.sender) { - tracing::debug!(sender = %event.sender, "matrix: skipping self-event"); - return; - } - // Build a wake body that covers all rooms with unread - // notifications, not just the triggering event. This lets - // the agent see the full backlog in a single wake prompt. - // Falls back to the per-event teaser if the collect fails - // (empty result means daemon is not seeing any unread yet — - // unlikely but possible during a sync race). - let unread = handlers::collect_unread(&client).await; - let body = if unread.is_empty() { - // Sync hasn't updated notification counts yet; fall back - // to the current event so the agent still wakes. - let text = match &event.content.msgtype { - MessageType::Text(t) => t.body.clone(), - MessageType::Notice(n) => n.body.clone(), - MessageType::Emote(e) => format!("* {}", e.body), - _ => format!("[{}]", event.content.msgtype()), - }; - let label = handlers::room_label(&room); - wake::format_wake_body(event.sender.as_str(), &label, &text) - } else { - wake::format_unread_summary(&unread) - }; - let body = wake::tag_account(account_tag.as_ref().as_deref(), body); - if let Err(e) = wake::send_wake(&socket, &body).await { - tracing::warn!(error = %e, "failed to deliver matrix wake to hyperhive"); - } else { - tracing::info!(room = %room.room_id(), "matrix wake delivered"); - } + // Current set of joined rooms carrying unread notifications. + let unread_ids: HashSet = client + .joined_rooms() + .into_iter() + .filter(|r| r.unread_notification_counts().notification_count > 0) + .map(|r| r.room_id().to_owned()) + .collect(); + + // Which unread rooms are newly-seen (not already successfully woken)? + let mut fresh = Vec::new(); + { + let mut seen = notified.lock().await; + // Drop rooms no longer unread (read / left) so a future new message + // in them wakes the agent again. + seen.retain(|id| unread_ids.contains(id)); + for id in &unread_ids { + if seen.insert(id.clone()) { + fresh.push(id.clone()); } } - }); - tracing::info!("matrix message handler installed"); + } + if fresh.is_empty() { + return; + } + + // Roll the freshly-seen rooms back out of `notified` so the next sweep + // retries them. Used on both the sync-race (empty collect) and the + // wake-send-failure paths. + let rollback = || async { + let mut seen = notified.lock().await; + for id in &fresh { + seen.remove(id); + } + }; + + // Same all-rooms unread summary the wake body has always used. Empty + // only under a sync race (counts changed between the scan and collect). + let unread = handlers::collect_unread(client).await; + if unread.is_empty() { + rollback().await; + return; + } + let body = wake::tag_account(account_tag, wake::format_unread_summary(&unread)); + if let Err(e) = wake::send_wake(socket, &body).await { + rollback().await; + tracing::warn!(error = %e, "matrix: unread-sweep wake failed; will retry next sync"); + } else { + tracing::info!(rooms = fresh.len(), "matrix: unread-sweep wake delivered"); + } } /// Wake the agent for any pending room invite it has not yet been diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index e37891ef..9fe371c7 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -19,11 +19,6 @@ use anyhow::{Context, Result}; use tokio::io::AsyncWriteExt; use tokio::net::UnixStream; -/// Max characters of `body` to embed in the wake payload. Shorter than -/// `forge_notify`'s 500-byte excerpt because the agent has a follow-up -/// `read_room` tool to fetch the full event. -pub const WAKE_BODY_TRUNCATE: usize = 100; - /// Send an `AgentRequest::Wake { from: "matrix", body }` to the hyperhive /// control socket at `socket`. Best-effort: returns Err on any plumbing /// failure; callers log + ignore so a wake delivery hiccup doesn't tear @@ -68,16 +63,6 @@ pub async fn send_wake(socket: &Path, body: impl AsRef) -> Result<()> { Ok(()) } -/// Format a wake-message body from a matrix event's sender + room -/// canonical alias + body text. Truncates at [`WAKE_BODY_TRUNCATE`] -/// chars with an ellipsis. Shape: `[matrix] in : ` -/// matches the `forge_notify` `[issue …]` framing convention. -#[must_use] -pub fn format_wake_body(sender: &str, room: &str, text: &str) -> String { - let truncated = truncate_chars(text, WAKE_BODY_TRUNCATE); - format!("[matrix] {sender} in {room}: {truncated}") -} - /// Format a wake-message body from a list of per-room unread summaries. /// Single-room / single-message case collapses to the terse one-liner /// format; multiple rooms expand to a bulleted list. Always appends a @@ -154,27 +139,6 @@ pub fn truncate_chars(s: &str, max: usize) -> String { mod tests { use super::*; - #[test] - fn format_wake_body_short_passes_through() { - let body = format_wake_body("@iris:matrix.darkest.space", "#general", "hi all"); - assert_eq!( - body, - "[matrix] @iris:matrix.darkest.space in #general: hi all" - ); - } - - #[test] - fn format_wake_body_long_truncates_with_ellipsis() { - let long = "x".repeat(200); - let body = format_wake_body("@iris:m", "#x", &long); - assert!(body.contains("xxxxxxx")); - assert!(body.ends_with("…")); - // Header + truncated body should be well under the absolute - // wake-message ceiling (forge_notify uses ~600 bytes; we're - // way under that). - assert!(body.len() < 200); - } - #[test] fn truncate_chars_handles_multibyte() { // `ü` is 2 bytes / 1 char. truncating to 3 chars on "üüüüüü"