hyperhive/hive-matrix-mcp/src/main.rs

//! `hive-matrix-daemon` binary — long-running matrix-sdk Client + sync
//! loop per matrix account. Bridges incoming room events to hyperhive
//! wake signals and serves its MCP tools directly over streamable-http
//! on `--http <addr>` — no stdio bridge, no separate bin claude has to
//! respawn every turn.
//!
//! Lifecycle:
//! 1. Read the account list: `accounts::configured()` (declared) plus
//!    `accounts::discover_linked` (linked in the swarm secret store).
//! 2. For each account: resolve its access token (`account_token` — the
//!    swarm secret store first, read by this agent as itself, then a declared
//!    token file) → 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 MCP tools against an account→Client registry; each tool call
//!    routes to its `account` arg (the primary account when omitted).
//! 4. Every [`LINKED_REFRESH`], re-read the linked accounts; exit with
//!    [`ACCOUNTS_CHANGED_EXIT`] when the set changed, so systemd restarts us.
//! 5. On the same period, try again each secondary that did not come up
//!    ([`retry_down`], backoff in [`retry_delay`]). One that comes up joins
//!    the registry; the accounts already up are not restarted.
//!
//! The PRIMARY account having no token, or one the homeserver rejects
//! (`M_UNKNOWN_TOKEN`, see `client::build_and_restore`, which also drops its
//! sdk state) → exit 0; systemd restarts us once a token exists (the
//! path-watcher for a token file, the store re-check timer for a stored
//! token). A SECONDARY account with no token, a rejected one, or a down
//! homeserver is left down and retried (step 5) while the others serve.

use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::convert::Infallible;
use std::path::Path;
use std::process::ExitCode;
use std::sync::Arc;

use anyhow::{Context, Result};
use clap::Parser;
use futures_util::stream::{FuturesUnordered, StreamExt};
use matrix_sdk::{Client, config::SyncSettings};
use tokio::time::Instant;

use hive_matrix_mcp::accounts::{AccountCfg, Registry};
use hive_matrix_mcp::client::PermanentBringUpError;
use hive_matrix_mcp::credential::Token;
use hive_matrix_mcp::{accounts, client, credential, mcp, paths, timeline, wake};

#[derive(Parser)]
#[command(name = "hive-matrix-daemon", about = "matrix-sdk client + MCP daemon")]
struct Cli {
    /// Serve the MCP tools over streamable-http on this address (e.g.
    /// `127.0.0.1:8792`). Bind loopback only.
    #[arg(long)]
    http: std::net::SocketAddr,
}

/// A per-account sync loop, boxed so loops for N accounts can be driven
/// concurrently on the main task. Deliberately NOT `Send`: matrix-sdk's
/// 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<()>>>>;

/// A [`SyncLoop`] that yields its account's name when it ends.
type NamedSyncLoop = std::pin::Pin<Box<dyn std::future::Future<Output = (String, Result<()>)>>>;

/// What [`retry_down`] needs to bring a secondary up again: its config and
/// its todo tag.
type Retry = (AccountCfg, Option<String>);

/// 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;

/// Retry delays (seconds) for transient secondary-account failures at
/// start. After the initial attempt we back off through these before
/// leaving the account to [`retry_down`].
/// Transient = anything that is NOT a `PermanentBringUpError` (bad/expired
/// token). A down homeserver at DNS-not-ready boot time is the typical
/// case; the total wait is ~52s before we give up.
const SECONDARY_RETRY_DELAYS_SECS: &[u64] = &[2, 5, 15, 30];

/// How often the daemon re-reads the store's linked accounts. A newly linked
/// account comes up within this long of the swarm UI storing it.
const LINKED_REFRESH: std::time::Duration = std::time::Duration::from_mins(2);

/// The longest wait between two tries of a secondary account that is down.
/// A rejected token is tried again at this pace until an operator links a
/// new one, so it bounds both the homeserver's load and how long a re-link
/// takes to come up.
const MAX_SECONDARY_RETRY: std::time::Duration = std::time::Duration::from_mins(16);

/// Why a secondary is down when it had no token to try.
const NO_TOKEN: &str = "no token";

/// Why a secondary is down when start-up gave up on it; the reason was logged
/// there.
const FAILED_AT_START: &str = "failed at start";

/// The exit status the daemon leaves with when the store's linked accounts
/// changed. `nix/agent-modules/matrix.nix` names the same number in the unit's
/// `RestartForceExitStatus` and `SuccessExitStatus`, so systemd restarts it
/// without counting a failure.
const ACCOUNTS_CHANGED_EXIT: u8 = 75;

