hive-matrix-mcp: test the retry path and the once-only todo clear

The startup todo wipe runs once in main, before any sync loop, so a
secondary that comes up later cannot clear the todos the running
accounts' sweeps already deduped. Nothing tested that.

main now reads HIVE_AGENT_SOCKET once at startup and hands the path to
wake::clear_all_todos, the only way left to wipe the whole set;
send_todo_clear clears one key only. bring_up_account is split into
restore_account (everything that talks to the network) and the
synchronous account_sync_loop, and the retry path restores with the
former and builds the loop itself. Neither the loop nor the retry path
is given the socket.

The new test runs a late secondary's retry and its sync loop against a
listener standing in for the harness socket, after startup's wipe, and
asserts the harness saw exactly one wipe. It pins that seam, not every
place in restore_account a wipe could be added.

hive-matrix-mcp/src/handlers.rs

@@ -695,11 +695,8 @@ pub async fn join_room(client: &Client, room_ref: &str) -> DaemonResponse {
             // Clear the invite todo so the accepted invite drops out of
             // `get_loose_ends` immediately (the sweep would also clear it
             // on its next tick, but this makes it instant).
-            let _ = crate::wake::send_todo_clear(
+            let _ =
-                Some(&crate::timeline::invite_key(room.room_id())),
+                crate::wake::send_todo_clear(&crate::timeline::invite_key(room.room_id())).await;
-                false,
-            )
-            .await;
             DaemonResponse::ok(&serde_json::json!({
                 "joined": true,
                 "room_id": room.room_id().to_string(),
@@ -733,11 +730,9 @@ pub async fn resolve_invite(
             match room.leave().await {
                 Ok(()) => {
                     // Clear the invite todo immediately on reject.
-                    let _ = crate::wake::send_todo_clear(
+                    let _ =
-                        Some(&crate::timeline::invite_key(room.room_id())),
+                        crate::wake::send_todo_clear(&crate::timeline::invite_key(room.room_id()))
-                        false,
+                            .await;
-                    )
-                    .await;
                     DaemonResponse::ok(&serde_json::json!({
                         "rejected": true,
                         "room_id": room.room_id().to_string(),

hive-matrix-mcp/src/main.rs

@@ -118,6 +118,7 @@ async fn main() -> Result<ExitCode> {
         .init();
 
     let cli = Cli::parse();
+    let todo_socket = wake::agent_socket();
     let declared = accounts::configured().context("read matrix account config")?;
     accounts::remove_delivered_files(&declared);
     let mut warned = HashSet::new();
@@ -188,8 +189,11 @@ async fn main() -> Result<ExitCode> {
     // daemon was down) don't linger, then let the first sweep rebuild the
     // set to match current reality. Best-effort; the sweep converges.
     // Once, before any sync loop runs: a secondary that comes up later must
-    // not wipe the todos the running accounts' sweeps already deduped.
+    // not wipe the todos the running accounts' sweeps already deduped, so
-    let _ = wake::send_todo_clear(None, true).await;
+    // `todo_socket` is passed to nothing that brings an account up.
+    if let Some(socket) = &todo_socket {
+        let _ = wake::clear_all_todos(socket).await;
+    }
 
     // Publish the live-account snapshot (BE-4) for the dashboard: the
     // accounts that restored, with effective homeserver + user id. Lands
@@ -488,11 +492,11 @@ impl<T> DownAccounts<T> {
     }
 }
 
-/// What a bring-up attempt yields; [`bring_up_account`]'s result.
+/// What a restore attempt yields; [`restore_account`]'s result.
-type BringUp = Result<Option<(Client, SyncLoop)>>;
+type Restore = Result<Option<Client>>;
 
 /// Every [`LINKED_REFRESH`], try again the secondary [`DownAccounts::due`]
-/// names, one per tick, with [`bring_up_account`]. Never returns.
+/// names, one per tick, with [`restore_account`]. Never returns.
 async fn retry_down(
     mut down: DownAccounts<Retry>,
     registry: Arc<Registry>,
@@ -503,22 +507,22 @@ async fn retry_down(
         &registry,
         &up,
         &paths::accounts_file(),
-        |cfg: AccountCfg, tag| async move { bring_up_account(&cfg, tag, false).await },
+        |cfg: AccountCfg| async move { restore_account(&cfg, false).await },
     )
     .await
 }
 
-/// [`retry_down`] with the bring-up step and the snapshot path injected.
+/// [`retry_down`] with the restore step and the snapshot path injected.
 async fn retry_loop<F, Fut>(
     down: &mut DownAccounts<Retry>,
     registry: &Registry,
     up: &tokio::sync::mpsc::UnboundedSender<NamedSyncLoop>,
     snapshot: &Path,
-    mut bring_up: F,
+    mut restore: F,
 ) -> Infallible
 where
-    F: FnMut(AccountCfg, Option<String>) -> Fut,
+    F: FnMut(AccountCfg) -> Fut,
-    Fut: std::future::Future<Output = BringUp>,
+    Fut: std::future::Future<Output = Restore>,
 {
     let mut tick = tokio::time::interval(LINKED_REFRESH);
     tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
@@ -526,33 +530,34 @@ where
     tick.tick().await;
     loop {
         let now = tick.tick().await;
-        retry_once(down, registry, up, snapshot, now, &mut bring_up).await;
+        retry_once(down, registry, up, snapshot, now, &mut restore).await;
     }
 }
 
 /// Try the secondary of `down` that is due at `now`, if any. One that comes up
-/// leaves `down`, joins `registry` and has its sync loop sent to `up`; the
+/// leaves `down`, joins `registry` and has its [`account_sync_loop`] sent to
-/// accounts already in `registry` are not touched. One that fails stays in
+/// `up`; the accounts already in `registry` are not touched. One that fails
-/// `down`, due again after a longer wait. No lock on `registry` is held while
+/// stays in `down`, due again after a longer wait. No lock on `registry` is
-/// `bring_up` runs.
+/// held while `restore` runs.
 async fn retry_once<F, Fut>(
     down: &mut DownAccounts<Retry>,
     registry: &Registry,
     up: &tokio::sync::mpsc::UnboundedSender<NamedSyncLoop>,
     snapshot: &Path,
     now: Instant,
-    bring_up: &mut F,
+    restore: &mut F,
 ) where
-    F: FnMut(AccountCfg, Option<String>) -> Fut,
+    F: FnMut(AccountCfg) -> Fut,
-    Fut: std::future::Future<Output = BringUp>,
+    Fut: std::future::Future<Output = Restore>,
 {
     let Some((name, (cfg, tag))) = down.due(now) else {
         return;
     };
     let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone());
-    match bring_up(cfg.clone(), tag.clone()).await {
+    match restore(cfg.clone()).await {
-        Ok(Some((client, sync_loop))) => {
+        Ok(Some(client)) => {
             tracing::info!(account = %name, "secondary matrix account came up");
+            let sync_loop = account_sync_loop(&client, tag);
             down.came_up(&name);
             registry.insert(name.clone(), client);
             if let Err(e) = registry.write_snapshot(snapshot) {
@@ -644,16 +649,25 @@ async fn ensure_recovery(client: &Client, account: &str) {
     }
 }
 
-/// Restore one account's client (when its token exists), install its
+/// Restore one account's client (when its token exists) and build its sync
-/// message handler, and build its sync loop. Returns
+/// loop. Returns `Ok(Some((client, sync_loop)))` when the account came up,
-/// `Ok(Some((client, sync_loop)))` when the account came up, `Ok(None)`
+/// `Ok(None)` when it has no token or homeserver (caller decides
-/// when its token file is absent (caller decides primary-vs-secondary
+/// primary-vs-secondary handling). The returned sync loop is driven by the
-/// handling). The returned sync loop is driven by the caller.
+/// caller.
 async fn bring_up_account(
     cfg: &AccountCfg,
     tag: Option<String>,
     is_primary: bool,
 ) -> Result<Option<(Client, SyncLoop)>> {
+    Ok(restore_account(cfg, is_primary).await?.map(|client| {
+        let sync_loop = account_sync_loop(&client, tag);
+        (client, sync_loop)
+    }))
+}
+
+/// Restore one account's client, when it has a homeserver and a token:
+/// everything in bringing an account up that talks to the network.
+async fn restore_account(cfg: &AccountCfg, is_primary: bool) -> Result<Option<Client>> {
     let Some(homeserver) = cfg.homeserver() else {
         // No homeserver for this account: the hive has none to offer (no
         // matrix vhost) or this agent's `services.hyperhive.agent.matrix.url` is null. Same
@@ -688,7 +702,13 @@ async fn bring_up_account(
     client::ensure_cross_signing(&client, &cfg.name).await;
     // Best-effort, and after cross-signing: see `client::ensure_recovery`.
     ensure_recovery(&client, &cfg.name).await;
+    Ok(Some(client))
+}
 
+/// `client`'s sync loop, sweeping invites and unread rooms into todos tagged
+/// `tag` after each sync. Builds the loop only; nothing runs until it is
+/// polled. Never wipes the todo set: that is `main`'s, once, at startup.
+fn account_sync_loop(client: &Client, tag: Option<String>) -> SyncLoop {
     let sync_client = client.clone();
     let cb_client = client.clone();
     // Separate dedup sets: one tracks invites already pushed as todos, the
@@ -696,7 +716,7 @@ async fn bring_up_account(
     // each sweep (see the sweep fns) so re-invites / new messages re-push.
     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 {
+    Box::pin(async move {
         sync_client
             .sync_with_callback(SyncSettings::default(), move |_response| {
                 let client = cb_client.clone();
@@ -712,8 +732,7 @@ async fn bring_up_account(
             .await
             .context("matrix-sdk sync loop exited")?;
         Ok(())
-    });
+    })
-    Ok(Some((client, sync_loop)))
 }
 
 #[cfg(test)]
@@ -730,8 +749,8 @@ mod tests {
     use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel};
 
     use super::{
-        AccountCfg, BringUp, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop,
+        AccountCfg, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, Registry,
-        Registry, Retry, SyncLoop, retry_delay, retry_loop, retry_once,
+        Restore, Retry, retry_delay, retry_loop, retry_once, wake,
     };
 
     #[test]
@@ -833,10 +852,6 @@ mod tests {
         }
     }
 
-    fn pending_loop() -> SyncLoop {
-        Box::pin(std::future::pending())
-    }
-
     /// A registry holding the primary `main` and a running secondary `other`.
     async fn registry() -> Registry {
         let r = Registry::new("main".to_owned());
@@ -877,9 +892,9 @@ mod tests {
         let (up, mut rx) = unbounded_channel();
         let snap = snapshot_path("up");
         let mut ccc = Some(client("ccc").await);
-        let mut bring_up = move |_: AccountCfg, _: Option<String>| {
+        let mut bring_up = move |_: AccountCfg| {
             let c = ccc.take().expect("tried once");
-            async move { Ok(Some((c, pending_loop()))) as BringUp }
+            async move { Ok(Some(c)) as Restore }
         };
 
         let now = t0 + LINKED_REFRESH;
@@ -915,9 +930,9 @@ mod tests {
             let (up, mut rx) = unbounded_channel();
             let snap = snapshot_path("fail");
             let mut outcome = Some(outcome);
-            let mut bring_up = move |_: AccountCfg, _: Option<String>| {
+            let mut bring_up = move |_: AccountCfg| {
                 let r = outcome.take().expect("tried once");
-                async move { r as BringUp }
+                async move { r as Restore }
             };
 
             let now = t0 + LINKED_REFRESH;
@@ -940,6 +955,56 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn a_secondary_that_comes_up_late_does_not_wipe_the_todos() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        let snap = snapshot_path("todos");
+        let socket = snap.with_file_name("agent.sock");
+        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
+        let ops = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let seen = Arc::clone(&ops);
+        let harness = tokio::spawn(async move {
+            loop {
+                let (mut conn, _) = listener.accept().await.unwrap();
+                let mut op = String::new();
+                conn.read_to_string(&mut op).await.unwrap();
+                seen.lock().unwrap().push(op);
+                conn.write_all(b"{\"ok\":true}\n").await.unwrap();
+            }
+        });
+        // Startup's wipe, as `main` runs it: the one the harness should see.
+        wake::clear_all_todos(&socket).await.unwrap();
+
+        let t0 = Instant::now();
+        let mut down = down_ccc(t0);
+        let reg = registry().await;
+        let (up, mut rx) = unbounded_channel();
+        let mut ccc = Some(client("ccc").await);
+        let mut restore = move |_: AccountCfg| {
+            let c = ccc.take().expect("tried once");
+            async move { Ok(Some(c)) as Restore }
+        };
+        retry_once(
+            &mut down,
+            &reg,
+            &up,
+            &snap,
+            t0 + LINKED_REFRESH,
+            &mut restore,
+        )
+        .await;
+        // Run the late account's sync loop; an offline client never syncs.
+        let late = rx.try_recv().expect("its sync loop is handed on");
+        let _ = tokio::time::timeout(Duration::from_secs(2), late).await;
+
+        harness.abort();
+        let ops = ops.lock().unwrap().clone();
+        let wipes = ops.iter().filter(|op| op.contains("\"all\":true")).count();
+        assert_eq!(wipes, 1, "todo ops the harness saw: {ops:?}");
+        std::fs::remove_dir_all(snap.parent().unwrap()).ok();
+    }
+
     #[tokio::test]
     async fn an_account_whose_wait_is_not_over_is_not_tried() {
         let t0 = Instant::now();
@@ -947,9 +1012,9 @@ mod tests {
         let reg = registry().await;
         let (up, _rx) = unbounded_channel();
         let mut tried = 0;
-        let mut bring_up = |_: AccountCfg, _: Option<String>| {
+        let mut bring_up = |_: AccountCfg| {
             tried += 1;
-            async { Ok(None) as BringUp }
+            async { Ok(None) as Restore }
         };
 
         let early = t0 + LINKED_REFRESH - Duration::from_secs(1);
@@ -976,12 +1041,12 @@ mod tests {
         let (gate_tx, gate_rx) = tokio::sync::oneshot::channel::<()>();
         let mut gate_rx = Some(gate_rx);
         let mut ccc = Some(client("ccc").await);
-        let mut bring_up = move |_: AccountCfg, _: Option<String>| {
+        let mut bring_up = move |_: AccountCfg| {
             let gate = gate_rx.take().expect("tried once");
             let c = ccc.take().expect("tried once");
             async move {
                 gate.await.unwrap();
-                Ok(Some((c, pending_loop()))) as BringUp
+                Ok(Some(c)) as Restore
             }
         };
 
@@ -1050,9 +1115,8 @@ mod tests {
                     .collect();
                 let (up, _rx) = unbounded_channel();
                 let snap = snapshot_path("readers");
-                let mut bring_up = |c: AccountCfg, _: Option<String>| async move {
+                let mut bring_up =
-                    Ok(Some((client(&c.name).await, pending_loop()))) as BringUp
+                    |c: AccountCfg| async move { Ok(Some(client(&c.name).await)) as Restore };
-                };
                 let mut now = t0 + LINKED_REFRESH;
                 for _ in 0..4 {
                     retry_once(&mut down, &reg, &up, &snap, now, &mut bring_up).await;
@@ -1081,9 +1145,9 @@ mod tests {
         let reg = registry().await;
         let (up, _rx) = unbounded_channel();
         let calls = std::cell::RefCell::new(Vec::new());
-        let bring_up = |_: AccountCfg, _: Option<String>| {
+        let bring_up = |_: AccountCfg| {
             calls.borrow_mut().push(t0.elapsed().as_secs() / 60);
-            async { Err(anyhow!("homeserver down")) as BringUp }
+            async { Err(anyhow!("homeserver down")) as Restore }
         };
 
         let _ = tokio::time::timeout(

hive-matrix-mcp/src/timeline.rs

@@ -60,10 +60,7 @@ pub async fn sweep_unread<S: BuildHasher>(
             .collect()
     };
     for id in stale {
-        if wake::send_todo_clear(Some(id.as_str()), false)
+        if wake::send_todo_clear(id.as_str()).await.is_ok() {
-            .await
-            .is_ok()
-        {
             notified.lock().await.remove(&id);
         }
     }
@@ -121,10 +118,7 @@ pub async fn sweep_invites<S: BuildHasher>(
             .collect()
     };
     for id in stale {
-        if wake::send_todo_clear(Some(&invite_key(&id)), false)
+        if wake::send_todo_clear(&invite_key(&id)).await.is_ok() {
-            .await
-            .is_ok()
-        {
             notified.lock().await.remove(&id);
         }
     }

hive-matrix-mcp/src/wake.rs

@@ -23,7 +23,8 @@ const TODO_SOCKET_RETRY: Retry = Retry::None;
 /// go — distinct from the host-served control socket used by `send_wake`.
 /// `None` when unset/empty, in which case todo sends are a best-effort
 /// no-op (a standalone daemon without the harness socket).
-fn agent_socket() -> Option<std::path::PathBuf> {
+#[must_use]
+pub fn agent_socket() -> Option<std::path::PathBuf> {
     std::env::var_os("HIVE_AGENT_SOCKET")
         .filter(|v| !v.is_empty())
         .map(std::path::PathBuf::from)
@@ -52,26 +53,40 @@ pub async fn send_todo_upsert(key: &str, summary: impl AsRef<str>) -> Result<()>
     notify(&socket, &payload, TODO_SOCKET_RETRY).await
 }
 
-/// Clear matrix-subsystem todos on the harness's in-agent socket. `key =
+/// Clear the matrix-subsystem todo `key` on the harness's in-agent socket
-/// Some(room)` clears one room's todo (it was read); `all = true` wipes the
+/// (its room was read, or its invite resolved). Best-effort: a no-op when
-/// whole matrix set (cancel-and-recreate on daemon restart). Best-effort:
+/// `HIVE_AGENT_SOCKET` isn't configured.
-/// a no-op when `HIVE_AGENT_SOCKET` isn't configured.
 ///
 /// # 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(key: Option<&str>, all: bool) -> Result<()> {
+pub async fn send_todo_clear(key: &str) -> Result<()> {
     let Some(socket) = agent_socket() else {
         return Ok(());
     };
+    send_clear(&socket, Some(key), false).await
+}
+
+/// Wipe the whole matrix todo set on `socket` (cancel-and-recreate on daemon
+/// start). Takes the socket instead of reading `HIVE_AGENT_SOCKET`, so only
+/// code handed the path can wipe the set.
+///
+/// # Errors
+///
+/// As [`send_todo_clear`].
+pub async fn clear_all_todos(socket: &std::path::Path) -> Result<()> {
+    send_clear(socket, None, true).await
+}
+
+async fn send_clear(socket: &std::path::Path, key: Option<&str>, all: bool) -> Result<()> {
     let payload = serde_json::json!({
         "cmd": "clear_todo",
         "subsystem": "matrix",
         "key": key,
         "all": all,
     });
-    notify(&socket, &payload, TODO_SOCKET_RETRY).await
+    notify(socket, &payload, TODO_SOCKET_RETRY).await
 }
 
 /// Format a wake-message body from a list of per-room unread summaries.