feat(#2569): matrix producer — sweep_unread pushes per-room todos instead of direct wakes

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<crate::protocol::RoomUnread> {
+    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<crate::protocol::RoomUnread>
         } 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
 }

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| {

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<HashSet<OwnedRoomId>>,
     account_tag: Option<&str>,
 ) {
-    // Current set of joined rooms carrying unread notifications.
-    let unread_ids: HashSet<OwnedRoomId> = 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<OwnedRoomId> = 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<OwnedRoomId> = {
+        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

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<str>) -> 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<str>) -> 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<str>) -> 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;