#[tokio::main]
async fn main() -> Result<ExitCode> {
    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)
        // This is a systemd-managed daemon — stderr always goes to journald,
        // never a human terminal, and journald doesn't strip ANSI escapes:
        // they land in victorialogs as raw byte-array spam otherwise.
        .with_ansi(false)
        .init();

    let cli = Cli::parse();
    let todo_socket = wake::agent_socket();
    let declared = accounts::configured().context("read matrix account config")?;
    accounts::remove_delivered_files(&declared);
    let mut warned = HashSet::new();
    let linked = linked_now(&declared, &mut warned).await.unwrap_or_default();
    let booted = account_names(&declared, &linked);
    let mut cfgs = declared.clone();
    cfgs.extend(linked);
    let multi = cfgs.len() > 1;
    let primary = cfgs[0].name.clone();
    let registry = Registry::new(primary);
    let mut sync_loops: Vec<NamedSyncLoop> = Vec::new();
    let mut down = DownAccounts::default();

    for (idx, cfg) in cfgs.into_iter().enumerate() {
        let is_primary = idx == 0;
        // Account-tag the todos only in multi-account mode so single-
        // account todo summaries stay byte-identical to the legacy format.
        let tag = multi.then(|| cfg.name.clone());
        match bring_up_account(&cfg, tag.clone(), is_primary).await {
            Ok(Some((client, sync_loop))) => {
                sync_loops.push(named(cfg.name.clone(), sync_loop));
                registry.insert(cfg.name, client);
            }
            // No token yet: for the primary that means the daemon isn't
            // useful — exit 0, and systemd restarts us once a token exists
            // (the path-watcher for a file, the re-check timer for the store).
            Ok(None) if is_primary => {
                tracing::warn!(
                    account = %cfg.name,
                    "primary matrix account has no token yet; exiting cleanly \
                     (systemd restarts us once one exists)"
                );
                return Ok(ExitCode::SUCCESS);
            }
            Ok(None) => {
                tracing::warn!(account = %cfg.name, "secondary matrix account has no token; will retry");
                down.failed(cfg.name.clone(), (cfg, tag), Instant::now(), NO_TOKEN);
            }
            // The primary failing to restore is fatal (propagate so
            // systemd retries on a transient blip — matches legacy
            // behaviour); a secondary failing is retried with backoff
            // here, then left to `retry_down`.
            Err(e) if is_primary => return Err(e.context("bring up primary matrix account")),
            Err(e) => match bring_up_secondary_with_retry(&cfg, tag.clone(), e).await {
                Some((client, sync_loop)) => {
                    sync_loops.push(named(cfg.name.clone(), sync_loop));
                    registry.insert(cfg.name, client);
                }
                None => {
                    down.failed(
                        cfg.name.clone(),
                        (cfg, tag),
                        Instant::now(),
                        FAILED_AT_START,
                    );
                }
            },
        }
    }

    if registry.is_empty() {
        tracing::warn!("no matrix accounts restored; exiting cleanly");
        return Ok(ExitCode::SUCCESS);
    }

    // Startup cancel-and-recreate (loose-ends v2): wipe this agent's
    // matrix todos so stale ones (rooms read / invites resolved while the
    // daemon was down) don't linger, then let the first sweep rebuild the
    // set to match current reality. Best-effort; the sweep converges.
    // Once, before any sync loop runs: a secondary that comes up later must
    // not wipe the todos the running accounts' sweeps already deduped, so
    // `todo_socket` is passed to nothing that brings an account up.
    if let Some(socket) = &todo_socket {
        let _ = wake::clear_all_todos(socket).await;
    }

    // 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 MCP tools against the registry. Spawned before driving
    // the sync loops so claude can reach the stable http URL as soon as
    // the first turn fires.
    let registry = Arc::new(registry);

    // Heartbeat: periodically rewrite the accounts snapshot so its mtime
    // 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(&registry);
    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 http_addr = cli.http;
    let mcp_registry = Arc::clone(&registry);
    tokio::spawn(async move {
        if let Err(e) = mcp::serve_http(http_addr, mcp_registry).await {
            tracing::error!(error = %e, "mcp http server exited");
        }
    });

    let (up_tx, up_rx) = tokio::sync::mpsc::unbounded_channel();
    let retry = Box::pin(retry_down(down, Arc::clone(&registry), up_tx));
    let changed = wait_for_linked_change(declared, booted, warned);
    Ok(drive(sync_loops, up_rx, changed, retry).await)
}

/// Drive all per-account sync loops concurrently on this task (they aren't
/// `Send`, so no `tokio::spawn`) until one returns or `changed` resolves,
/// adding each loop `up` delivers as `retry` brings a secondary up.
/// matrix-sdk reconnects internally, so any loop returning is exceptional —
/// log it and exit so systemd restarts the whole daemon cleanly.
async fn drive(
    sync_loops: Vec<NamedSyncLoop>,
    mut up: tokio::sync::mpsc::UnboundedReceiver<NamedSyncLoop>,
    changed: impl std::future::Future<Output = ()>,
    retry: impl std::future::Future<Output = Infallible>,
) -> ExitCode {
    let mut loops: FuturesUnordered<NamedSyncLoop> = sync_loops.into_iter().collect();
    tokio::pin!(changed, retry);
    loop {
        tokio::select! {
            Some((account, result)) = loops.next() => {
                match result {
                    Ok(()) => tracing::warn!(
                        %account,
                        "a matrix sync loop exited cleanly; restarting daemon"
                    ),
                    Err(e) => {
                        tracing::error!(%account, error = %format!("{e:#}"), "a matrix sync loop errored; restarting daemon");
                    }
                }
                return ExitCode::SUCCESS;
            }
            () = &mut changed => {
                tracing::info!("the store's linked matrix accounts changed; exiting to be restarted onto them");
                return ExitCode::from(ACCOUNTS_CHANGED_EXIT);
            }
            never = &mut retry => match never {},
            Some(sync_loop) = up.recv() => loops.push(sync_loop),
        }
    }
}

