diff --git a/hive-matrix-mcp/src/accounts.rs b/hive-matrix-mcp/src/accounts.rs deleted file mode 100644 index 9e95b06b..00000000 --- a/hive-matrix-mcp/src/accounts.rs +++ /dev/null @@ -1,141 +0,0 @@ -//! Multi-account configuration + the dispatch registry. -//! -//! A single `hive-matrix-daemon` can serve N matrix accounts (one -//! matrix-sdk `Client` each, with its own session/store dir + sync -//! loop). The account list comes from the `HIVE_MATRIX_ACCOUNTS` env -//! var (JSON, written by the nix harness module from -//! `hyperhive.matrixAccounts`); when that var is absent the daemon -//! synthesizes the single legacy account from the existing -//! `HIVE_MATRIX_*` env so every current single-account agent keeps -//! working with zero config change. -//! -//! The FIRST configured account is the **primary**: it is selected when -//! a tool call omits `account`, so single-account callers never specify -//! one. - -use std::collections::HashMap; -use std::path::PathBuf; -use std::sync::Arc; - -use matrix_sdk::Client; -use serde::Deserialize; - -use crate::paths; - -/// One declared matrix account. `homeserver` is optional per account -/// (defaults to the daemon-wide `HIVE_MATRIX_URL`) so accounts on the -/// same homeserver need not repeat it. -#[derive(Debug, Clone, Deserialize)] -pub struct AccountCfg { - /// Logical name the agent uses to address this account - /// (`account` arg on the MCP tools). Unique within the daemon. - pub name: String, - /// Path to the bearer-token file for this account (hive-c0re - /// writes it; the daemon reads it). - pub token_file: PathBuf, - /// Per-account matrix-sdk sqlite store dir (crypto keys + cache). - pub state_dir: PathBuf, - /// Homeserver URL; falls back to [`paths::homeserver_url`] when absent. - #[serde(default)] - pub homeserver: Option, -} - -impl AccountCfg { - /// Resolve the effective homeserver URL (per-account override or - /// the daemon-wide default). - #[must_use] - pub fn homeserver(&self) -> String { - self.homeserver - .clone() - .unwrap_or_else(paths::homeserver_url) - } -} - -/// Read the configured account list. Parses `HIVE_MATRIX_ACCOUNTS` -/// (JSON array) when set; otherwise returns the single legacy account -/// built from `HIVE_MATRIX_*` / `HYPERHIVE_STATE_DIR`. -/// -/// # Errors -/// -/// Returns an error if `HIVE_MATRIX_ACCOUNTS` is set but is not valid -/// JSON, is empty, or contains duplicate account names. -pub fn configured() -> anyhow::Result> { - let Some(raw) = std::env::var_os("HIVE_MATRIX_ACCOUNTS") else { - // Legacy single-account fallback — the only deployed shape until - // the nix `matrixAccounts` option lands. - return Ok(vec![AccountCfg { - name: "default".to_owned(), - token_file: paths::token_file(), - state_dir: paths::matrix_state_dir(), - homeserver: None, - }]); - }; - let raw = raw.to_string_lossy(); - let accounts: Vec = serde_json::from_str(&raw) - .map_err(|e| anyhow::anyhow!("parse HIVE_MATRIX_ACCOUNTS as JSON array: {e}"))?; - if accounts.is_empty() { - anyhow::bail!("HIVE_MATRIX_ACCOUNTS is an empty array — declare at least one account"); - } - let mut seen = std::collections::HashSet::new(); - for a in &accounts { - if !seen.insert(a.name.as_str()) { - anyhow::bail!( - "duplicate matrix account name {:?} in HIVE_MATRIX_ACCOUNTS", - a.name - ); - } - } - Ok(accounts) -} - -/// Account name → live `Client` map plus the primary-account name used -/// when a request omits `account`. Built once at daemon startup from -/// the accounts that successfully restored a session. -pub struct Registry { - primary: String, - by_name: HashMap>, -} - -impl Registry { - /// Build an empty registry whose primary is `primary`. Clients are - /// added with [`Registry::insert`] as each account restores. - #[must_use] - pub fn new(primary: String) -> Self { - Self { - primary, - by_name: HashMap::new(), - } - } - - /// Register a restored client under `name`. - pub fn insert(&mut self, name: String, client: Client) { - self.by_name.insert(name, Arc::new(client)); - } - - /// Whether any account restored successfully. - #[must_use] - pub fn is_empty(&self) -> bool { - self.by_name.is_empty() - } - - /// Resolve a request's `account` to a client. `None` → the primary - /// account. Returns a human-readable error (listing known accounts) - /// when the name is unknown — surfaced to the agent as a tool error. - /// - /// # Errors - /// - /// Errors when the named account (or the primary, if `None`) has no - /// live client — e.g. an unknown name, or the primary failed to - /// restore at startup. - pub fn resolve(&self, account: Option<&str>) -> Result<&Arc, String> { - let name = account.unwrap_or(&self.primary); - self.by_name.get(name).ok_or_else(|| { - let mut known: Vec<&str> = self.by_name.keys().map(String::as_str).collect(); - known.sort_unstable(); - format!( - "unknown matrix account {name:?}; available accounts: [{}]", - known.join(", ") - ) - }) - } -} diff --git a/hive-matrix-mcp/src/bin/mcp.rs b/hive-matrix-mcp/src/bin/mcp.rs index 6d5db39f..36c15fe9 100644 --- a/hive-matrix-mcp/src/bin/mcp.rs +++ b/hive-matrix-mcp/src/bin/mcp.rs @@ -7,13 +7,6 @@ //! 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. -//! -//! 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. use anyhow::{Context, Result}; use rmcp::{ @@ -28,7 +21,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; use hive_matrix_mcp::paths; -use hive_matrix_mcp::protocol::{DaemonOp, DaemonRequest, DaemonResponse, InviteAction}; +use hive_matrix_mcp::protocol::{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 @@ -54,11 +47,6 @@ async fn round_trip(req: DaemonRequest) -> Result { 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, op: DaemonOp) -> Result { - round_trip(DaemonRequest { account, op }).await -} - /// 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. @@ -80,10 +68,6 @@ struct SendMessageArgs { /// Message body. Markdown is rendered to HTML by the daemon /// (`text_markdown`); plain text passes through unchanged. body: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -93,10 +77,6 @@ struct SendDmArgs { /// the recipient. user_id: String, body: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -110,10 +90,6 @@ struct SendFileArgs { /// Optional caption, sent as a follow-up text message in the room. #[serde(default)] caption: Option, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -121,10 +97,6 @@ struct OpenDmArgs { /// Matrix user id (`@user:server`) to open a DM with. The DM room is /// created if one doesn't already exist. user_id: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -135,10 +107,6 @@ struct SendReactionArgs { /// Reaction key — usually an emoji (`👍`, `❤️`) but any string /// works per matrix spec. key: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -146,37 +114,20 @@ struct SendReplyArgs { room: String, event_id: String, body: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] struct MarkReadArgs { room: String, event_id: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] -struct ListRoomsArgs { - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, -} +struct ListRoomsArgs {} #[derive(Debug, Deserialize, JsonSchema)] struct ListRoomMembersArgs { room: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -185,10 +136,6 @@ struct ReadRoomArgs { /// Maximum events to return (default 50, max 200). Newest first. #[serde(default)] limit: Option, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -202,28 +149,15 @@ struct DownloadFileArgs { /// named after the attachment; the returned `path` is where to read it. #[serde(default)] dest_path: Option, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] -struct ListInvitesArgs { - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, -} +struct ListInvitesArgs {} #[derive(Debug, Deserialize, JsonSchema)] struct JoinRoomArgs { /// Matrix room id (`!abc:server`) or canonical alias (`#name:server`). room: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -234,10 +168,6 @@ struct ResolveInviteArgs { /// What to do with the invite: `"accept"` (join the room) or /// `"reject"` (decline and leave). action: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -247,10 +177,6 @@ struct InviteUserArgs { room: String, /// Matrix user id of the invitee (`@user:server`). user_id: String, - /// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`). - /// Omit to use the agent's primary account. - #[serde(default)] - account: Option, } struct MatrixBridge { @@ -281,13 +207,10 @@ impl MatrixBridge { )] async fn send_message(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::SendMessage { - room: args.room, - body: args.body, - }, - ) + round_trip(DaemonRequest::SendMessage { + room: args.room, + body: args.body, + }) .await, ) } @@ -298,13 +221,10 @@ impl MatrixBridge { then mark_read the latest event first.")] async fn send_dm(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::SendDm { - user_id: args.user_id, - body: args.body, - }, - ) + round_trip(DaemonRequest::SendDm { + user_id: args.user_id, + body: args.body, + }) .await, ) } @@ -318,14 +238,11 @@ impl MatrixBridge { )] async fn send_file(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::SendFile { - room: args.room, - path: args.path, - caption: args.caption, - }, - ) + round_trip(DaemonRequest::SendFile { + room: args.room, + path: args.path, + caption: args.caption, + }) .await, ) } @@ -336,12 +253,9 @@ impl MatrixBridge { …) to deliver into the DM — there is no per-tool DM variant.")] async fn open_dm(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::OpenDm { - user_id: args.user_id, - }, - ) + round_trip(DaemonRequest::OpenDm { + user_id: args.user_id, + }) .await, ) } @@ -353,14 +267,11 @@ impl MatrixBridge { )] async fn send_reaction(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::SendReaction { - room: args.room, - event_id: args.event_id, - key: args.key, - }, - ) + round_trip(DaemonRequest::SendReaction { + room: args.room, + event_id: args.event_id, + key: args.key, + }) .await, ) } @@ -371,14 +282,11 @@ impl MatrixBridge { latest event first.")] async fn send_reply(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::SendReply { - room: args.room, - event_id: args.event_id, - body: args.body, - }, - ) + round_trip(DaemonRequest::SendReply { + room: args.room, + event_id: args.event_id, + body: args.body, + }) .await, ) } @@ -388,13 +296,10 @@ impl MatrixBridge { participants can see.")] async fn mark_read(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::MarkRead { - room: args.room, - event_id: args.event_id, - }, - ) + round_trip(DaemonRequest::MarkRead { + room: args.room, + event_id: args.event_id, + }) .await, ) } @@ -403,8 +308,8 @@ impl MatrixBridge { description = "List rooms this agent has joined. Each row has the room \ id, canonical alias (when set), display name, and joined-member count." )] - async fn list_rooms(&self, Parameters(args): Parameters) -> String { - render(call(args.account, DaemonOp::ListRooms).await) + async fn list_rooms(&self, Parameters(_): Parameters) -> String { + render(round_trip(DaemonRequest::ListRooms).await) } #[tool( @@ -412,8 +317,8 @@ impl MatrixBridge { Each row has the room id, canonical alias (when set), and display name. \ Use `resolve_invite` to accept or reject an invite." )] - async fn list_invites(&self, Parameters(args): Parameters) -> String { - render(call(args.account, DaemonOp::ListInvites).await) + async fn list_invites(&self, Parameters(_): Parameters) -> String { + render(round_trip(DaemonRequest::ListInvites).await) } #[tool( @@ -424,7 +329,7 @@ impl MatrixBridge { `list_rooms`." )] async fn join_room(&self, Parameters(args): Parameters) -> String { - render(call(args.account, DaemonOp::JoinRoom { room: args.room }).await) + render(round_trip(DaemonRequest::JoinRoom { room: args.room }).await) } #[tool( @@ -444,13 +349,10 @@ impl MatrixBridge { } }; render( - call( - args.account, - DaemonOp::ResolveInvite { - room: args.room, - action, - }, - ) + round_trip(DaemonRequest::ResolveInvite { + room: args.room, + action, + }) .await, ) } @@ -463,13 +365,10 @@ impl MatrixBridge { )] async fn invite_user(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::InviteUser { - room: args.room, - user_id: args.user_id, - }, - ) + round_trip(DaemonRequest::InviteUser { + room: args.room, + user_id: args.user_id, + }) .await, ) } @@ -479,7 +378,7 @@ impl MatrixBridge { Each row carries the user id and resolved display name." )] async fn list_room_members(&self, Parameters(args): Parameters) -> String { - render(call(args.account, DaemonOp::ListRoomMembers { room: args.room }).await) + render(round_trip(DaemonRequest::ListRoomMembers { room: args.room }).await) } #[tool(description = "Read the most recent N events from a matrix room \ @@ -487,13 +386,10 @@ impl MatrixBridge { type, and best-effort plain-text body.")] async fn read_room(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::ReadRoom { - room: args.room, - limit: args.limit, - }, - ) + round_trip(DaemonRequest::ReadRoom { + room: args.room, + limit: args.limit, + }) .await, ) } @@ -507,14 +403,11 @@ impl MatrixBridge { )] async fn download_file(&self, Parameters(args): Parameters) -> String { render( - call( - args.account, - DaemonOp::DownloadFile { - room: args.room, - event_id: args.event_id, - dest_path: args.dest_path, - }, - ) + round_trip(DaemonRequest::DownloadFile { + room: args.room, + event_id: args.event_id, + dest_path: args.dest_path, + }) .await, ) } @@ -528,9 +421,7 @@ impl MatrixBridge { timeline with `read_room`. See pending invites with `list_invites`; \ accept or reject an invite with `resolve_invite`; join a public \ room with `join_room`. Room references accept ids (!abc:server) \ - or aliases (#name:server); user references use @user:server. Every \ - tool takes an optional `account` (a name from the agent's matrix \ - accounts) — omit it to act as the primary account.")] + or aliases (#name:server); user references use @user:server.")] impl ServerHandler for MatrixBridge {} #[tokio::main] diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 12efc512..32dd1ea0 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -1,31 +1,25 @@ //! `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. +//! loop per agent. Bridges incoming room events to hyperhive wake +//! signals and serves the unix socket the stdio MCP bridge talks to. //! //! Lifecycle: -//! 1. Read the configured account list (`accounts::configured()` — -//! `HIVE_MATRIX_ACCOUNTS` JSON, or the single legacy account). -//! 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). +//! 1. Read access token from `paths::token_file()` (fail clean if absent). +//! 2. Whoami probe → recover `user_id` + `device_id` → restore +//! matrix-sdk session (no login flow). +//! 3. Install the message-event handler that fires hyperhive wakes. +//! 4. Spawn the unix socket listener for the MCP bridge. +//! 5. Run sync forever. //! -//! Standalone-degraded boot: the PRIMARY account having no token file → -//! exit 0 cleanly so systemd's path-watcher restarts us once hive-c0re -//! provisions it. A SECONDARY account missing its token is skipped (the -//! daemon still serves the others). +//! Standalone-degraded boot: missing token file → exit 0 cleanly so +//! systemd's `ConditionPathExists=` doesn't have to be perfectly +//! synced with hive-c0re's token-provisioning timing. //! //! Stale-token recovery: handled in `client::build_and_restore` — see //! that module for the `M_UNKNOWN_TOKEN` detection + cleanup flow. -use std::sync::Arc; - use anyhow::{Context, Result}; -use matrix_sdk::{Client, config::SyncSettings}; +use matrix_sdk::config::SyncSettings; -mod accounts; mod client; mod handlers; mod paths; @@ -34,14 +28,6 @@ mod socket; mod timeline; mod wake; -use accounts::{AccountCfg, Registry}; - -/// 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 -/// sync future isn't `Send`, so these run on the current task (via -/// `select_all`) rather than `tokio::spawn`/`JoinSet`. -type SyncLoop = std::pin::Pin>>>; - #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -52,130 +38,68 @@ async fn main() -> Result<()> { .with_writer(std::io::stderr) .init(); - let cfgs = accounts::configured().context("read matrix account config")?; + let homeserver = paths::homeserver_url(); + let token_file = paths::token_file(); + let state_dir = paths::matrix_state_dir(); let mcp_socket = paths::daemon_socket(); let hyperhive_socket = paths::hyperhive_socket(); - let multi = cfgs.len() > 1; - let primary = cfgs[0].name.clone(); - let mut registry = Registry::new(primary); - let mut sync_loops: Vec = Vec::new(); - for (idx, cfg) in cfgs.into_iter().enumerate() { - let is_primary = idx == 0; - // Account-tag the wakes only in multi-account mode so single- - // account wake bodies stay byte-identical to the legacy format. - let tag = multi.then(|| cfg.name.clone()); - match bring_up_account(&cfg, &hyperhive_socket, tag).await { - Ok(Some((client, sync_loop))) => { - registry.insert(cfg.name, client); - sync_loops.push(sync_loop); - } - // No token yet: for the primary that means the daemon isn't - // useful — exit 0 like the legacy single-account path so the - // systemd path-watcher restarts us when the token appears. - Ok(None) if is_primary => { - tracing::warn!( - account = %cfg.name, - "primary matrix account has no token yet; exiting cleanly \ - (systemd restarts us when hive-c0re provisions it)" - ); - return Ok(()); - } - Ok(None) => { - tracing::warn!(account = %cfg.name, "secondary matrix account has no token; skipping"); - } - // The primary failing to restore is fatal (propagate so - // systemd retries on a transient blip — matches legacy - // behaviour); a secondary failing is logged and skipped. - Err(e) if is_primary => return Err(e.context("bring up primary matrix account")), - Err(e) => { - tracing::error!(account = %cfg.name, error = %format!("{e:#}"), "secondary matrix account failed to restore; skipping"); - } - } - } - - if registry.is_empty() { - tracing::warn!("no matrix accounts restored; exiting cleanly"); + if !tokio::fs::try_exists(&token_file).await.unwrap_or(false) { + tracing::warn!( + path = %token_file.display(), + "matrix token file absent; exiting cleanly (hive-c0re will provision \ + it on first agent registration, then systemd restarts us)" + ); return Ok(()); } - // 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. - let registry = Arc::new(registry); + tracing::info!( + homeserver, + token_file = %token_file.display(), + state_dir = %state_dir.display(), + "hive-matrix-daemon starting" + ); + + let matrix_client = client::build_and_restore(&homeserver, &token_file, &state_dir) + .await + .context("build matrix client")?; + + timeline::install_message_handler(&matrix_client, hyperhive_socket.clone()); + + // Spawn the unix socket server before sync starts so the MCP + // bridge can connect as soon as the first claude turn fires. The + // socket dispatches against the same `Client` we sync on, so any + // tool call benefits from the sync state-cache. + let socket_client = matrix_client.clone(); let socket_listener = mcp_socket.clone(); tokio::spawn(async move { - if let Err(e) = socket::serve(&socket_listener, registry).await { + if let Err(e) = socket::serve(&socket_listener, socket_client).await { tracing::error!(error = %e, "mcp socket server exited"); } }); - // Drive all per-account sync loops concurrently on this task (they - // aren't `Send`, so no `tokio::spawn`). matrix-sdk reconnects - // internally, so any loop returning is exceptional — log it and exit - // so systemd restarts the whole daemon cleanly. - let (result, idx, _rest) = futures_util::future::select_all(sync_loops).await; - match result { - Ok(()) => tracing::warn!( - account_index = idx, - "a matrix sync loop exited cleanly; restarting daemon" - ), - Err(e) => { - tracing::error!(account_index = idx, error = %format!("{e:#}"), "a matrix sync loop errored; restarting daemon"); - } - } + // Sync forever; matrix-sdk handles reconnection internally. After + // every sync, sweep pending invites and wake the agent for any new + // one: the StrippedRoomMemberEvent handler dispatched unreliably + // (cold-start invites + handler races produced no wake), so + // invite-waking lives on this post-sync sweep with a dedup set. + let invite_socket = std::sync::Arc::new(hyperhive_socket); + let invite_notified = + std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); + let sweep_client = matrix_client.clone(); + let sync_settings = SyncSettings::default(); + matrix_client + .sync_with_callback(sync_settings, move |_response| { + let client = sweep_client.clone(); + let socket = invite_socket.clone(); + let notified = invite_notified.clone(); + async move { + timeline::sweep_invites(&client, &socket, ¬ified).await; + matrix_sdk::LoopCtrl::Continue + } + }) + .await + .context("matrix-sdk sync loop exited")?; Ok(()) } - -/// Restore one account's client (when its token exists), install its -/// message handler, and build its sync loop. Returns -/// `Ok(Some((client, sync_loop)))` when the account came up, `Ok(None)` -/// when its token file is absent (caller decides primary-vs-secondary -/// handling). The returned sync loop is driven by the caller. -async fn bring_up_account( - cfg: &AccountCfg, - hyperhive_socket: &std::path::Path, - tag: Option, -) -> Result> { - let homeserver = cfg.homeserver(); - if !tokio::fs::try_exists(&cfg.token_file) - .await - .unwrap_or(false) - { - return Ok(None); - } - tracing::info!( - account = %cfg.name, - homeserver, - token_file = %cfg.token_file.display(), - state_dir = %cfg.state_dir.display(), - "bringing up matrix account" - ); - let client = client::build_and_restore(&homeserver, &cfg.token_file, &cfg.state_dir) - .await - .with_context(|| format!("build matrix client for account {}", cfg.name))?; - timeline::install_message_handler(&client, hyperhive_socket.to_path_buf(), tag.clone()); - - let sync_client = client.clone(); - let cb_client = client.clone(); - let invite_socket = Arc::new(hyperhive_socket.to_path_buf()); - let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new())); - let sync_loop: SyncLoop = Box::pin(async move { - sync_client - .sync_with_callback(SyncSettings::default(), move |_response| { - let client = cb_client.clone(); - let socket = invite_socket.clone(); - let notified = invite_notified.clone(); - let tag = tag.clone(); - async move { - timeline::sweep_invites(&client, &socket, ¬ified, tag.as_deref()).await; - matrix_sdk::LoopCtrl::Continue - } - }) - .await - .context("matrix-sdk sync loop exited")?; - Ok(()) - }); - Ok(Some((client, sync_loop))) -} diff --git a/hive-matrix-mcp/src/protocol.rs b/hive-matrix-mcp/src/protocol.rs index d072c02b..b847c3e4 100644 --- a/hive-matrix-mcp/src/protocol.rs +++ b/hive-matrix-mcp/src/protocol.rs @@ -9,34 +9,12 @@ 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, - /// The matrix operation to perform on the resolved account. - pub op: DaemonOp, -} - -/// The matrix operation a [`DaemonRequest`] carries. The MCP bridge +/// Request from the stdio MCP bridge to the daemon. 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 { +pub enum DaemonRequest { /// 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. @@ -226,53 +204,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 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); - } -} diff --git a/hive-matrix-mcp/src/socket.rs b/hive-matrix-mcp/src/socket.rs index 91b2d76a..78f8acd3 100644 --- a/hive-matrix-mcp/src/socket.rs +++ b/hive-matrix-mcp/src/socket.rs @@ -11,15 +11,13 @@ 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}; +use crate::protocol::{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) -> Result<()> { +/// would otherwise hit EADDRINUSE). +pub async fn serve(socket_path: &Path, client: Client) -> Result<()> { let _ = tokio::fs::remove_file(socket_path).await; if let Some(parent) = socket_path.parent() { tokio::fs::create_dir_all(parent) @@ -29,26 +27,27 @@ pub async fn serve(socket_path: &Path, registry: Arc) -> Result<()> { 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"); + let client = Arc::new(client); loop { let (stream, _) = listener .accept() .await .context("accept connection on mcp socket")?; - let registry = registry.clone(); + let client = client.clone(); tokio::spawn(async move { - if let Err(e) = handle_connection(stream, ®istry).await { + if let Err(e) = handle_connection(stream, &client).await { tracing::warn!(error = %e, "mcp socket connection error"); } }); } } -async fn handle_connection(stream: UnixStream, registry: &Registry) -> Result<()> { +async fn handle_connection(stream: UnixStream, client: &Client) -> 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::(&line) { - Ok(req) => dispatch(req, registry).await, + Ok(req) => dispatch(req, client).await, Err(e) => DaemonResponse::error(format!("parse request: {e}")), }; let mut json = serde_json::to_string(&response)?; @@ -59,63 +58,49 @@ async fn handle_connection(stream: UnixStream, registry: &Registry) -> Result<() 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})); - } - 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})), - 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 { +async fn dispatch(req: DaemonRequest, client: &Client) -> DaemonResponse { + match req { + DaemonRequest::Ping => DaemonResponse::ok(&serde_json::json!({"ok": true})), + DaemonRequest::SendMessage { room, body } => { + handlers::send_message(client, &room, &body).await + } + DaemonRequest::SendDm { user_id, body } => handlers::send_dm(client, &user_id, &body).await, + DaemonRequest::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 { + DaemonRequest::OpenDm { user_id } => handlers::open_dm(client, &user_id).await, + DaemonRequest::SendReaction { room, event_id, key, } => handlers::send_reaction(client, &room, &event_id, &key).await, - DaemonOp::SendReply { + DaemonRequest::SendReply { room, event_id, body, } => handlers::send_reply(client, &room, &event_id, &body).await, - DaemonOp::MarkRead { room, event_id } => { + DaemonRequest::MarkRead { room, event_id } => { handlers::mark_read(client, &room, &event_id).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 } => { + DaemonRequest::ListRooms => handlers::list_rooms(client).await, + DaemonRequest::ListInvites => handlers::list_invites(client), + DaemonRequest::JoinRoom { room } => handlers::join_room(client, &room).await, + DaemonRequest::ResolveInvite { room, action } => { handlers::resolve_invite(client, &room, action).await } - DaemonOp::InviteUser { room, user_id } => { + DaemonRequest::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 } => handlers::read_room(client, &room, limit).await, - DaemonOp::DownloadFile { + DaemonRequest::ListRoomMembers { room } => handlers::list_room_members(client, &room).await, + DaemonRequest::ReadRoom { room, limit } => handlers::read_room(client, &room, limit).await, + DaemonRequest::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, + DaemonRequest::UnreadCount => handlers::unread_count(client), + DaemonRequest::UnreadSummary => handlers::unread_summary(client).await, } } diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index a60fe479..41553827 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -27,24 +27,13 @@ use crate::{handlers, wake}; /// `hyperhive_socket`. The wake body summarises ALL rooms with unread /// notifications at wake time (not just the triggering event) so the /// agent receives a full picture in one prompt. -/// -/// `account_tag` is `Some(name)` only in multi-account mode (N>1 matrix -/// accounts on this daemon); when set it is prepended to the wake body -/// so the agent knows which account to `read_room` on. In the -/// single-account case it is `None` and the wake body is unchanged. -pub fn install_message_handler( - client: &Client, - hyperhive_socket: PathBuf, - account_tag: Option, -) { +pub fn install_message_handler(client: &Client, hyperhive_socket: PathBuf) { let socket = Arc::new(hyperhive_socket); - let account_tag = Arc::new(account_tag); let own_user = client.user_id().map(std::borrow::ToOwned::to_owned); client.add_event_handler({ let socket = socket.clone(); move |event: OriginalSyncRoomMessageEvent, room: Room, client: Client| { let socket = socket.clone(); - let account_tag = account_tag.clone(); let own_user = own_user.clone(); async move { // INFO so the live host journal shows the handler actually @@ -93,7 +82,6 @@ pub fn install_message_handler( } else { wake::format_unread_summary(&unread) }; - let body = wake::tag_account(account_tag.as_ref().as_deref(), body); if let Err(e) = wake::send_wake(&socket, &body).await { tracing::warn!(error = %e, "failed to deliver matrix wake to hyperhive"); } else { @@ -118,12 +106,7 @@ pub fn install_message_handler( /// 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( - client: &Client, - socket: &Path, - notified: &Mutex>, - account_tag: Option<&str>, -) { +pub async fn sweep_invites(client: &Client, socket: &Path, notified: &Mutex>) { let current = client.invited_rooms(); let current_ids: HashSet = current.iter().map(|r| r.room_id().to_owned()).collect(); @@ -151,12 +134,9 @@ pub async fn sweep_invites( let room_id = room.room_id().to_owned(); let label = room.name().unwrap_or_else(|| room_id.to_string()); tracing::info!(%room_id, "matrix: pending invite swept, waking agent"); - let body = wake::tag_account( - account_tag, - format!( - "[matrix] invited to {label} ({room_id}) — \ - use list_invites to see pending invites, resolve_invite to accept or reject" - ), + let body = format!( + "[matrix] invited to {label} ({room_id}) — \ + use list_invites to see pending invites, resolve_invite to accept or reject" ); if let Err(e) = wake::send_wake(socket, &body).await { tracing::warn!(error = %e, "matrix: failed to deliver invite-wake to hyperhive"); diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 8196b8c1..48d4ed53 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -119,18 +119,6 @@ pub fn format_unread_summary(rooms: &[crate::protocol::RoomUnread]) -> String { out } -/// Prepend an account marker to a wake `body` when the daemon serves -/// more than one matrix account. `tag` is `Some(name)` only in -/// multi-account mode; `None` returns `body` unchanged so single-account -/// wakes keep their exact format. Shape: `[acct:] `. -#[must_use] -pub fn tag_account(tag: Option<&str>, body: String) -> String { - match tag { - Some(name) => format!("[acct:{name}] {body}"), - None => body, - } -} - /// Truncate `s` to `max` Unicode chars, appending `…` when cut. /// Char-based not byte-based so multi-byte content (most chat) doesn't /// get cut mid-codepoint. @@ -189,19 +177,4 @@ mod tests { let s = "hi"; assert_eq!(truncate_chars(s, 100), "hi"); } - - #[test] - fn tag_account_none_is_passthrough() { - let body = "[matrix] @a:s in #x: hi".to_owned(); - assert_eq!(tag_account(None, body.clone()), body); - } - - #[test] - fn tag_account_some_prepends_marker() { - let body = "[matrix] @a:s in #x: hi".to_owned(); - assert_eq!( - tag_account(Some("ccc"), body), - "[acct:ccc] [matrix] @a:s in #x: hi" - ); - } }