feat(#2659): serve hive-matrix-mcp over persistent streamable-http, drop stdio bridge

hive-matrix-mcp/src/accounts.rs

@@ -274,7 +274,8 @@ impl Registry {
     /// primary flag. Registry membership == a session restored, so every
     /// entry is reported `live`. Sorted primary-first then by name for a
     /// stable order in the dashboard. Account-agnostic — the caller does
-    /// not resolve a single client (see `socket::dispatch`).
+    /// not resolve a single client (see `MatrixMcp::resolve` in
+    /// `crate::mcp`).
     #[must_use]
     pub fn list(&self) -> Vec<AccountStatus> {
         let mut out: Vec<AccountStatus> = self

hive-matrix-mcp/src/handlers.rs

@@ -1,13 +1,12 @@
-//! Per-tool dispatch — each daemon request from the MCP bridge resolves
-//! to one of these handlers. Returns a `DaemonResponse` shaped for the
-//! wire protocol (the MCP bridge unwraps `Ok { payload }` and returns
-//! the payload to claude; `Error { message }` becomes the tool-call
-//! error message claude sees).
+//! Per-tool dispatch — each MCP tool call in [`crate::mcp`] resolves to
+//! one of these handlers. Returns a [`DaemonResponse`] which
+//! `crate::mcp::render` turns into the tool-result string claude sees
+//! (`Ok { payload }` → pretty JSON, `Error { message }` → a "matrix
+//! error: …" prefixed string).
 //!
 //! Tool surface mirrors damocles-daemon's v0 set per the operator's call:
 //! `send_message`, `send_dm`, `send_reaction`, `send_reply`, `mark_read`,
-//! `send_redact`, `list_rooms`, `list_room_members`, `read_room`. Plus a
-//! `ping` for the MCP bridge's liveness probe.
+//! `send_redact`, `list_rooms`, `list_room_members`, `read_room`.
 
 use matrix_sdk::{
     Client,

hive-matrix-mcp/src/lib.rs

@@ -1,14 +1,17 @@
-//! `hive-matrix-mcp` library: shared types + helpers used by both the
-//! `hive-matrix-daemon` binary (long-running matrix-sdk Client) and the
-//! `hive-matrix-mcp` binary (stdio MCP bridge claude spawns per turn).
-//!
-//! The wire protocol between the two binaries lives in [`protocol`].
-//! Path helpers (token file, daemon socket) live in [`paths`].
+//! Shared library for the `hive-matrix-daemon` binary — the only binary
+//! this crate produces. The daemon owns the per-account matrix-sdk
+//! `Client` registry (built at startup, one per configured account) and
+//! serves its MCP tools directly over streamable-http (see [`mcp`]) —
+//! no stdio bridge, no per-turn respawn, no round-trip unix socket.
 //!
 //! Architecture rationale and tool surface mirror the existing
 //! `damocles-daemon` v0 set.
 
+pub mod accounts;
 pub mod client;
+pub mod handlers;
+pub mod mcp;
 pub mod paths;
 pub mod protocol;
+pub mod timeline;
 pub mod wake;

hive-matrix-mcp/src/main.rs

@@ -1,6 +1,8 @@
 //! `hive-matrix-daemon` binary — long-running matrix-sdk Client + sync
 //! loop per matrix account. Bridges incoming room events to hyperhive
-//! wake signals and serves the unix socket the stdio MCP bridge talks to.
+//! wake signals and serves its MCP tools directly over streamable-http
+//! on `--http <addr>` — no stdio bridge, no separate bin claude has to
+//! respawn every turn.
 //!
 //! Lifecycle:
 //! 1. Read the configured account list (`accounts::configured()` —
@@ -8,9 +10,9 @@
 //! 2. For each account: whoami probe → recover `user_id` + `device_id`
 //!    → restore matrix-sdk session (no login flow), install the
 //!    message-event handler, and spawn its own sync loop.
-//! 3. Serve the unix socket against an account→Client registry; each
-//!    MCP request routes to the account named in its `account` field
-//!    (the primary account when omitted).
+//! 3. Serve the MCP tools against an account→Client registry; each tool
+//!    call routes to the account named in its `account` arg (the
+//!    primary account when omitted).
 //!
 //! Standalone-degraded boot: the PRIMARY account having no token file →
 //! exit 0 cleanly so systemd's path-watcher restarts us once hive-c0re
@@ -27,19 +29,21 @@
 use std::sync::Arc;
 
 use anyhow::{Context, Result};
+use clap::Parser;
 use matrix_sdk::{Client, config::SyncSettings};
 
-mod accounts;
-mod client;
-mod handlers;
-mod paths;
-mod protocol;
-mod socket;
-mod timeline;
-mod wake;
+use hive_matrix_mcp::accounts::{AccountCfg, Registry};
+use hive_matrix_mcp::client::PermanentBringUpError;
+use hive_matrix_mcp::{accounts, client, mcp, paths, timeline, wake};
 
-use accounts::{AccountCfg, Registry};
-use client::PermanentBringUpError;
+#[derive(Parser)]
+#[command(name = "hive-matrix-daemon", about = "matrix-sdk client + MCP daemon")]
+struct Cli {
+    /// Serve the MCP tools over streamable-http on this address (e.g.
+    /// `127.0.0.1:8792`). Bind loopback only.
+    #[arg(long)]
+    http: std::net::SocketAddr,
+}
 
 /// A per-account sync loop, boxed so loops for N accounts can be driven
 /// concurrently on the main task. Deliberately NOT `Send`: matrix-sdk's
@@ -71,8 +75,8 @@ async fn main() -> Result<()> {
         .with_writer(std::io::stderr)
         .init();
 
+    let cli = Cli::parse();
     let cfgs = accounts::configured().context("read matrix account config")?;
-    let mcp_socket = paths::daemon_socket();
     let multi = cfgs.len() > 1;
     let primary = cfgs[0].name.clone();
     let mut registry = Registry::new(primary);
@@ -131,9 +135,9 @@ async fn main() -> Result<()> {
         tracing::warn!(error = %format!("{e:#}"), "failed to write matrix-accounts snapshot");
     }
 
-    // Serve the socket against the registry. Spawned before driving the
-    // sync loops so the MCP bridge can connect as soon as the first
-    // claude turn fires.
+    // Serve the MCP tools against the registry. Spawned before driving
+    // the sync loops so claude can reach the stable http URL as soon as
+    // the first turn fires.
     let registry = Arc::new(registry);
 
     // Heartbeat: periodically rewrite the accounts snapshot so its mtime
@@ -155,10 +159,11 @@ async fn main() -> Result<()> {
         }
     });
 
-    let socket_listener = mcp_socket.clone();
+    let http_addr = cli.http;
+    let mcp_registry = Arc::clone(&registry);
     tokio::spawn(async move {
-        if let Err(e) = socket::serve(&socket_listener, registry).await {
-            tracing::error!(error = %e, "mcp socket server exited");
+        if let Err(e) = mcp::serve_http(http_addr, mcp_registry).await {
+            tracing::error!(error = %e, "mcp http server exited");
         }
     });
 
@@ -298,7 +303,7 @@ async fn bring_up_account(
     // matrix todos so stale ones (rooms read / invites resolved while the
     // daemon was down) don't linger, then let the first sweep rebuild the
     // set to match current reality. Best-effort; the sweep converges.
-    let _ = crate::wake::send_todo_clear(None, true).await;
+    let _ = wake::send_todo_clear(None, true).await;
     let sync_loop: SyncLoop = Box::pin(async move {
         sync_client
             .sync_with_callback(SyncSettings::default(), move |_response| {

hive-matrix-mcp/src/mcp.rs

@@ -1,78 +1,43 @@
-//! `hive-matrix-mcp` binary — stdio MCP server claude spawns per turn.
-//! Thin protocol bridge: every tool call → connect to the daemon's
-//! unix socket → write a JSON request line → read the JSON response →
-//! return the payload (or the error message) to claude.
-//!
-//! No matrix-sdk dep at this entrypoint. The daemon owns the heavy
-//! Client + sync; the bridge is pure serde + tokio I/O. Means the MCP
-//! binary cold-starts in milliseconds even though the daemon takes
-//! seconds to bring up sync.
+//! MCP tool surface for `hive-matrix-daemon`, served directly over
+//! streamable-http — no stdio bridge, no per-turn respawn, no
+//! round-trip unix socket. The daemon already owns the account
+//! registry in-process (built at startup from the restored matrix-sdk
+//! `Client`s), so each tool call resolves `account` against
+//! [`crate::accounts::Registry`] and calls straight into
+//! [`crate::handlers`]. Mirrors `hive-bash-mcp::mcp`'s shape (and, one
+//! level up, `hive-agent-mcp::mcp::serve_http`): a persistent daemon,
+//! stable URL claude reconnects to every turn instead of respawning a
+//! stdio child.
 //!
 //! Multi-account: every tool carries an optional `account` arg naming
 //! which matrix account to act as (a `name` from
 //! `hyperhive.matrixAccounts`); omitting it selects the agent's primary
-//! account. The bridge wraps each operation in a [`DaemonRequest`]
-//! envelope carrying that account; the daemon routes to the matching
-//! client.
+//! account (or errors, listing the choices, when more than one account
+//! is configured — see [`crate::accounts::Registry::resolve`]).
+
+use std::sync::Arc;
 
-use anyhow::{Context, Result};
 use rmcp::{
-    ServerHandler, ServiceExt,
+    ServerHandler,
     handler::server::wrapper::Parameters,
     schemars::{self, JsonSchema},
     tool, tool_handler, tool_router,
-    transport::stdio,
 };
 use serde::Deserialize;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
-use tokio::net::UnixStream;
 
-use hive_matrix_mcp::paths;
-use hive_matrix_mcp::protocol::{DaemonOp, DaemonRequest, DaemonResponse, InviteAction};
-
-/// Send `req` to the daemon and read back the response. Each call is a
-/// fresh unix-socket connection — short-lived (the daemon dispatch is
-/// a single round-trip) so connection pooling would be over-engineering.
-async fn round_trip(req: DaemonRequest) -> Result<DaemonResponse> {
-    let socket = paths::daemon_socket();
-    let stream = UnixStream::connect(&socket).await.with_context(|| {
-        format!(
-            "matrix daemon unreachable at {} — it may be starting up or restarting \
-             (the daemon rebinds its socket a few seconds after a restart); retry shortly",
-            socket.display()
-        )
-    })?;
-    let (reader, mut writer) = stream.into_split();
-    let mut line = serde_json::to_string(&req)?;
-    line.push('\n');
-    writer
-        .write_all(line.as_bytes())
-        .await
-        .context("write request to daemon socket")?;
-    writer.shutdown().await.ok();
-    let mut buf = String::new();
-    BufReader::new(reader)
-        .read_line(&mut buf)
-        .await
-        .context("read response from daemon socket")?;
-    serde_json::from_str(&buf).context("parse daemon response")
-}
-
-/// Wrap an op in a [`DaemonRequest`] envelope for `account` and send it.
-async fn call(account: Option<String>, op: DaemonOp) -> Result<DaemonResponse> {
-    round_trip(DaemonRequest { account, op }).await
-}
+use crate::accounts::Registry;
+use crate::handlers;
+use crate::protocol::{DaemonResponse, InviteAction};
 
 /// Turn a `DaemonResponse` into the string claude sees as the tool
 /// result. Ok payloads are pretty-printed JSON; errors get a clear
 /// "matrix error: …" prefix so claude can pattern-match on it.
-fn render(resp: Result<DaemonResponse>) -> String {
+fn render(resp: DaemonResponse) -> String {
     match resp {
-        Ok(DaemonResponse::Ok { payload }) => {
+        DaemonResponse::Ok { payload } => {
             serde_json::to_string_pretty(&payload).unwrap_or_else(|_| payload.to_string())
         }
-        Ok(DaemonResponse::Error { message }) => format!("matrix error: {message}"),
-        Err(e) => format!("matrix bridge error: {e:#}"),
+        DaemonResponse::Error { message } => format!("matrix error: {message}"),
     }
 }
 
@@ -282,25 +247,25 @@ struct InviteUserArgs {
     account: Option<String>,
 }
 
-struct MatrixBridge {
-    #[allow(
-        dead_code,
-        reason = "populated by the #[tool_router] macro; the generated \
-                  ServerHandler wiring consumes it, the field is never read directly"
-    )]
-    tool_router: rmcp::handler::server::router::tool::ToolRouter<Self>,
+#[derive(Clone)]
+struct MatrixMcp {
+    registry: Arc<Registry>,
 }
 
-impl MatrixBridge {
-    fn new() -> Self {
-        Self {
-            tool_router: Self::tool_router(),
-        }
+impl MatrixMcp {
+    /// Resolve `account` against the registry, rendering the "unknown
+    /// account" / "ambiguous, pick one" error the same way a handler
+    /// error would render (so a bad `account` arg and a bad room/event
+    /// arg look the same to claude).
+    fn resolve(&self, account: Option<&str>) -> Result<&matrix_sdk::Client, String> {
+        self.registry
+            .resolve(account)
+            .map(std::convert::AsRef::as_ref)
     }
 }
 
 #[tool_router]
-impl MatrixBridge {
+impl MatrixMcp {
     #[tool(
         description = "Post a plain-text or markdown message to a matrix room. \
             `room` is either a room id (!abc:server) or alias (#name:server). \
@@ -309,16 +274,11 @@ impl MatrixBridge {
             you don't talk over messages you haven't seen."
     )]
     async fn send_message(&self, Parameters(args): Parameters<SendMessageArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::SendMessage {
-                    room: args.room,
-                    body: args.body,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::send_message(client, &args.room, &args.body).await)
     }
 
     #[tool(description = "Open (or reuse) a direct message room with `user_id` \
@@ -326,16 +286,11 @@ impl MatrixBridge {
             and has unread messages, the send is rejected with a hint — read_room \
             then mark_read the latest event first.")]
     async fn send_dm(&self, Parameters(args): Parameters<SendDmArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::SendDm {
-                    user_id: args.user_id,
-                    body: args.body,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::send_dm(client, &args.user_id, &args.body).await)
     }
 
     #[tool(
@@ -346,17 +301,11 @@ impl MatrixBridge {
             room has unread messages — read_room then mark_read first."
     )]
     async fn send_file(&self, Parameters(args): Parameters<SendFileArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::SendFile {
-                    room: args.room,
-                    path: args.path,
-                    caption: args.caption,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::send_file(client, &args.room, &args.path, args.caption.as_deref()).await)
     }
 
     #[tool(description = "Resolve (find-or-create) the DM room with `user_id` \
@@ -364,15 +313,11 @@ impl MatrixBridge {
             returned room id with the room-based tools (`send_file`, `send_message`, \
             …) to deliver into the DM — there is no per-tool DM variant.")]
     async fn open_dm(&self, Parameters(args): Parameters<OpenDmArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::OpenDm {
-                    user_id: args.user_id,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::open_dm(client, &args.user_id).await)
     }
 
     #[tool(
@@ -381,17 +326,11 @@ impl MatrixBridge {
             standard clients."
     )]
     async fn send_reaction(&self, Parameters(args): Parameters<SendReactionArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::SendReaction {
-                    room: args.room,
-                    event_id: args.event_id,
-                    key: args.key,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::send_reaction(client, &args.room, &args.event_id, &args.key).await)
     }
 
     #[tool(description = "Reply to a specific matrix event in a room, threaded \
@@ -399,33 +338,22 @@ impl MatrixBridge {
             if the room still has unread messages — read_room then mark_read the \
             latest event first.")]
     async fn send_reply(&self, Parameters(args): Parameters<SendReplyArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::SendReply {
-                    room: args.room,
-                    event_id: args.event_id,
-                    body: args.body,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::send_reply(client, &args.room, &args.event_id, &args.body).await)
     }
 
     #[tool(description = "Mark a specific event as read for this agent. Updates \
             the room's unread indicator + sends a read receipt other \
             participants can see.")]
     async fn mark_read(&self, Parameters(args): Parameters<MarkReadArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::MarkRead {
-                    room: args.room,
-                    event_id: args.event_id,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::mark_read(client, &args.room, &args.event_id).await)
     }
 
     #[tool(description = "Redact (delete) a specific matrix event in a room — \
@@ -433,16 +361,12 @@ impl MatrixBridge {
             `reason`. Works on your own events; redacting others' needs moderator \
             power level. Irreversible.")]
     async fn send_redact(&self, Parameters(args): Parameters<SendRedactArgs>) -> String {
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
         render(
-            call(
-                args.account,
-                DaemonOp::SendRedact {
-                    room: args.room,
-                    event_id: args.event_id,
-                    reason: args.reason,
-                },
-            )
-            .await,
+            handlers::send_redact(client, &args.room, &args.event_id, args.reason.as_deref()).await,
         )
     }
 
@@ -451,7 +375,11 @@ impl MatrixBridge {
             id, canonical alias (when set), display name, and joined-member count."
     )]
     async fn list_rooms(&self, Parameters(args): Parameters<ListRoomsArgs>) -> String {
-        render(call(args.account, DaemonOp::ListRooms).await)
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::list_rooms(client).await)
     }
 
     #[tool(
@@ -460,7 +388,11 @@ impl MatrixBridge {
             Use `resolve_invite` to accept or reject an invite."
     )]
     async fn list_invites(&self, Parameters(args): Parameters<ListInvitesArgs>) -> String {
-        render(call(args.account, DaemonOp::ListInvites).await)
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::list_invites(client))
     }
 
     #[tool(
@@ -471,7 +403,11 @@ impl MatrixBridge {
             `list_rooms`."
     )]
     async fn join_room(&self, Parameters(args): Parameters<JoinRoomArgs>) -> String {
-        render(call(args.account, DaemonOp::JoinRoom { room: args.room }).await)
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::join_room(client, &args.room).await)
     }
 
     #[tool(
@@ -481,6 +417,10 @@ impl MatrixBridge {
             invites with `list_invites`."
     )]
     async fn resolve_invite(&self, Parameters(args): Parameters<ResolveInviteArgs>) -> String {
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
         let action = match args.action.trim().to_ascii_lowercase().as_str() {
             "accept" => InviteAction::Accept,
             "reject" => InviteAction::Reject,
@@ -490,16 +430,7 @@ impl MatrixBridge {
                 );
             }
         };
-        render(
-            call(
-                args.account,
-                DaemonOp::ResolveInvite {
-                    room: args.room,
-                    action,
-                },
-            )
-            .await,
-        )
+        render(handlers::resolve_invite(client, &args.room, action).await)
     }
 
     #[tool(
@@ -509,16 +440,11 @@ impl MatrixBridge {
             invite. The invitee then sees a pending invite they accept with `join_room`."
     )]
     async fn invite_user(&self, Parameters(args): Parameters<InviteUserArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::InviteUser {
-                    room: args.room,
-                    user_id: args.user_id,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::invite_user(client, &args.room, &args.user_id).await)
     }
 
     #[tool(
@@ -526,7 +452,11 @@ impl MatrixBridge {
             Each row carries the user id and resolved display name."
     )]
     async fn list_room_members(&self, Parameters(args): Parameters<ListRoomMembersArgs>) -> String {
-        render(call(args.account, DaemonOp::ListRoomMembers { room: args.room }).await)
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::list_room_members(client, &args.room).await)
     }
 
     #[tool(description = "Read events from a matrix room (default 50, max 200), \
@@ -539,18 +469,11 @@ impl MatrixBridge {
             outside the current window. `from` and `until` are mutually \
             exclusive.")]
     async fn read_room(&self, Parameters(args): Parameters<ReadRoomArgs>) -> String {
-        render(
-            call(
-                args.account,
-                DaemonOp::ReadRoom {
-                    room: args.room,
-                    limit: args.limit,
-                    from: args.from,
-                    until: args.until,
-                },
-            )
-            .await,
-        )
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
+        render(handlers::read_room(client, &args.room, args.limit, args.from, args.until).await)
     }
 
     #[tool(
@@ -561,14 +484,16 @@ impl MatrixBridge {
             counterpart of send_file."
     )]
     async fn download_file(&self, Parameters(args): Parameters<DownloadFileArgs>) -> String {
+        let client = match self.resolve(args.account.as_deref()) {
+            Ok(c) => c,
+            Err(e) => return format!("matrix error: {e}"),
+        };
         render(
-            call(
-                args.account,
-                DaemonOp::DownloadFile {
-                    room: args.room,
-                    event_id: args.event_id,
-                    dest_path: args.dest_path,
-                },
+            handlers::download_file(
+                client,
+                &args.room,
+                &args.event_id,
+                args.dest_path.as_deref(),
             )
             .await,
         )
@@ -598,52 +523,73 @@ impl MatrixBridge {
         all rejected if the room has unread messages — always call `read_room` \
         then `mark_read` on the latest event before sending to a room you \
         haven't read yet.")]
-impl ServerHandler for MatrixBridge {}
+impl ServerHandler for MatrixMcp {}
 
-#[tokio::main]
-async fn main() -> Result<()> {
-    tracing_subscriber::fmt()
-        .with_env_filter(
-            tracing_subscriber::EnvFilter::try_from_default_env()
-                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")),
+/// Plain-JSON status endpoint for the primary account's unread rooms —
+/// NOT part of the claude-facing MCP tool surface. `hive-agent-mcp`'s
+/// `get_loose_ends` hits this directly (same container, loopback only)
+/// to prepend a matrix-unread entry, mirroring the pre-http unix-socket
+/// `unread_summary` side channel the daemon used to serve. Best-effort:
+/// only resolves when exactly one account is configured (same rule as
+/// an MCP tool call omitting `account`) — a multi-account agent's extra
+/// accounts aren't reachable from here, same restriction the caller
+/// already documents for cross-agent queries.
+async fn unread_summary_handler(
+    axum::extract::State(registry): axum::extract::State<Arc<Registry>>,
+) -> axum::Json<serde_json::Value> {
+    let payload = match registry.resolve(None) {
+        Ok(client) => handlers::unread_summary(client.as_ref()).await,
+        Err(message) => crate::protocol::DaemonResponse::error(message),
+    };
+    axum::Json(
+        serde_json::to_value(payload)
+            .unwrap_or_else(|e| serde_json::json!({ "kind": "error", "message": e.to_string() })),
+    )
+}
+
+/// Run the MCP server over HTTP (rmcp streamable-http transport) on
+/// `addr`, dispatching against `registry`. Also serves a small
+/// non-MCP `/unread-summary` status endpoint (see
+/// [`unread_summary_handler`]).
+///
+/// Sole transport — there is no stdio mode. Long-lived so claude
+/// reconnects to the stable URL each turn instead of respawning a
+/// stdio child; since the daemon already owns the account registry
+/// in-process, tool calls need no round-trip to anywhere.
+///
+/// Binds loopback only in practice; the default `allowed_hosts`
+/// (`localhost`/`127.0.0.1`/`::1`) rejects Host headers from anywhere
+/// else for `/mcp`; `/unread-summary` is plain axum with no such
+/// guard, but the listener itself is loopback-only so this is moot.
+///
+/// # Errors
+///
+/// Returns an error if the listener cannot bind `addr` or the HTTP
+/// server exits with a fatal error.
+pub async fn serve_http(addr: std::net::SocketAddr, registry: Arc<Registry>) -> anyhow::Result<()> {
+    use rmcp::transport::streamable_http_server::{
+        StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
+    };
+    let session_manager = std::sync::Arc::new(LocalSessionManager::default());
+    let mcp_registry = registry.clone();
+    let service = StreamableHttpService::new(
+        move || {
+            Ok(MatrixMcp {
+                registry: mcp_registry.clone(),
+            })
+        },
+        session_manager,
+        StreamableHttpServerConfig::default(),
+    );
+    let app = axum::Router::new()
+        .nest_service("/mcp", service)
+        .route(
+            "/unread-summary",
+            axum::routing::get(unread_summary_handler),
         )
-        .with_writer(std::io::stderr)
-        .init();
-
-    // Standalone-degraded boot: matrix isn't provisioned for this agent
-    // (no token file) → exit 0 cleanly so claude doesn't register a
-    // matrix MCP server it can never use.
-    //
-    // Gate on the TOKEN, not the daemon socket. The daemon binds its
-    // socket only AFTER restoring its matrix session (~10s on a cold
-    // boot), so an exists-check on the socket here raced the daemon's
-    // startup: during that window the socket was absent, the bridge
-    // exited, and claude lost the matrix tools for the WHOLE session
-    // (the bridge isn't respawned mid-turn). The token, by contrast, is
-    // written by hive-c0re at provisioning time and is present well
-    // before the daemon finishes booting — so it cleanly distinguishes
-    // "matrix not set up for this agent" (token absent → exit) from
-    // "daemon still coming up" (token present → keep serving). When the
-    // token exists we serve regardless of socket state: tool calls
-    // `connect()` per-call and simply error until the daemon is up, but
-    // the tools stay registered for the session.
-    let token_file = paths::token_file();
-    if !tokio::fs::try_exists(&token_file).await.unwrap_or(false) {
-        tracing::warn!(
-            path = %token_file.display(),
-            "matrix not provisioned (no token file); exiting cleanly so MCP startup doesn't fail"
-        );
-        return Ok(());
-    }
-
-    let bridge = MatrixBridge::new();
-    let service = bridge
-        .serve(stdio())
-        .await
-        .context("serve MCP over stdio")?;
-    service
-        .waiting()
-        .await
-        .context("MCP service exited unexpectedly")?;
+        .with_state(registry);
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+    tracing::info!(%addr, "serving hive-matrix MCP over streamable-http at /mcp");
+    axum::serve(listener, app).await?;
     Ok(())
 }

hive-matrix-mcp/src/paths.rs

@@ -1,5 +1,4 @@
-//! Per-agent filesystem paths used by both `hive-matrix-daemon` and the
-//! stdio MCP bridge.
+//! Per-agent filesystem paths used by `hive-matrix-daemon`.
 //!
 //! All paths are overridable via env vars for dev / test scenarios and
 //! to let the harness point the daemon at non-default locations when
@@ -13,14 +12,6 @@ use std::path::PathBuf;
 /// `localhost` to the same machine.
 pub const DEFAULT_HOMESERVER: &str = "http://localhost:8008";
 
-/// Default unix socket path the daemon listens on inside the agent
-/// container. The stdio MCP bridge `connect()`s here on every tool call.
-/// Lives under systemd's `RuntimeDirectory=hive-matrix` (a tmpfs path
-/// that disappears on container restart — fine, because the daemon
-/// recreates the socket on its own boot) so the agent unix user
-/// can bind a socket inside it without root in `/run`.
-pub const DEFAULT_DAEMON_SOCKET: &str = "/run/hive-matrix/socket";
-
 /// Resolve the matrix access-token file path. Override via
 /// `HIVE_MATRIX_TOKEN_FILE`; default is `<HYPERHIVE_STATE_DIR>/matrix-token`,
 /// the path `hive-c0re::matrix::ensure_user_for` writes to on agent
@@ -41,14 +32,6 @@ pub fn homeserver_url() -> String {
     std::env::var("HIVE_MATRIX_URL").unwrap_or_else(|_| DEFAULT_HOMESERVER.to_owned())
 }
 
-/// Resolve the daemon's unix socket path. Override via
-/// `HIVE_MATRIX_SOCKET`; default is `/run/hive-matrix/socket`.
-#[must_use]
-pub fn daemon_socket() -> PathBuf {
-    std::env::var_os("HIVE_MATRIX_SOCKET")
-        .map_or_else(|| PathBuf::from(DEFAULT_DAEMON_SOCKET), PathBuf::from)
-}
-
 /// Persistent sqlite store directory for matrix-sdk's state (event
 /// cache, devices, etc.). Lives under the per-agent state dir so it
 /// survives container restart but gets wiped on `destroy --purge`.

hive-matrix-mcp/src/protocol.rs

@@ -1,208 +1,18 @@
-//! Wire types for the daemon ↔ stdio-MCP-bridge unix socket protocol.
+//! Shared response/DTO shapes for the matrix tool surface.
 //!
-//! Same shape as `damocles-daemon`'s `DaemonRequest`/`DaemonResponse`
-//! (which this is forked from in spirit). Each request is a single
-//! JSON line; each response is a single JSON line back. The stdio MCP
-//! bridge holds a fresh connection per tool call — claude's tool
-//! lifecycle is shorter than a persistent matrix-sdk Client wants to
-//! live, so the daemon stays alive and the MCP reconnects per call.
+//! `hive-matrix-daemon` serves its MCP tools directly over
+//! streamable-http (see [`crate::mcp`]) — there is no separate bridge
+//! process and no wire protocol between two binaries any more, so this
+//! module carries only the handler-facing result type
+//! ([`DaemonResponse`]) and small DTOs ([`InviteAction`],
+//! [`RoomUnread`]) shared between [`crate::handlers`] and its callers
+//! (the MCP tool router, the wake-signal formatter).
 
 use serde::{Deserialize, Serialize};
 
-/// Request envelope from the stdio MCP bridge to the daemon: which
-/// matrix `account` to act as, plus the operation itself. The daemon
-/// holds an account→Client registry (one client per declared matrix
-/// account) and routes `op` to the resolved client.
-///
-/// `account` is nested rather than flattened onto [`DaemonOp`] so we
-/// dodge the serde "internally-tagged enum + `#[serde(flatten)]`"
-/// edge cases; the bridge and daemon ship together so the wire shape
-/// is private. Wire:
-/// `{"account":"ccc","op":{"method":"send_message","room":…,"body":…}}`.
-#[derive(Debug, Serialize, Deserialize)]
-pub struct DaemonRequest {
-    /// Logical account name to act as (matches a `name` in
-    /// `hyperhive.matrixAccounts`). `None` selects the primary account
-    /// (the first declared one / the single legacy account), so
-    /// single-account callers omit it entirely.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub account: Option<String>,
-    /// The matrix operation to perform on the resolved account.
-    pub op: DaemonOp,
-}
-
-/// The matrix operation a [`DaemonRequest`] carries. The MCP bridge
-/// owns the on-wire shape claude sees; this enum is the internal
-/// shape the daemon dispatches over.
-#[derive(Debug, Serialize, Deserialize)]
-#[serde(tag = "method")]
-pub enum DaemonOp {
-    /// Post a plain-text or markdown message to a room. `room` accepts
-    /// either a matrix room id (`!abc:server`) or a canonical alias
-    /// (`#name:server`); the daemon resolves aliases server-side.
-    #[serde(rename = "send_message")]
-    SendMessage { room: String, body: String },
-
-    /// Open (or reuse) a DM with `user_id` and post `body`. Creates
-    /// the DM room if one doesn't already exist between this agent
-    /// and the user.
-    #[serde(rename = "send_dm")]
-    SendDm { user_id: String, body: String },
-
-    /// Upload a local file and post it as an attachment to `room`
-    /// (id or alias). `caption`, when set, is sent as a follow-up
-    /// text message in the same room.
-    #[serde(rename = "send_file")]
-    SendFile {
-        room: String,
-        path: String,
-        caption: Option<String>,
-    },
-
-    /// Resolve (find-or-create) the DM room with `user_id` and return
-    /// its room id, without sending anything. Lets a caller obtain the
-    /// DM room id and then use the room-based tools (`send_file`,
-    /// `send_message`, …) against it — so there is no per-tool `_dm`
-    /// variant.
-    #[serde(rename = "open_dm")]
-    OpenDm { user_id: String },
-
-    /// React to a specific event with an emoji `key`. Matrix-spec
-    /// `m.reaction` annotation.
-    #[serde(rename = "send_reaction")]
-    SendReaction {
-        room: String,
-        event_id: String,
-        key: String,
-    },
-
-    /// Reply to `event_id` in `room` with `body` as a threaded reply.
-    /// Sets the `m.in_reply_to` relation so matrix clients render the
-    /// thread.
-    #[serde(rename = "send_reply")]
-    SendReply {
-        room: String,
-        event_id: String,
-        body: String,
-    },
-
-    /// Mark `event_id` (in `room`) as read for this agent. Sends a
-    /// read receipt; bumps the room's "unread" indicator down on
-    /// matrix clients (and for other agents).
-    #[serde(rename = "mark_read")]
-    MarkRead { room: String, event_id: String },
-
-    /// Redact `event_id` in `room` — ask the homeserver to strip the
-    /// event's content (matrix-spec `m.room.redaction`), optionally with
-    /// a human-readable `reason`. The agent must have a high enough power
-    /// level (its own events, or moderator rights for others'); the
-    /// server rejects otherwise.
-    #[serde(rename = "send_redact")]
-    SendRedact {
-        room: String,
-        event_id: String,
-        reason: Option<String>,
-    },
-
-    /// List rooms the agent has joined. Returns each room's id +
-    /// canonical alias (when present) + name + member count.
-    #[serde(rename = "list_rooms")]
-    ListRooms,
-
-    /// List the members of a room. Each entry carries the matrix
-    /// user id + the resolved display name (when set).
-    #[serde(rename = "list_room_members")]
-    ListRoomMembers { room: String },
-
-    /// Read events from a room's timeline. Caller gets each event's id,
-    /// sender, `server_ts`, type, and body (best-effort plain-text extraction
-    /// from `m.text` / `m.notice` etc.). With neither cursor, returns the last
-    /// `limit` events (newest-first). `from` / `until` anchor at an event id
-    /// (mutually exclusive): `until` reads the anchor + `limit-1` events before
-    /// it (into the past); `from` reads the anchor + `limit-1` events after it.
-    #[serde(rename = "read_room")]
-    ReadRoom {
-        room: String,
-        limit: Option<usize>,
-        #[serde(default)]
-        from: Option<String>,
-        #[serde(default)]
-        until: Option<String>,
-    },
-
-    /// Download the media attachment carried by `event_id` in `room`
-    /// and write it to a local file (`dest_path`, or a temp file named
-    /// after the attachment when omitted), returning the path. The
-    /// read-side counterpart of `send_file`.
-    #[serde(rename = "download_file")]
-    DownloadFile {
-        room: String,
-        event_id: String,
-        dest_path: Option<String>,
-    },
-
-    /// List rooms this agent has been invited to but not yet joined.
-    /// Returns each room's id, canonical alias (when present), and
-    /// display name.
-    #[serde(rename = "list_invites")]
-    ListInvites,
-
-    /// Join a room by id (`!abc:server`) or alias (`#name:server`).
-    /// Accepts a pending invite if one exists; also joins public rooms
-    /// the agent hasn't been explicitly invited to. After joining the
-    /// room will appear in `list_rooms`.
-    #[serde(rename = "join_room")]
-    JoinRoom { room: String },
-
-    /// Resolve a pending invite to `room` (id or alias) by either
-    /// accepting it (join) or rejecting it (decline + leave). For rooms
-    /// you were *invited* to; `join_room` is the path for joining a
-    /// public room you weren't invited to.
-    #[serde(rename = "resolve_invite")]
-    ResolveInvite { room: String, action: InviteAction },
-
-    /// Invite `user_id` (`@user:server`) to `room` (id or alias). The
-    /// calling agent must already be a member with a high enough power
-    /// level to invite. Idempotent-ish: inviting an already-joined or
-    /// already-invited user surfaces the matrix error from the server.
-    #[serde(rename = "invite_user")]
-    InviteUser { room: String, user_id: String },
-
-    /// Return the count of rooms with unread notifications. Used by
-    /// the harness `get_loose_ends` to surface unread matrix activity
-    /// without exposing message content.
-    #[serde(rename = "unread_count")]
-    UnreadCount,
-
-    /// 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. Used by
-    /// `get_loose_ends` and the wake-signal formatter.
-    #[serde(rename = "unread_summary")]
-    UnreadSummary,
-
-    /// List the matrix accounts the daemon currently has a live,
-    /// restored session for. Account-agnostic (does not resolve a single
-    /// client — handled before client resolution in `socket::dispatch`):
-    /// returns each restored account's name, homeserver, user id, primary
-    /// flag, and a `live` flag. Backs the dashboard's per-account status
-    /// (BE-4) — turns BE-1's token-present list into true online/offline
-    /// + backfills the homeserver BE-1 leaves null.
-    #[serde(rename = "list_accounts")]
-    ListAccounts,
-
-    /// Liveness probe — fast "are you up?" round-trip that doesn't
-    /// touch matrix-sdk. Not used by the in-tree stdio MCP bridge
-    /// (which surfaces a daemon-down condition as a normal tool-call
-    /// connect error); reserved for external clients that want an
-    /// explicit health check without doing real work.
-    #[serde(rename = "ping")]
-    Ping,
-}
-
-/// Whether to accept or reject a pending invite in
-/// [`DaemonRequest::ResolveInvite`]. Serialises as `"accept"` /
-/// `"reject"` on the wire.
+/// Whether to accept or reject a pending invite (`resolve_invite`
+/// tool). Serialises as `"accept"` / `"reject"` on the wire (kept
+/// `Serialize`/`Deserialize` for the JSON DTOs handlers build).
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum InviteAction {
@@ -212,7 +22,7 @@ pub enum InviteAction {
     Reject,
 }
 
-/// One entry in the [`DaemonRequest::UnreadSummary`] response payload.
+/// One entry in the `unread_summary` response payload.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct RoomUnread {
     /// Canonical alias (`#name:server`) or room id (`!id:server`).
@@ -229,9 +39,13 @@ pub struct RoomUnread {
     pub last_sender: Option<String>,
 }
 
-/// Response shape: `ok` carries the payload (any JSON; the MCP bridge
-/// passes it back to claude as the tool result), `error` carries a
-/// human-readable error string.
+/// Result shape every [`crate::handlers`] function returns: `Ok`
+/// carries the payload (any JSON; the MCP tool router renders it as
+/// the tool result string), `Error` carries a human-readable error
+/// string. Kept as a distinct type (rather than each handler
+/// returning a bare `String`) so the tool router can uniformly render
+/// success vs error without every handler duplicating that
+/// formatting.
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum DaemonResponse {
@@ -258,117 +72,3 @@ impl DaemonResponse {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn envelope_round_trips_with_account() {
-        let req = DaemonRequest {
-            account: Some("ccc".to_owned()),
-            op: DaemonOp::SendMessage {
-                room: "!r:s".to_owned(),
-                body: "hi".to_owned(),
-            },
-        };
-        let line = serde_json::to_string(&req).unwrap();
-        // account + nested tagged op present on the wire.
-        assert!(line.contains("\"account\":\"ccc\""), "wire: {line}");
-        assert!(line.contains("\"method\":\"send_message\""), "wire: {line}");
-        let back: DaemonRequest = serde_json::from_str(&line).unwrap();
-        assert_eq!(back.account.as_deref(), Some("ccc"));
-        matches!(back.op, DaemonOp::SendMessage { .. });
-    }
-
-    #[test]
-    fn envelope_defaults_account_to_none_and_omits_it() {
-        // Single-account callers send no `account`; it must default to
-        // None and not appear on the wire (skip_serializing_if).
-        let req = DaemonRequest {
-            account: None,
-            op: DaemonOp::ListRooms,
-        };
-        let line = serde_json::to_string(&req).unwrap();
-        assert!(
-            !line.contains("account"),
-            "wire should omit account: {line}"
-        );
-        let back: DaemonRequest = serde_json::from_str(&line).unwrap();
-        assert!(back.account.is_none());
-        matches!(back.op, DaemonOp::ListRooms);
-    }
-
-    #[test]
-    fn send_redact_round_trips_with_reason() {
-        let req = DaemonRequest {
-            account: None,
-            op: DaemonOp::SendRedact {
-                room: "!r:s".to_owned(),
-                event_id: "$e".to_owned(),
-                reason: Some("spam".to_owned()),
-            },
-        };
-        let line = serde_json::to_string(&req).unwrap();
-        assert!(line.contains("\"method\":\"send_redact\""), "wire: {line}");
-        let back: DaemonRequest = serde_json::from_str(&line).unwrap();
-        match back.op {
-            DaemonOp::SendRedact {
-                room,
-                event_id,
-                reason,
-            } => {
-                assert_eq!(room, "!r:s");
-                assert_eq!(event_id, "$e");
-                assert_eq!(reason.as_deref(), Some("spam"));
-            }
-            other => panic!("wrong variant: {other:?}"),
-        }
-    }
-
-    #[test]
-    fn send_redact_round_trips_without_reason() {
-        let req = DaemonRequest {
-            account: None,
-            op: DaemonOp::SendRedact {
-                room: "!r:s".to_owned(),
-                event_id: "$e".to_owned(),
-                reason: None,
-            },
-        };
-        let line = serde_json::to_string(&req).unwrap();
-        let back: DaemonRequest = serde_json::from_str(&line).unwrap();
-        match back.op {
-            DaemonOp::SendRedact { reason, .. } => assert_eq!(reason, None),
-            other => panic!("wrong variant: {other:?}"),
-        }
-    }
-
-    #[test]
-    fn unit_variant_op_parses_inside_envelope() {
-        // A bare op with no fields still parses when wrapped.
-        let parsed: DaemonRequest =
-            serde_json::from_str(r#"{"op":{"method":"unread_count"}}"#).unwrap();
-        assert!(parsed.account.is_none());
-        matches!(parsed.op, DaemonOp::UnreadCount);
-    }
-
-    #[test]
-    fn list_accounts_parses_as_account_agnostic_unit_variant() {
-        // Registry-wide op: no fields, and callers omit `account`.
-        let parsed: DaemonRequest =
-            serde_json::from_str(r#"{"op":{"method":"list_accounts"}}"#).unwrap();
-        assert!(parsed.account.is_none());
-        matches!(parsed.op, DaemonOp::ListAccounts);
-        // And it serialises back to the same tagged shape.
-        let line = serde_json::to_string(&DaemonRequest {
-            account: None,
-            op: DaemonOp::ListAccounts,
-        })
-        .unwrap();
-        assert!(
-            line.contains("\"method\":\"list_accounts\""),
-            "wire: {line}"
-        );
-    }
-}

hive-matrix-mcp/src/socket.rs

@@ -1,143 +0,0 @@
-//! Unix socket server: the daemon listens here, the stdio MCP bridge
-//! `connect()`s on every tool call. One JSON request line in, one
-//! JSON response line out. Connections are short-lived (per tool call)
-//! so the loop is just accept → dispatch → reply → close.
-
-use std::path::Path;
-use std::sync::Arc;
-
-use anyhow::{Context, Result};
-use matrix_sdk::Client;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
-use tokio::net::{UnixListener, UnixStream};
-
-use crate::accounts::Registry;
-use crate::handlers;
-use crate::protocol::{DaemonOp, DaemonRequest, DaemonResponse};
-
-/// Start listening on `socket_path` and serve forever. Removes any
-/// stale socket file first (daemon restart after a non-clean shutdown
-/// would otherwise hit EADDRINUSE). `registry` resolves each request's
-/// `account` to the matrix client that serves it.
-pub async fn serve(socket_path: &Path, registry: Arc<Registry>) -> Result<()> {
-    let _ = tokio::fs::remove_file(socket_path).await;
-    if let Some(parent) = socket_path.parent() {
-        tokio::fs::create_dir_all(parent)
-            .await
-            .with_context(|| format!("mkdir {}", parent.display()))?;
-    }
-    let listener = UnixListener::bind(socket_path)
-        .with_context(|| format!("bind unix socket {}", socket_path.display()))?;
-    tracing::info!(path = %socket_path.display(), "mcp socket listener up");
-    loop {
-        let (stream, _) = listener
-            .accept()
-            .await
-            .context("accept connection on mcp socket")?;
-        let registry = registry.clone();
-        tokio::spawn(async move {
-            if let Err(e) = handle_connection(stream, &registry).await {
-                tracing::warn!(error = %e, "mcp socket connection error");
-            }
-        });
-    }
-}
-
-async fn handle_connection(stream: UnixStream, registry: &Registry) -> Result<()> {
-    let (reader, mut writer) = stream.into_split();
-    let mut lines = BufReader::new(reader).lines();
-    while let Some(line) = lines.next_line().await? {
-        let response = match serde_json::from_str::<DaemonRequest>(&line) {
-            Ok(req) => dispatch(req, registry).await,
-            Err(e) => DaemonResponse::error(format!("parse request: {e}")),
-        };
-        let mut json = serde_json::to_string(&response)?;
-        json.push('\n');
-        writer.write_all(json.as_bytes()).await?;
-        writer.flush().await?;
-    }
-    Ok(())
-}
-
-async fn dispatch(req: DaemonRequest, registry: &Registry) -> DaemonResponse {
-    // Ping is account-agnostic — answer without resolving a client so a
-    // health probe works even before any account restores.
-    if matches!(req.op, DaemonOp::Ping) {
-        return DaemonResponse::ok(&serde_json::json!({"ok": true}));
-    }
-    // ListAccounts is registry-wide, not per-account — answer before
-    // resolving a single client (the `account` field is meaningless for
-    // it, and it must work even if the primary failed to restore).
-    if matches!(req.op, DaemonOp::ListAccounts) {
-        return DaemonResponse::ok(&registry.list());
-    }
-    let client = match registry.resolve(req.account.as_deref()) {
-        Ok(c) => c,
-        Err(msg) => return DaemonResponse::error(msg),
-    };
-    dispatch_op(req.op, client).await
-}
-
-async fn dispatch_op(op: DaemonOp, client: &Client) -> DaemonResponse {
-    match op {
-        // Unreachable in practice: `dispatch` handles Ping before resolving
-        // a client (so a health probe works before any account restores).
-        // Kept for an exhaustive match.
-        DaemonOp::Ping => DaemonResponse::ok(&serde_json::json!({"ok": true})),
-        // Unreachable in practice: `dispatch` handles ListAccounts before
-        // resolving a client (it is registry-wide, not per-account). Kept
-        // for an exhaustive match — there is no client-scoped meaning.
-        DaemonOp::ListAccounts => DaemonResponse::error(
-            "list_accounts is registry-wide; handled before client resolution",
-        ),
-        DaemonOp::SendMessage { room, body } => handlers::send_message(client, &room, &body).await,
-        DaemonOp::SendDm { user_id, body } => handlers::send_dm(client, &user_id, &body).await,
-        DaemonOp::SendFile {
-            room,
-            path,
-            caption,
-        } => handlers::send_file(client, &room, &path, caption.as_deref()).await,
-        DaemonOp::OpenDm { user_id } => handlers::open_dm(client, &user_id).await,
-        DaemonOp::SendReaction {
-            room,
-            event_id,
-            key,
-        } => handlers::send_reaction(client, &room, &event_id, &key).await,
-        DaemonOp::SendReply {
-            room,
-            event_id,
-            body,
-        } => handlers::send_reply(client, &room, &event_id, &body).await,
-        DaemonOp::MarkRead { room, event_id } => {
-            handlers::mark_read(client, &room, &event_id).await
-        }
-        DaemonOp::SendRedact {
-            room,
-            event_id,
-            reason,
-        } => handlers::send_redact(client, &room, &event_id, reason.as_deref()).await,
-        DaemonOp::ListRooms => handlers::list_rooms(client).await,
-        DaemonOp::ListInvites => handlers::list_invites(client),
-        DaemonOp::JoinRoom { room } => handlers::join_room(client, &room).await,
-        DaemonOp::ResolveInvite { room, action } => {
-            handlers::resolve_invite(client, &room, action).await
-        }
-        DaemonOp::InviteUser { room, user_id } => {
-            handlers::invite_user(client, &room, &user_id).await
-        }
-        DaemonOp::ListRoomMembers { room } => handlers::list_room_members(client, &room).await,
-        DaemonOp::ReadRoom {
-            room,
-            limit,
-            from,
-            until,
-        } => handlers::read_room(client, &room, limit, from, until).await,
-        DaemonOp::DownloadFile {
-            room,
-            event_id,
-            dest_path,
-        } => handlers::download_file(client, &room, &event_id, dest_path.as_deref()).await,
-        DaemonOp::UnreadCount => handlers::unread_count(client),
-        DaemonOp::UnreadSummary => handlers::unread_summary(client).await,
-    }
-}

hive-matrix-mcp/src/timeline.rs

@@ -15,6 +15,7 @@
 //! never wakes on its own message.
 
 use std::collections::HashSet;
+use std::hash::BuildHasher;
 
 use matrix_sdk::{Client, ruma::OwnedRoomId};
 use tokio::sync::Mutex;
@@ -39,9 +40,9 @@ use crate::{handlers, wake};
 /// out by `collect_unread_with_ids` (after a rebuild a stale read receipt can
 /// otherwise leave a self-authored message counted as unread and self-wake
 /// the agent).
-pub async fn sweep_unread(
+pub async fn sweep_unread<S: BuildHasher>(
     client: &Client,
-    notified: &Mutex<HashSet<OwnedRoomId>>,
+    notified: &Mutex<HashSet<OwnedRoomId, S>>,
     account_tag: Option<&str>,
 ) {
     let unread = handlers::collect_unread_with_ids(client).await;
@@ -52,7 +53,11 @@ pub async fn sweep_unread(
     // 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()
+        active
+            .iter()
+            .filter(|id| !unread_ids.contains(*id))
+            .cloned()
+            .collect()
     };
     for id in stale {
         if wake::send_todo_clear(Some(id.as_str()), false)
@@ -95,9 +100,9 @@ pub async fn sweep_unread(
 /// than on every sync tick; it is pruned to the current invite set each
 /// pass so a withdrawn-then-reissued invite wakes again. The agent
 /// decides whether to accept or reject by calling `resolve_invite`.
-pub async fn sweep_invites(
+pub async fn sweep_invites<S: BuildHasher>(
     client: &Client,
-    notified: &Mutex<HashSet<OwnedRoomId>>,
+    notified: &Mutex<HashSet<OwnedRoomId, S>>,
     account_tag: Option<&str>,
 ) {
     let current = client.invited_rooms();
@@ -110,7 +115,10 @@ pub async fn sweep_invites(
     // tick on failure).
     let stale: Vec<OwnedRoomId> = {
         let seen = notified.lock().await;
-        seen.difference(&current_ids).cloned().collect()
+        seen.iter()
+            .filter(|id| !current_ids.contains(*id))
+            .cloned()
+            .collect()
     };
     for id in stale {
         if wake::send_todo_clear(Some(&invite_key(&id)), false)