/// `sync_loop`, yielding `account` beside its result.
fn named(account: String, sync_loop: SyncLoop) -> NamedSyncLoop {
    Box::pin(async move { (account, sync_loop.await) })
}

/// The accounts the store links to this agent beyond `declared`, or `None`
/// when the store could not be read.
///
/// Each linked name that cannot be brought up is logged the first time it is
/// seen and recorded in `warned`, so a refresh does not repeat it.
async fn linked_now(
    declared: &[AccountCfg],
    warned: &mut HashSet<String>,
) -> Option<Vec<AccountCfg>> {
    let found = match accounts::discover_linked(declared).await {
        Ok(found) => found,
        Err(e) => {
            tracing::warn!(
                error = %format!("{e:#}"),
                "could not read this agent's linked matrix accounts from the store"
            );
            return None;
        }
    };
    for name in found.missing {
        if warned.insert(name.clone()) {
            tracing::warn!(
                account = %name,
                "the store lists this matrix account but holds no credential for it; skipping"
            );
        }
    }
    for name in found.no_homeserver {
        if warned.insert(name.clone()) {
            tracing::warn!(
                account = %name,
                "this linked matrix account was stored without a homeserver; skipping \
                 (re-link it from the swarm UI with one)"
            );
        }
    }
    Some(found.accounts)
}

/// Every account name the daemon would bring up from `declared` + `linked`.
fn account_names(declared: &[AccountCfg], linked: &[AccountCfg]) -> BTreeSet<String> {
    declared
        .iter()
        .chain(linked)
        .map(|a| a.name.clone())
        .collect()
}

/// Re-read the store's linked accounts every [`LINKED_REFRESH`] and resolve
/// once they name a different set than `booted`. A failed read is logged and
/// skipped rather than counted as a change.
async fn wait_for_linked_change(
    declared: Vec<AccountCfg>,
    booted: BTreeSet<String>,
    mut warned: HashSet<String>,
) {
    let mut tick = tokio::time::interval(LINKED_REFRESH);
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    // The first tick is immediate, and boot has just read the store.
    tick.tick().await;
    loop {
        tick.tick().await;
        let Some(linked) = linked_now(&declared, &mut warned).await else {
            continue;
        };
        if account_names(&declared, &linked) != booted {
            return;
        }
    }
}

/// Try to bring up a secondary account, retrying with exponential backoff
/// on transient failures (anything that is NOT a `PermanentBringUpError`).
/// Returns `Some((client, sync_loop))` on success, or `None` to signal
/// that the account stays down and is left to [`retry_down`] (permanent
/// failure, no token, or all retries exhausted).
async fn bring_up_secondary_with_retry(
    cfg: &AccountCfg,
    tag: Option<String>,
    first_error: anyhow::Error,
) -> Option<(Client, SyncLoop)> {
    // Permanent failure: the homeserver rejected the token. Trying the same
    // token again at once won't help — leave it to `retry_down`.
    if first_error
        .downcast_ref::<PermanentBringUpError>()
        .is_some()
    {
        tracing::error!(
            account = %cfg.name,
            error = %format!("{first_error:#}"),
            "secondary matrix account: permanent failure; leaving it down"
        );
        return None;
    }
    // Transient failure (network/DNS/homeserver 5xx): retry with backoff.
    tracing::warn!(
        account = %cfg.name,
        error = %format!("{first_error:#}"),
        "secondary matrix account bring-up failed (transient); will retry"
    );
    for &delay in SECONDARY_RETRY_DELAYS_SECS {
        tracing::info!(
            account = %cfg.name,
            delay_s = delay,
            "retrying secondary account bring-up after backoff"
        );
        tokio::time::sleep(std::time::Duration::from_secs(delay)).await;
        match bring_up_account(cfg, tag.clone(), false).await {
            Ok(Some((client, sync_loop))) => {
                tracing::info!(account = %cfg.name, "secondary matrix account recovered");
                return Some((client, sync_loop));
            }
            Ok(None) => {
                tracing::warn!(account = %cfg.name, "secondary matrix account has no token; leaving it down");
                return None;
            }
            Err(re) if re.downcast_ref::<PermanentBringUpError>().is_some() => {
                tracing::error!(
                    account = %cfg.name,
                    error = %format!("{re:#}"),
                    "secondary matrix account: permanent failure on retry; leaving it down"
                );
                return None;
            }
            Err(re) => {
                tracing::warn!(
                    account = %cfg.name,
                    error = %format!("{re:#}"),
                    "secondary matrix account bring-up still failing (transient)"
                );
            }
        }
    }
    tracing::error!(
        account = %cfg.name,
        retries = SECONDARY_RETRY_DELAYS_SECS.len(),
        "secondary matrix account failed after all retries; leaving it down"
    );
    None
}

/// The wait before the next try of a secondary that has failed `failures`
/// times in a row: one [`LINKED_REFRESH`], doubled per further failure, up to
/// [`MAX_SECONDARY_RETRY`].
fn retry_delay(failures: u32) -> std::time::Duration {
    let doublings = failures.saturating_sub(1).min(16);
    LINKED_REFRESH
        .saturating_mul(1 << doublings)
        .min(MAX_SECONDARY_RETRY)
}

/// Secondary accounts that did not come up, each with what it takes to try it
/// again (`T`) and the earliest time to.
struct DownAccounts<T> {
    by_name: BTreeMap<String, Down<T>>,
}

