diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index c69bf547..9dd83b26 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -901,6 +901,21 @@ pub fn unread_count(client: &Client) -> DaemonResponse { /// mind if latency becomes a concern. #[must_use] pub async fn collect_unread(client: &Client) -> Vec { + collect_unread_with_ids(client) + .await + .into_iter() + .map(|(_, ru)| ru) + .collect() +} + +/// Like [`collect_unread`] but pairs each entry with its `OwnedRoomId`. +/// The todo producer (loose-ends v2, #2569) needs the room id as the +/// per-room upsert/dedup key, which the claude-facing `RoomUnread` +/// payload intentionally doesn't carry. +#[must_use] +pub async fn collect_unread_with_ids( + client: &Client, +) -> Vec<(matrix_sdk::ruma::OwnedRoomId, crate::protocol::RoomUnread)> { use crate::protocol::RoomUnread; let mut result = Vec::new(); for room in client.joined_rooms() { @@ -915,12 +930,15 @@ pub async fn collect_unread(client: &Client) -> Vec } else { (None, None) }; - result.push(RoomUnread { - label, - count, - last_body, - last_sender, - }); + result.push(( + room.room_id().to_owned(), + RoomUnread { + label, + count, + last_body, + last_sender, + }, + )); } result } diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 49773875..bfba0573 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -299,6 +299,11 @@ async fn bring_up_account( // 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())); + // Startup cancel-and-recreate (loose-ends v2, #2569): wipe this agent's + // matrix todos so stale ones (rooms read while the daemon was down) don't + // linger, then let the first sweep rebuild the set to match current + // unread reality. Best-effort; the sweep converges regardless. + let _ = crate::wake::send_todo_clear(hyperhive_socket, None, true).await; let sync_loop: SyncLoop = Box::pin(async move { sync_client .sync_with_callback(SyncSettings::default(), move |_response| { diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 9947866c..79ac2ad9 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -20,23 +20,19 @@ use tokio::sync::Mutex; use crate::{handlers, wake}; -/// 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. +/// Push a *todo* (loose-ends v2, #2569) for each joined room carrying +/// unread notifications, and clear the todo for rooms that have been read. +/// Replaces the old direct-wake path: instead of firing an all-rooms wake, +/// each unread room becomes a per-room `upsert_todo` keyed by its room id, +/// and hive-c0re coalesces exactly one wake when the todo set changes. /// -/// `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. +/// Driven from the post-sync callback — the same reliable path +/// `sweep_invites` uses — so it re-checks unread each tick; a dropped +/// upsert is retried on the next one (self-healing once the socket is +/// back). `notified` tracks which rooms currently have an active todo so +/// a room that becomes read gets its todo cleared. `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. @@ -46,55 +42,42 @@ pub async fn sweep_unread( notified: &Mutex>, account_tag: Option<&str>, ) { - // 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(); + let unread = handlers::collect_unread_with_ids(client).await; + let unread_ids: HashSet = unread.iter().map(|(id, _)| id.clone()).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()); + // Rooms we previously pushed a todo for that are no longer unread → the + // agent read them; clear their todo. Snapshot under the lock, send + // outside it, drop from `notified` on success (retry next tick on fail). + let stale: Vec = { + let active = notified.lock().await; + active.difference(&unread_ids).cloned().collect() + }; + for id in stale { + if wake::send_todo_clear(socket, Some(id.as_str()), false) + .await + .is_ok() + { + notified.lock().await.remove(&id); + } + } + + // Upsert a todo per currently-unread room. hive-c0re coalesces the wake + // iff the summary is new or changed, so re-upserting an unchanged room + // every sync tick is a cheap server-side no-op (no re-wake). + for (id, ru) in &unread { + let summary = wake::tag_account( + account_tag, + wake::format_unread_summary(std::slice::from_ref(ru)), + ); + match wake::send_todo_upsert(socket, id.as_str(), &summary).await { + Ok(()) => { + notified.lock().await.insert(id.clone()); + } + Err(e) => { + tracing::warn!(error = %e, room = %id, "matrix: todo upsert failed; will retry next sync"); } } } - 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 9fe371c7..9e041a98 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -34,15 +34,65 @@ use tokio::net::UnixStream; /// Returns an error on socket connect failure, serialisation failure, /// or I/O error writing to or reading from the socket. pub async fn send_wake(socket: &Path, body: impl AsRef) -> Result<()> { - use tokio::io::AsyncBufReadExt; - let payload = serde_json::json!({ "cmd": "wake", "from": "matrix", "body": body.as_ref(), "transient": true, }); - let line = format!("{}\n", serde_json::to_string(&payload)?); + send_line(socket, &payload).await +} + +/// Upsert a matrix-subsystem *todo* (loose-ends v2, #2569) on the +/// hyperhive control socket — the replacement for a direct wake. `key` is +/// the room id (the dedup key); hive-c0re coalesces a wake iff the todo is +/// new or its `summary` changed. Best-effort like [`send_wake`]. +/// +/// # Errors +/// +/// Returns an error on socket connect failure, serialisation failure, +/// or I/O error writing to or reading from the socket. +pub async fn send_todo_upsert(socket: &Path, key: &str, summary: impl AsRef) -> Result<()> { + let payload = serde_json::json!({ + "cmd": "upsert_todo", + "subsystem": "matrix", + "key": key, + "summary": summary.as_ref(), + }); + send_line(socket, &payload).await +} + +/// Clear matrix-subsystem todos. `key = Some(room)` clears one room's +/// todo (it was read); `all = true` wipes the whole matrix set +/// (cancel-and-recreate on daemon restart). Best-effort. +/// +/// # Errors +/// +/// Returns an error on socket connect failure, serialisation failure, +/// or I/O error writing to or reading from the socket. +pub async fn send_todo_clear(socket: &Path, key: Option<&str>, all: bool) -> Result<()> { + let payload = serde_json::json!({ + "cmd": "clear_todo", + "subsystem": "matrix", + "key": key, + "all": all, + }); + send_line(socket, &payload).await +} + +/// Write one JSON request line to the hyperhive control socket and drain +/// the response line (best-effort — the reply is not acted on, we just +/// read it so the server doesn't get ECONNRESET on its write-back). +/// Shared by [`send_wake`] and the todo senders. +/// +/// # Errors +/// +/// Returns an error on socket connect failure, serialisation failure, +/// or I/O error writing to or reading from the socket. +async fn send_line(socket: &Path, payload: &serde_json::Value) -> Result<()> { + use tokio::io::AsyncBufReadExt; + + let line = format!("{}\n", serde_json::to_string(payload)?); let stream = UnixStream::connect(socket) .await .with_context(|| format!("connect hyperhive socket {}", socket.display()))?; @@ -50,13 +100,11 @@ pub async fn send_wake(socket: &Path, body: impl AsRef) -> Result<()> { write .write_all(line.as_bytes()) .await - .with_context(|| format!("write wake to {}", socket.display()))?; + .with_context(|| format!("write to {}", socket.display()))?; write .shutdown() .await .with_context(|| format!("shutdown write to {}", socket.display()))?; - // Drain the response line so the server doesn't get ECONNRESET on - // its write-back. We don't act on the response — best-effort wake. let mut reader = tokio::io::BufReader::new(read); let mut resp = String::new(); let _ = reader.read_line(&mut resp).await;