feat: multi-account matrix daemon (account-routed mcp surface)

This commit is contained in:
damocles 2026-06-15 20:23:42 +02:00
commit 8e79eb4f26
7 changed files with 613 additions and 156 deletions

View file

@ -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<String>,
}
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<Vec<AccountCfg>> {
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<AccountCfg> = 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<String, Arc<Client>>,
}
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<Client>, 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(", ")
)
})
}
}

View file

@ -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<DaemonResponse> {
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
}
/// 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<String>,
}
#[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<String>,
}
#[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<String>,
/// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`).
/// Omit to use the agent's primary account.
#[serde(default)]
account: Option<String>,
}
#[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<String>,
}
#[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<String>,
}
#[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<String>,
}
#[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<String>,
}
#[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<String>,
}
#[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<String>,
}
#[derive(Debug, Deserialize, JsonSchema)]
@ -136,6 +185,10 @@ struct ReadRoomArgs {
/// Maximum events to return (default 50, max 200). Newest first.
#[serde(default)]
limit: Option<usize>,
/// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`).
/// Omit to use the agent's primary account.
#[serde(default)]
account: Option<String>,
}
#[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<String>,
/// Matrix account to act as (a `name` from `hyperhive.matrixAccounts`).
/// Omit to use the agent's primary account.
#[serde(default)]
account: Option<String>,
}
#[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<String>,
}
#[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<String>,
}
#[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<String>,
}
#[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<String>,
}
struct MatrixBridge {
@ -207,10 +281,13 @@ impl MatrixBridge {
)]
async fn send_message(&self, Parameters(args): Parameters<SendMessageArgs>) -> 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<SendDmArgs>) -> 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<SendFileArgs>) -> 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<OpenDmArgs>) -> 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<SendReactionArgs>) -> 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<SendReplyArgs>) -> 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<MarkReadArgs>) -> 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<ListRoomsArgs>) -> String {
render(round_trip(DaemonRequest::ListRooms).await)
async fn list_rooms(&self, Parameters(args): Parameters<ListRoomsArgs>) -> 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<ListInvitesArgs>) -> String {
render(round_trip(DaemonRequest::ListInvites).await)
async fn list_invites(&self, Parameters(args): Parameters<ListInvitesArgs>) -> 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<JoinRoomArgs>) -> 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<InviteUserArgs>) -> 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<ListRoomMembersArgs>) -> 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<ReadRoomArgs>) -> 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<DownloadFileArgs>) -> 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]

View file

@ -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<Box<dyn std::future::Future<Output = Result<()>>>>;
#[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<SyncLoop> = 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, &notified).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<String>,
) -> Result<Option<(Client, SyncLoop)>> {
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, &notified, tag.as_deref()).await;
matrix_sdk::LoopCtrl::Continue
}
})
.await
.context("matrix-sdk sync loop exited")?;
Ok(())
});
Ok(Some((client, sync_loop)))
}

View file

@ -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<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 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);
}
}

View file

@ -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<Registry>) -> 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, &registry).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::<DaemonRequest>(&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,
}
}

View file

@ -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<String>,
) {
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<HashSet<OwnedRoomId>>) {
pub async fn sweep_invites(
client: &Client,
socket: &Path,
notified: &Mutex<HashSet<OwnedRoomId>>,
account_tag: Option<&str>,
) {
let current = client.invited_rooms();
let current_ids: HashSet<OwnedRoomId> =
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<Hash
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 = format!(
"[matrix] invited to {label} ({room_id}) — \
use list_invites to see pending invites, resolve_invite to accept or reject"
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"
),
);
if let Err(e) = wake::send_wake(socket, &body).await {
tracing::warn!(error = %e, "matrix: failed to deliver invite-wake to hyperhive");

View file

@ -119,6 +119,18 @@ 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:<name>] <body>`.
#[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"
);
}
}