struct Down<T> {
    retry: T,
    failures: u32,
    not_before: Instant,
    /// Why the last try failed.
    reason: String,
}

impl<T> Default for DownAccounts<T> {
    fn default() -> Self {
        Self {
            by_name: BTreeMap::new(),
        }
    }
}

impl<T> DownAccounts<T> {
    /// Record that `name` failed to come up at `at`, for `reason`. `retry` is
    /// kept from the first failure on. True when the failure is news — the
    /// account's first, or for a different reason than the last — and so
    /// worth a warning.
    fn failed(&mut self, name: String, retry: T, at: Instant, reason: &str) -> bool {
        let down = self.by_name.entry(name).or_insert(Down {
            retry,
            failures: 0,
            not_before: at,
            reason: String::new(),
        });
        let news = down.failures == 0 || down.reason != reason;
        reason.clone_into(&mut down.reason);
        down.failures += 1;
        down.not_before = at + retry_delay(down.failures);
        news
    }

    /// Record that `name` came up; it is never due again.
    fn came_up(&mut self, name: &str) {
        self.by_name.remove(name);
    }

    /// The account to try at `now`: of those whose wait is over, the one
    /// that has waited longest past it.
    fn due(&self, now: Instant) -> Option<(&str, &T)> {
        self.by_name
            .iter()
            .filter(|(_, d)| d.not_before <= now)
            .min_by_key(|(_, d)| d.not_before)
            .map(|(name, d)| (name.as_str(), &d.retry))
    }
}

/// What a restore attempt yields; [`restore_account`]'s result.
type Restore = Result<Option<Client>>;

/// Every [`LINKED_REFRESH`], try again the secondary [`DownAccounts::due`]
/// names, one per tick, with [`restore_account`]. Never returns.
async fn retry_down(
    mut down: DownAccounts<Retry>,
    registry: Arc<Registry>,
    up: tokio::sync::mpsc::UnboundedSender<NamedSyncLoop>,
) -> Infallible {
    retry_loop(
        &mut down,
        &registry,
        &up,
        &paths::accounts_file(),
        |cfg: AccountCfg| async move { restore_account(&cfg, false).await },
    )
    .await
}

/// [`retry_down`] with the restore step and the snapshot path injected.
async fn retry_loop<F, Fut>(
    down: &mut DownAccounts<Retry>,
    registry: &Registry,
    up: &tokio::sync::mpsc::UnboundedSender<NamedSyncLoop>,
    snapshot: &Path,
    mut restore: F,
) -> Infallible
where
    F: FnMut(AccountCfg) -> Fut,
    Fut: std::future::Future<Output = Restore>,
{
    let mut tick = tokio::time::interval(LINKED_REFRESH);
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    // The first tick is immediate, and boot has just tried every account.
    tick.tick().await;
    loop {
        let now = tick.tick().await;
        retry_once(down, registry, up, snapshot, now, &mut restore).await;
    }
}

/// Try the secondary of `down` that is due at `now`, if any. One that comes up
/// leaves `down`, joins `registry` and has its [`account_sync_loop`] sent to
/// `up`; the accounts already in `registry` are not touched. One that fails
/// stays in `down`, due again after a longer wait. No lock on `registry` is
/// held while `restore` runs.
async fn retry_once<F, Fut>(
    down: &mut DownAccounts<Retry>,
    registry: &Registry,
    up: &tokio::sync::mpsc::UnboundedSender<NamedSyncLoop>,
    snapshot: &Path,
    now: Instant,
    restore: &mut F,
) where
    F: FnMut(AccountCfg) -> Fut,
    Fut: std::future::Future<Output = Restore>,
{
    let Some((name, (cfg, tag))) = down.due(now) else {
        return;
    };
    let (name, cfg, tag) = (name.to_owned(), cfg.clone(), tag.clone());
    match restore(cfg.clone()).await {
        Ok(Some(client)) => {
            tracing::info!(account = %name, "secondary matrix account came up");
            let sync_loop = account_sync_loop(&client, tag);
            down.came_up(&name);
            registry.insert(name.clone(), client);
            if let Err(e) = registry.write_snapshot(snapshot) {
                tracing::warn!(error = %format!("{e:#}"), "failed to write matrix-accounts snapshot");
            }
            // `drive` holds the receiver for as long as it polls this.
            let _ = up.send(named(name, sync_loop));
        }
        Ok(None) => {
            if down.failed(name.clone(), (cfg, tag), now, NO_TOKEN) {
                tracing::warn!(account = %name, "secondary matrix account has no token; will keep retrying");
            } else {
                tracing::debug!(account = %name, "secondary matrix account still has no token");
            }
        }
        Err(e) => {
            let reason = format!("{e:#}");
            if down.failed(name.clone(), (cfg, tag), now, &reason) {
                tracing::warn!(account = %name, error = %reason, "secondary matrix account failing; will keep retrying");
            } else {
                tracing::debug!(account = %name, error = %reason, "secondary matrix account still failing");
            }
        }
    }
}

