//! `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. //! //! 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). //! //! 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 (`M_UNKNOWN_TOKEN`): handled in //! `client::build_and_restore`, and it mirrors the missing-token policy //! above — a rejected PRIMARY token drops the stale token + sdk state and //! exits 0 for systemd re-provisioning, while a rejected SECONDARY token //! is removed and that one account is skipped so the daemon keeps serving //! the primary and any other healthy account. use std::sync::Arc; use anyhow::{Context, Result}; use matrix_sdk::{Client, config::SyncSettings}; mod accounts; mod client; mod handlers; mod paths; mod protocol; 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>>>; /// How often the daemon rewrites the accounts snapshot to advance its /// mtime, so the dashboard's `as_of` tracks daemon liveness. 30s keeps /// the staleness window small while the rewrite cost (one tmp + rename of /// a tiny file) is negligible. const ACCOUNTS_HEARTBEAT_SECS: u64 = 30; #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), ) .with_writer(std::io::stderr) .init(); 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(); 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, is_primary).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(()); } // Publish the live-account snapshot (BE-4) for the dashboard: the // accounts that restored, with effective homeserver + user id. Lands // in the host-visible state dir so hive-c0re reads it to show live // up/down + backfill homeserver. Best-effort — a failed write must // not stop the daemon from serving. if let Err(e) = registry.write_snapshot(&paths::accounts_file()) { tracing::warn!(error = %format!("{e:#}"), "failed to write matrix-accounts snapshot"); } // Serve the socket against the registry. Spawned before driving the // sync loops so the MCP bridge can connect as soon as the first // claude turn fires. let registry = Arc::new(registry); // Heartbeat: periodically rewrite the accounts snapshot so its mtime // advances while the daemon lives. The dashboard derives `as_of` from // the file mtime, so a stalled mtime now means the daemon is down — // which lets the dashboard dim accounts whose snapshot is stale rather // than reporting the boot-time set forever (BE-4 follow-up). let hb_registry = Arc::clone(®istry); tokio::spawn(async move { let mut tick = tokio::time::interval(std::time::Duration::from_secs(ACCOUNTS_HEARTBEAT_SECS)); tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let path = paths::accounts_file(); loop { tick.tick().await; if let Err(e) = hb_registry.heartbeat_snapshot(&path) { tracing::warn!(error = %format!("{e:#}"), "matrix-accounts heartbeat write failed"); } } }); let socket_listener = mcp_socket.clone(); tokio::spawn(async move { if let Err(e) = socket::serve(&socket_listener, registry).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"); } } 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, is_primary: bool, ) -> 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, is_primary) .await .with_context(|| format!("build matrix client for account {}", cfg.name))?; // Best-effort: sync the agent icon to this account's matrix avatar over // the live (authenticated, correct-homeserver) Client. Replaces the old // curl oneshot; failures are swallowed inside sync_avatar. client::sync_avatar(&client, &cfg.state_dir, &cfg.name).await; 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))) }