diff --git a/hive-matrix-mcp/src/accounts.rs b/hive-matrix-mcp/src/accounts.rs new file mode 100644 index 00000000..9e95b06b --- /dev/null +++ b/hive-matrix-mcp/src/accounts.rs @@ -0,0 +1,141 @@ +//! 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 36c15fe9..6d5db39f 100644 --- a/hive-matrix-mcp/src/bin/mcp.rs +++ b/hive-matrix-mcp/src/bin/mcp.rs @@ -7,6 +7,13 @@ //! 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::{ @@ -21,7 +28,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; use hive_matrix_mcp::paths; -use hive_matrix_mcp::protocol::{DaemonRequest, DaemonResponse, InviteAction}; +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 @@ -47,6 +54,11 @@ 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. @@ -68,6 +80,10 @@ 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)] @@ -77,6 +93,10 @@ 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)] @@ -90,6 +110,10 @@ 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)] @@ -97,6 +121,10 @@ 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)] @@ -107,6 +135,10 @@ 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)] @@ -114,20 +146,37 @@ 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 {} +struct ListRoomsArgs { + /// 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 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)] @@ -136,6 +185,10 @@ 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)] @@ -149,15 +202,28 @@ 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 {} +struct ListInvitesArgs { + /// 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 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)] @@ -168,6 +234,10 @@ 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)] @@ -177,6 +247,10 @@ 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 { @@ -207,10 +281,13 @@ impl MatrixBridge { )] async fn send_message(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::SendMessage { - room: args.room, - body: args.body, - }) + call( + args.account, + DaemonOp::SendMessage { + room: args.room, + body: args.body, + }, + ) .await, ) } @@ -221,10 +298,13 @@ impl MatrixBridge { then mark_read the latest event first.")] async fn send_dm(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::SendDm { - user_id: args.user_id, - body: args.body, - }) + call( + args.account, + DaemonOp::SendDm { + user_id: args.user_id, + body: args.body, + }, + ) .await, ) } @@ -238,11 +318,14 @@ impl MatrixBridge { )] async fn send_file(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::SendFile { - room: args.room, - path: args.path, - caption: args.caption, - }) + call( + args.account, + DaemonOp::SendFile { + room: args.room, + path: args.path, + caption: args.caption, + }, + ) .await, ) } @@ -253,9 +336,12 @@ impl MatrixBridge { …) to deliver into the DM — there is no per-tool DM variant.")] async fn open_dm(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::OpenDm { - user_id: args.user_id, - }) + call( + args.account, + DaemonOp::OpenDm { + user_id: args.user_id, + }, + ) .await, ) } @@ -267,11 +353,14 @@ impl MatrixBridge { )] async fn send_reaction(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::SendReaction { - room: args.room, - event_id: args.event_id, - key: args.key, - }) + call( + args.account, + DaemonOp::SendReaction { + room: args.room, + event_id: args.event_id, + key: args.key, + }, + ) .await, ) } @@ -282,11 +371,14 @@ impl MatrixBridge { latest event first.")] async fn send_reply(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::SendReply { - room: args.room, - event_id: args.event_id, - body: args.body, - }) + call( + args.account, + DaemonOp::SendReply { + room: args.room, + event_id: args.event_id, + body: args.body, + }, + ) .await, ) } @@ -296,10 +388,13 @@ impl MatrixBridge { participants can see.")] async fn mark_read(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::MarkRead { - room: args.room, - event_id: args.event_id, - }) + call( + args.account, + DaemonOp::MarkRead { + room: args.room, + event_id: args.event_id, + }, + ) .await, ) } @@ -308,8 +403,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(_): Parameters) -> String { - render(round_trip(DaemonRequest::ListRooms).await) + async fn list_rooms(&self, Parameters(args): Parameters) -> String { + render(call(args.account, DaemonOp::ListRooms).await) } #[tool( @@ -317,8 +412,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(_): Parameters) -> String { - render(round_trip(DaemonRequest::ListInvites).await) + async fn list_invites(&self, Parameters(args): Parameters) -> String { + render(call(args.account, DaemonOp::ListInvites).await) } #[tool( @@ -329,7 +424,7 @@ impl MatrixBridge { `list_rooms`." )] async fn join_room(&self, Parameters(args): Parameters) -> String { - render(round_trip(DaemonRequest::JoinRoom { room: args.room }).await) + render(call(args.account, DaemonOp::JoinRoom { room: args.room }).await) } #[tool( @@ -349,10 +444,13 @@ impl MatrixBridge { } }; render( - round_trip(DaemonRequest::ResolveInvite { - room: args.room, - action, - }) + call( + args.account, + DaemonOp::ResolveInvite { + room: args.room, + action, + }, + ) .await, ) } @@ -365,10 +463,13 @@ impl MatrixBridge { )] async fn invite_user(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::InviteUser { - room: args.room, - user_id: args.user_id, - }) + call( + args.account, + DaemonOp::InviteUser { + room: args.room, + user_id: args.user_id, + }, + ) .await, ) } @@ -378,7 +479,7 @@ impl MatrixBridge { Each row carries the user id and resolved display name." )] async fn list_room_members(&self, Parameters(args): Parameters) -> String { - render(round_trip(DaemonRequest::ListRoomMembers { room: args.room }).await) + render(call(args.account, DaemonOp::ListRoomMembers { room: args.room }).await) } #[tool(description = "Read the most recent N events from a matrix room \ @@ -386,10 +487,13 @@ impl MatrixBridge { type, and best-effort plain-text body.")] async fn read_room(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::ReadRoom { - room: args.room, - limit: args.limit, - }) + call( + args.account, + DaemonOp::ReadRoom { + room: args.room, + limit: args.limit, + }, + ) .await, ) } @@ -403,11 +507,14 @@ impl MatrixBridge { )] async fn download_file(&self, Parameters(args): Parameters) -> String { render( - round_trip(DaemonRequest::DownloadFile { - room: args.room, - event_id: args.event_id, - dest_path: args.dest_path, - }) + call( + args.account, + DaemonOp::DownloadFile { + room: args.room, + event_id: args.event_id, + dest_path: args.dest_path, + }, + ) .await, ) } @@ -421,7 +528,9 @@ 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.")] + 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.")] impl ServerHandler for MatrixBridge {} #[tokio::main] diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index 32dd1ea0..12efc512 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -1,25 +1,31 @@ //! `hive-matrix-daemon` binary — long-running matrix-sdk Client + sync -//! loop per agent. Bridges incoming room events to hyperhive wake -//! signals and serves the unix socket the stdio MCP bridge talks to. +//! loop per matrix account. Bridges incoming room events to hyperhive +//! wake signals and serves the unix socket the stdio MCP bridge talks to. //! //! Lifecycle: -//! 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. +//! 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). //! -//! 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. +//! 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). //! //! Stale-token recovery: handled in `client::build_and_restore` — see //! that module for the `M_UNKNOWN_TOKEN` detection + cleanup flow. -use anyhow::{Context, Result}; -use matrix_sdk::config::SyncSettings; +use std::sync::Arc; +use anyhow::{Context, Result}; +use matrix_sdk::{Client, config::SyncSettings}; + +mod accounts; mod client; mod handlers; mod paths; @@ -28,6 +34,14 @@ 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() @@ -38,68 +52,130 @@ async fn main() -> Result<()> { .with_writer(std::io::stderr) .init(); - let homeserver = paths::homeserver_url(); - let token_file = paths::token_file(); - let state_dir = paths::matrix_state_dir(); + let cfgs = accounts::configured().context("read matrix account config")?; 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(); - 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)" - ); + 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"); return Ok(()); } - 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(); + // 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); let socket_listener = mcp_socket.clone(); tokio::spawn(async move { - if let Err(e) = socket::serve(&socket_listener, socket_client).await { + if let Err(e) = socket::serve(&socket_listener, registry).await { tracing::error!(error = %e, "mcp socket server exited"); } }); - // 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")?; + // 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"); + } + } 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 b847c3e4..d072c02b 100644 --- a/hive-matrix-mcp/src/protocol.rs +++ b/hive-matrix-mcp/src/protocol.rs @@ -9,12 +9,34 @@ use serde::{Deserialize, Serialize}; -/// Request from the stdio MCP bridge to the daemon. The MCP bridge +/// 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 /// 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 DaemonRequest { +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. @@ -204,3 +226,53 @@ 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 78f8acd3..03bf1c8d 100644 --- a/hive-matrix-mcp/src/socket.rs +++ b/hive-matrix-mcp/src/socket.rs @@ -11,13 +11,15 @@ 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::{DaemonRequest, DaemonResponse}; +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). -pub async fn serve(socket_path: &Path, client: Client) -> Result<()> { +/// 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<()> { let _ = tokio::fs::remove_file(socket_path).await; if let Some(parent) = socket_path.parent() { tokio::fs::create_dir_all(parent) @@ -27,27 +29,26 @@ pub async fn serve(socket_path: &Path, client: Client) -> 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 client = client.clone(); + let registry = registry.clone(); tokio::spawn(async move { - if let Err(e) = handle_connection(stream, &client).await { + if let Err(e) = handle_connection(stream, ®istry).await { tracing::warn!(error = %e, "mcp socket connection error"); } }); } } -async fn handle_connection(stream: UnixStream, client: &Client) -> Result<()> { +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::(&line) { - Ok(req) => dispatch(req, client).await, + Ok(req) => dispatch(req, registry).await, Err(e) => DaemonResponse::error(format!("parse request: {e}")), }; let mut json = serde_json::to_string(&response)?; @@ -58,49 +59,60 @@ async fn handle_connection(stream: UnixStream, client: &Client) -> Result<()> { Ok(()) } -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 { +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 { + 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 { room, path, caption, } => handlers::send_file(client, &room, &path, caption.as_deref()).await, - DaemonRequest::OpenDm { user_id } => handlers::open_dm(client, &user_id).await, - DaemonRequest::SendReaction { + 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, - DaemonRequest::SendReply { + DaemonOp::SendReply { room, event_id, body, } => handlers::send_reply(client, &room, &event_id, &body).await, - DaemonRequest::MarkRead { room, event_id } => { + DaemonOp::MarkRead { room, event_id } => { handlers::mark_read(client, &room, &event_id).await } - 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 } => { + 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 } - DaemonRequest::InviteUser { room, user_id } => { + DaemonOp::InviteUser { room, user_id } => { handlers::invite_user(client, &room, &user_id).await } - DaemonRequest::ListRoomMembers { room } => handlers::list_room_members(client, &room).await, - DaemonRequest::ReadRoom { room, limit } => handlers::read_room(client, &room, limit).await, - DaemonRequest::DownloadFile { + DaemonOp::ListRoomMembers { room } => handlers::list_room_members(client, &room).await, + DaemonOp::ReadRoom { room, limit } => handlers::read_room(client, &room, limit).await, + DaemonOp::DownloadFile { room, event_id, dest_path, } => handlers::download_file(client, &room, &event_id, dest_path.as_deref()).await, - DaemonRequest::UnreadCount => handlers::unread_count(client), - DaemonRequest::UnreadSummary => handlers::unread_summary(client).await, + DaemonOp::UnreadCount => handlers::unread_count(client), + DaemonOp::UnreadSummary => handlers::unread_summary(client).await, } } diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 41553827..a60fe479 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/hive-matrix-mcp/src/timeline.rs @@ -27,13 +27,24 @@ 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. -pub fn install_message_handler(client: &Client, hyperhive_socket: PathBuf) { +/// +/// `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, +) { 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 @@ -82,6 +93,7 @@ pub fn install_message_handler(client: &Client, hyperhive_socket: PathBuf) { } 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 { @@ -106,7 +118,12 @@ pub fn install_message_handler(client: &Client, hyperhive_socket: PathBuf) { /// 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>) { +pub async fn sweep_invites( + client: &Client, + socket: &Path, + notified: &Mutex>, + account_tag: Option<&str>, +) { let current = client.invited_rooms(); let current_ids: HashSet = current.iter().map(|r| r.room_id().to_owned()).collect(); @@ -134,9 +151,12 @@ pub async fn sweep_invites(client: &Client, socket: &Path, notified: &Mutex 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. @@ -177,4 +189,19 @@ 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" + ); + } }