/// Resolve `cfg`'s access token, preferring the copy this agent can fetch
/// itself.
///
/// 🏛️ The store read is the point: the credential at
/// `swarm/agents/<agent>/matrix/<account>` is **this agent's**, and it is read
/// from inside this agent's container under this agent's own certificate,
/// never by the hive on the agent's behalf. It is held in memory from here to
/// `restore_session` and is written nowhere.
///
/// `main` included: `swarm-controller` mints it with the swarm's appservice
/// token and stores it at `swarm/agents/<agent>/matrix/main`, and no hive
/// mints it any more. An agent whose harness forwards no
/// [`credential::ENV_AGENT`] cannot name its own subtree, so it reads the file.
///
/// The file is what remains when the store answers nothing: a `tokenFile` an
/// operator declared, or a `main` token a hive minted before the swarm did. An
/// account the store links has no file. A store read that *fails* falls back
/// too, and says why.
///
/// # Errors
/// As [`credential::Token::from_file`]: a token file that exists but cannot be
/// read, or is empty.
async fn account_token(cfg: &AccountCfg) -> Result<Option<Token>> {
    if let Some(agent) = credential::agent_name() {
        match Token::from_store(&agent, &cfg.name).await {
            Ok(Some(token)) => return Ok(Some(token)),
            // No store in this deployment: the declared file is the whole
            // mechanism here.
            Ok(None) => {}
            Err(e) => tracing::warn!(
                account = %cfg.name,
                error = %format!("{e:#}"),
                "could not read this account's credential from the store as this agent; \
                 falling back to its declared token file"
            ),
        }
    }
    match &cfg.token_file {
        Some(path) => Token::from_file(path).await,
        None => Ok(None),
    }
}

/// Read `account`'s recovery passphrase from the store and hand it to
/// [`client::ensure_recovery`]. With none — no store, or `swarm-controller`
/// has not minted it yet — key backup stays off until the next start.
async fn ensure_recovery(client: &Client, account: &str) {
    let found = match credential::agent_name() {
        Some(agent) => credential::Passphrase::from_store(&agent, account).await,
        None => Ok(None),
    };
    match found {
        Ok(Some(passphrase)) => client::ensure_recovery(client, account, &passphrase).await,
        Ok(None) => tracing::info!(
            account,
            "no matrix recovery passphrase in the store; key backup stays off"
        ),
        Err(e) => tracing::warn!(
            account,
            error = %format!("{e:#}"),
            "could not read the matrix recovery passphrase; key backup stays off"
        ),
    }
}

/// Restore one account's client (when its token exists) and build its sync
/// loop. Returns `Ok(Some((client, sync_loop)))` when the account came up,
/// `Ok(None)` when it has no token or homeserver (caller decides
/// primary-vs-secondary handling). The returned sync loop is driven by the
/// caller.
async fn bring_up_account(
    cfg: &AccountCfg,
    tag: Option<String>,
    is_primary: bool,
) -> Result<Option<(Client, SyncLoop)>> {
    Ok(restore_account(cfg, is_primary).await?.map(|client| {
        let sync_loop = account_sync_loop(&client, tag);
        (client, sync_loop)
    }))
}

/// Restore one account's client, when it has a homeserver and a token:
/// everything in bringing an account up that talks to the network.
async fn restore_account(cfg: &AccountCfg, is_primary: bool) -> Result<Option<Client>> {
    let Some(homeserver) = cfg.homeserver() else {
        // No homeserver for this account: the hive has none to offer (no
        // matrix vhost) or this agent's `services.hyperhive.agent.matrix.url` is null. Same
        // no-op as a missing token — an absent integration, not a guess at
        // one.
        tracing::info!(
            account = %cfg.name,
            "no homeserver configured (HIVE_MATRIX_URL unset); skipping account"
        );
        return Ok(None);
    };
    let Some(token) = account_token(cfg).await? else {
        return Ok(None);
    };
    tracing::info!(
        account = %cfg.name,
        homeserver,
        credential = %token.origin(),
        state_dir = %cfg.state_dir.display(),
        "bringing up matrix account"
    );
    let client = client::build_and_restore(&homeserver, &token, &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;
    // Best-effort: bootstrap cross-signing so the account's device isn't
    // flagged "unverified" in other users' clients. Idempotent + swallows
    // failures (a UIAA-requiring homeserver) inside ensure_cross_signing.
    client::ensure_cross_signing(&client, &cfg.name).await;
    // Best-effort, and after cross-signing: see `client::ensure_recovery`.
    ensure_recovery(&client, &cfg.name).await;
    Ok(Some(client))
}

/// `client`'s sync loop, sweeping invites and unread rooms into todos tagged
/// `tag` after each sync. Builds the loop only; nothing runs until it is
/// polled. Never wipes the todo set: that is `main`'s, once, at startup.
fn account_sync_loop(client: &Client, tag: Option<String>) -> SyncLoop {
    let sync_client = client.clone();
    let cb_client = client.clone();
    // Separate dedup sets: one tracks invites already pushed as todos, the
    // other unread-message rooms. Both are pruned to their current state
    // each sweep (see the sweep fns) so re-invites / new messages re-push.
    let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new()));
    let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new()));
    Box::pin(async move {
        sync_client
            .sync_with_callback(SyncSettings::default(), move |_response| {
                let client = cb_client.clone();
                let invite_notified = invite_notified.clone();
                let unread_notified = unread_notified.clone();
                let tag = tag.clone();
                async move {
                    timeline::sweep_invites(&client, &invite_notified, tag.as_deref()).await;
                    timeline::sweep_unread(&client, &unread_notified, tag.as_deref()).await;
                    matrix_sdk::LoopCtrl::Continue
                }
            })
            .await
            .context("matrix-sdk sync loop exited")?;
        Ok(())
    })
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use tokio::time::Instant;

    use std::path::Path;
    use std::sync::Arc;

    use anyhow::anyhow;
    use matrix_sdk::Client;
    use tokio::sync::mpsc::{UnboundedReceiver, unbounded_channel};

    use super::{
        AccountCfg, DownAccounts, LINKED_REFRESH, MAX_SECONDARY_RETRY, NamedSyncLoop, Registry,
        Restore, Retry, retry_delay, retry_loop, retry_once, wake,
    };

    #[test]
    fn the_wait_starts_at_one_refresh_and_doubles_up_to_the_cap() {
        assert_eq!(retry_delay(1), LINKED_REFRESH);
        assert_eq!(retry_delay(2), LINKED_REFRESH * 2);
        assert_eq!(retry_delay(3), LINKED_REFRESH * 4);
        assert_eq!(retry_delay(5), MAX_SECONDARY_RETRY);
        assert_eq!(retry_delay(u32::MAX), MAX_SECONDARY_RETRY);
    }

    #[test]
    fn a_failed_account_is_due_once_its_wait_is_over_and_not_before() {
        let t0 = Instant::now();
        let mut down = DownAccounts::default();
        down.failed("ccc".to_owned(), 7, t0, "r");

        assert_eq!(down.due(t0), None);
        assert_eq!(down.due(t0 + LINKED_REFRESH - Duration::from_secs(1)), None);
        assert_eq!(down.due(t0 + LINKED_REFRESH), Some(("ccc", &7)));
    }

    #[test]
    fn each_failure_in_a_row_pushes_the_next_try_further_out() {
        let t0 = Instant::now();
        let mut down = DownAccounts::default();
        down.failed("ccc".to_owned(), (), t0, "r");
        let t1 = t0 + LINKED_REFRESH;
        down.failed("ccc".to_owned(), (), t1, "r");

        assert_eq!(down.due(t1 + LINKED_REFRESH), None);
        assert_eq!(down.due(t1 + LINKED_REFRESH * 2), Some(("ccc", &())));
    }

    #[test]
    fn an_account_that_came_up_is_never_due_again() {
        let t0 = Instant::now();
        let mut down = DownAccounts::default();
        down.failed("ccc".to_owned(), (), t0, "r");
        down.came_up("ccc");

        assert_eq!(down.due(t0 + MAX_SECONDARY_RETRY * 10), None);
    }

    #[test]
    fn the_longest_overdue_account_goes_first_and_the_others_keep_their_schedule() {
        let t0 = Instant::now();
        let mut down = DownAccounts::default();
        down.failed(
            "late".to_owned(),
            "late cfg",
            t0 + Duration::from_secs(30),
            "r",
        );
        down.failed("early".to_owned(), "early cfg", t0, "r");
        let now = t0 + LINKED_REFRESH * 2;

        assert_eq!(down.due(now), Some(("early", &"early cfg")));
        down.failed("early".to_owned(), "early cfg", now, "r");
        assert_eq!(down.due(now), Some(("late", &"late cfg")));
    }

    #[test]
    fn a_failure_is_news_the_first_time_and_when_its_reason_changes() {
        let t0 = Instant::now();
        let mut down = DownAccounts::default();
        assert!(down.failed("ccc".to_owned(), (), t0, "rejected"));
        assert!(!down.failed("ccc".to_owned(), (), t0, "rejected"));
        assert!(down.failed("ccc".to_owned(), (), t0, "no token"));
        assert!(!down.failed("ccc".to_owned(), (), t0, "no token"));
        assert!(down.failed("other".to_owned(), (), t0, "no token"));
    }

    #[test]
    fn the_first_retry_target_is_kept_across_failures() {
        let t0 = Instant::now();
        let mut down = DownAccounts::default();
        down.failed("ccc".to_owned(), "first", t0, "r");
        down.failed("ccc".to_owned(), "second", t0, "r");

        assert_eq!(down.due(t0 + MAX_SECONDARY_RETRY), Some(("ccc", &"first")));
    }

    /// A client that never touches the network: built, not restored.
    async fn client(host: &str) -> Client {
        Client::builder()
            .homeserver_url(format!("https://{host}.example"))
            .build()
            .await
            .expect("an offline client")
    }

    fn cfg(name: &str) -> AccountCfg {
        AccountCfg {
            name: name.to_owned(),
            token_file: None,
            state_dir: std::env::temp_dir().join(format!("hh-main-test-{name}")),
            homeserver: Some(format!("https://{name}.example")),
        }
    }

    /// A registry holding the primary `main` and a running secondary `other`.
    async fn registry() -> Registry {
        let r = Registry::new("main".to_owned());
        r.insert("main".to_owned(), client("main").await);
        r.insert("other".to_owned(), client("other").await);
        r
    }

    fn down_ccc(at: Instant) -> DownAccounts<Retry> {
        let mut down = DownAccounts::default();
        down.failed("ccc".to_owned(), (cfg("ccc"), None), at, "boot");
        down
    }

    fn snapshot_path(tag: &str) -> std::path::PathBuf {
        let dir = std::env::temp_dir().join(format!("hh-main-snap-{tag}-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        dir.join("matrix-accounts.json")
    }

    fn names(r: &Registry) -> Vec<String> {
        r.list().into_iter().map(|a| a.name).collect()
    }

    fn sent(rx: &mut UnboundedReceiver<NamedSyncLoop>) -> usize {
        std::iter::from_fn(|| rx.try_recv().ok()).count()
    }

    #[tokio::test]
    async fn a_secondary_that_comes_up_joins_the_registry_and_leaves_the_down_set() {
        let t0 = Instant::now();
        let mut down = down_ccc(t0);
        let reg = registry().await;
        let (main_before, other_before) = (
            reg.resolve(Some("main")).unwrap(),
            reg.resolve(Some("other")).unwrap(),
        );
        let (up, mut rx) = unbounded_channel();
        let snap = snapshot_path("up");
        let mut ccc = Some(client("ccc").await);
        let mut bring_up = move |_: AccountCfg| {
            let c = ccc.take().expect("tried once");
            async move { Ok(Some(c)) as Restore }
        };

        let now = t0 + LINKED_REFRESH;
        retry_once(&mut down, &reg, &up, &snap, now, &mut bring_up).await;

        assert!(reg.resolve(Some("ccc")).is_ok());
        assert!(down.due(now + MAX_SECONDARY_RETRY * 10).is_none());
        assert_eq!(sent(&mut rx), 1);
        // The accounts already up are the very same clients.
        assert!(Arc::ptr_eq(
            &main_before,
            &reg.resolve(Some("main")).unwrap()
        ));
        assert!(Arc::ptr_eq(
            &other_before,
            &reg.resolve(Some("other")).unwrap()
        ));
        let published = std::fs::read_to_string(&snap).unwrap();
        assert!(
            published.contains("\"ccc\""),
            "snapshot lists it: {published}"
        );
        std::fs::remove_dir_all(snap.parent().unwrap()).ok();
    }

    #[tokio::test]
    async fn a_secondary_that_fails_stays_down_with_a_longer_wait_and_disturbs_nobody() {
        for outcome in [Err(anyhow!("whoami: 502")), Ok(None)] {
            let t0 = Instant::now();
            let mut down = down_ccc(t0);
            let reg = registry().await;
            let main_before = reg.resolve(Some("main")).unwrap();
            let (up, mut rx) = unbounded_channel();
            let snap = snapshot_path("fail");
            let mut outcome = Some(outcome);
            let mut bring_up = move |_: AccountCfg| {
                let r = outcome.take().expect("tried once");
                async move { r as Restore }
            };

            let now = t0 + LINKED_REFRESH;
            retry_once(&mut down, &reg, &up, &snap, now, &mut bring_up).await;

            // Still down; its second failure waits twice as long as its first.
            assert!(reg.resolve(Some("ccc")).is_err());
            assert_eq!(names(&reg), ["main", "other"]);
            assert_eq!(sent(&mut rx), 0);
            assert!(!snap.exists(), "no snapshot for a failed try");
            assert!(
                down.due(now + retry_delay(2) - Duration::from_secs(1))
                    .is_none()
            );
            assert!(down.due(now + retry_delay(2)).is_some());
            assert!(Arc::ptr_eq(
                &main_before,
                &reg.resolve(Some("main")).unwrap()
            ));
        }
    }

    #[tokio::test]
    async fn a_secondary_that_comes_up_late_does_not_wipe_the_todos() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let snap = snapshot_path("todos");
        let socket = snap.with_file_name("agent.sock");
        let listener = tokio::net::UnixListener::bind(&socket).unwrap();
        let ops = Arc::new(std::sync::Mutex::new(Vec::new()));
        let seen = Arc::clone(&ops);
        let harness = tokio::spawn(async move {
            loop {
                let (mut conn, _) = listener.accept().await.unwrap();
                let mut op = String::new();
                conn.read_to_string(&mut op).await.unwrap();
                seen.lock().unwrap().push(op);
                conn.write_all(b"{\"ok\":true}\n").await.unwrap();
            }
        });
        // Startup's wipe, as `main` runs it: the one the harness should see.
        wake::clear_all_todos(&socket).await.unwrap();

        let t0 = Instant::now();
        let mut down = down_ccc(t0);
        let reg = registry().await;
        let (up, mut rx) = unbounded_channel();
        let mut ccc = Some(client("ccc").await);
        let mut restore = move |_: AccountCfg| {
            let c = ccc.take().expect("tried once");
            async move { Ok(Some(c)) as Restore }
        };
        retry_once(
            &mut down,
            &reg,
            &up,
            &snap,
            t0 + LINKED_REFRESH,
            &mut restore,
        )
        .await;
        // Run the late account's sync loop; an offline client never syncs.
        let late = rx.try_recv().expect("its sync loop is handed on");
        let _ = tokio::time::timeout(Duration::from_secs(2), late).await;

        harness.abort();
        let ops = ops.lock().unwrap().clone();
        let wipes = ops.iter().filter(|op| op.contains("\"all\":true")).count();
        assert_eq!(wipes, 1, "todo ops the harness saw: {ops:?}");
        std::fs::remove_dir_all(snap.parent().unwrap()).ok();
    }

    #[tokio::test]
    async fn an_account_whose_wait_is_not_over_is_not_tried() {
        let t0 = Instant::now();
        let mut down = down_ccc(t0);
        let reg = registry().await;
        let (up, _rx) = unbounded_channel();
        let mut tried = 0;
        let mut bring_up = |_: AccountCfg| {
            tried += 1;
            async { Ok(None) as Restore }
        };

        let early = t0 + LINKED_REFRESH - Duration::from_secs(1);
        retry_once(
            &mut down,
            &reg,
            &up,
            Path::new("/nonexistent/x.json"),
            early,
            &mut bring_up,
        )
        .await;

        assert_eq!(tried, 0);
    }

    #[tokio::test]
    async fn readers_are_not_blocked_while_a_bring_up_is_pending() {
        let t0 = Instant::now();
        let mut down = down_ccc(t0);
        let reg = Arc::new(registry().await);
        let (up, _rx) = unbounded_channel();
        let snap = snapshot_path("pending");
        let (gate_tx, gate_rx) = tokio::sync::oneshot::channel::<()>();
        let mut gate_rx = Some(gate_rx);
        let mut ccc = Some(client("ccc").await);
        let mut bring_up = move |_: AccountCfg| {
            let gate = gate_rx.take().expect("tried once");
            let c = ccc.take().expect("tried once");
            async move {
                gate.await.unwrap();
                Ok(Some(c)) as Restore
            }
        };

        // A reader on its own thread, so a lock held across the bring-up
        // blocks it, not the runtime: it must finish before the gate opens.
        let (done_tx, done_rx) = tokio::sync::oneshot::channel();
        let reader_reg = Arc::clone(&reg);
        let reader = std::thread::spawn(move || {
            let seen = reader_reg.list().len();
            let resolved = reader_reg.resolve(Some("main")).is_ok();
            let _ = done_tx.send((seen, resolved));
        });
        let retry = retry_once(
            &mut down,
            &reg,
            &up,
            &snap,
            t0 + LINKED_REFRESH,
            &mut bring_up,
        );
        let open_gate_after_reader = async {
            let seen = done_rx.await.unwrap();
            gate_tx.send(()).unwrap();
            seen
        };
        let ((), seen) = tokio::time::timeout(Duration::from_secs(10), async {
            tokio::join!(retry, open_gate_after_reader)
        })
        .await
        .expect("a reader was blocked by a pending bring-up");

        assert_eq!(seen, (2, true));
        assert!(reg.resolve(Some("ccc")).is_ok());
        reader.join().unwrap();
        std::fs::remove_dir_all(snap.parent().unwrap()).ok();
    }

    #[test]
    fn inserting_while_readers_run_neither_deadlocks_nor_loses_an_account() {
        // The whole scenario runs on its own thread: a deadlock blocks that
        // thread, and only the channel timeout below can see it.
        let (result_tx, result_rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
            rt.block_on(async {
                let t0 = Instant::now();
                let mut down = DownAccounts::default();
                for n in ["a", "b", "c", "d"] {
                    down.failed(n.to_owned(), (cfg(n), None), t0, "boot");
                }
                let reg = Arc::new(registry().await);
                let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
                let readers: Vec<_> = (0..4)
                    .map(|_| {
                        let (reg, stop) = (Arc::clone(&reg), Arc::clone(&stop));
                        std::thread::spawn(move || {
                            while !stop.load(std::sync::atomic::Ordering::Relaxed) {
                                let _ = reg.list();
                                assert!(reg.resolve(Some("main")).is_ok());
                            }
                        })
                    })
                    .collect();
                let (up, _rx) = unbounded_channel();
                let snap = snapshot_path("readers");
                let mut bring_up =
                    |c: AccountCfg| async move { Ok(Some(client(&c.name).await)) as Restore };
                let mut now = t0 + LINKED_REFRESH;
                for _ in 0..4 {
                    retry_once(&mut down, &reg, &up, &snap, now, &mut bring_up).await;
                    now += Duration::from_secs(1);
                }
                stop.store(true, std::sync::atomic::Ordering::Relaxed);
                for r in readers {
                    r.join().unwrap();
                }
                std::fs::remove_dir_all(snap.parent().unwrap()).ok();
                let mut got = names(&reg);
                got.sort();
                result_tx.send(got).unwrap();
            });
        });
        let got = result_rx
            .recv_timeout(Duration::from_secs(20))
            .expect("deadlock or panic while inserting under readers");
        assert_eq!(got, ["a", "b", "c", "d", "main", "other"]);
    }

    #[tokio::test(start_paused = true)]
    async fn the_loop_retries_a_failing_account_on_a_growing_wait() {
        let t0 = Instant::now();
        let mut down = down_ccc(t0);
        let reg = registry().await;
        let (up, _rx) = unbounded_channel();
        let calls = std::cell::RefCell::new(Vec::new());
        let bring_up = |_: AccountCfg| {
            calls.borrow_mut().push(t0.elapsed().as_secs() / 60);
            async { Err(anyhow!("homeserver down")) as Restore }
        };

        let _ = tokio::time::timeout(
            Duration::from_mins(31),
            retry_loop(
                &mut down,
                &reg,
                &up,
                Path::new("/nonexistent/x.json"),
                bring_up,
            ),
        )
        .await;

        // Waits of 2, 4, 8 and 16 minutes after the failures at 0, 2, 6, 14.
        assert_eq!(*calls.borrow(), [2, 6, 14, 30]);
        assert_eq!(names(&reg), ["main", "other"]);
    }
}