use std::path::PathBuf;
use std::sync::Arc;

use anyhow::{Context as _, Result};
use clap::{Parser, Subcommand};

// `hive-c0re` is bin-only: this binary owns the whole daemon module
// tree. The operator CLI moved to the standalone `hivectl` crate (it
// talks to the daemon over the host admin socket), so there's no longer
// a library shared between two binaries — the modules that used to live
// in `src/lib.rs` are declared here directly.
//
// Cohesive clusters live in directory submodules (`stores`, `stats`,
// `agent_config`, `workers`); each child is re-exported at the crate
// root so `crate::broker::…` style paths keep resolving unchanged.
mod actions;
mod agent_config;
mod container_view;
mod coordinator;
mod dashboard;
mod dashboard_events;
mod forge;
mod gateway_nginx;
mod job_queue;
mod lifecycle;
mod loose_ends;
mod matrix;
mod meta;
mod migrate;
mod paths;
mod priv_client;
mod questions;
mod server;
mod snapshot_push;
mod socket_server;
mod stats;
mod stores;
mod webhook_secret;
mod workers;

pub(crate) use agent_config::{capabilities, limits, resource_limits, tool_groups, topology};
pub(crate) use stats::{
    container_stats, hive_stats, host_stats, otel_metrics, sweep_health, warnings,
};
pub(crate) use stores::{
    approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts,
};
pub(crate) use workers::{
    agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, scheduled_prompts_worker,
};

use coordinator::{Coordinator, HiveEnv, ServeConfig};

#[derive(Parser)]
#[command(name = "hive-c0re", about = "hyperhive coordinator daemon")]
struct Cli {
    /// Path to the host admin socket.
    #[arg(long, global = true, default_value = crate::paths::HOST_SOCKET)]
    socket: PathBuf,

    #[command(subcommand)]
    cmd: Cmd,
}

#[derive(Subcommand)]
enum Cmd {
    /// Run the coordinator daemon.
    Serve {
        /// Path to a JSON config file holding the host-level daemon config
        /// (the [`ServeConfig`](crate::coordinator::ServeConfig) shape:
        /// the container-injected `HiveEnv` fields + the hive-c0re-local
        /// `model_prices` table). Used as the base; any per-flag override
        /// below wins over the file. Absent → start from the built-in
        /// defaults. The NixOS module passes a generated config here so the
        /// `ExecStart` stays short instead of carrying every setting — and
        /// the context-window / price-table JSON blobs — as flags.
        #[arg(long)]
        config: Option<PathBuf>,
        /// Path to the sqlite message store.
        #[arg(long, default_value = crate::paths::BROKER_DB)]
        db: PathBuf,
        /// Override: URL of the hyperhive flake. Inlined into each
        /// per-agent `flake.nix` as the `hyperhive` input.
        #[arg(long)]
        hyperhive_flake: Option<String>,
        /// Override: store-path URL of the nixpkgs to wire into the meta
        /// flake as `inputs.nixpkgs.url`. Empty = legacy
        /// `follows = "hyperhive/nixpkgs"` fallback.
        #[arg(long)]
        nixpkgs_flake: Option<String>,
        /// Override: dashboard HTTP port.
        #[arg(long)]
        dashboard_port: Option<u16>,
        /// Override: operator pronouns (free text). Threaded into each
        /// container's harness via `HIVE_OPERATOR_PRONOUNS`.
        #[arg(long)]
        operator_pronouns: Option<String>,
        /// Override: per-model context-window sizes, as a JSON object
        /// mapping model-family short name to token count. Threaded into
        /// each container as `HIVE_CONTEXT_WINDOW_TOKENS_<KEY_UPPER>` env
        /// vars. Set via the `services.hive-c0re.contextWindowTokens`
        /// NixOS option.
        #[arg(long)]
        context_window_tokens: Option<String>,
        /// Override: systemd `CPUQuota=` applied to every agent container
        /// via a drop-in (e.g. `"200%"` = 2 cores). Set via
        /// `services.hyperhive.agentCpuQuota`.
        #[arg(long)]
        agent_cpu_quota: Option<String>,
        /// Override: systemd `MemoryMax=` applied to every agent
        /// container. Set via `services.hyperhive.agentMemoryMax`.
        #[arg(long)]
        agent_memory_max: Option<String>,
        /// Override: systemd `CPUWeight=` applied to every agent container
        /// via the same drop-in — a cgroup v2 relative share under
        /// contention (1..=10000), not a cap. Set via
        /// `services.hyperhive.agentCpuWeight`.
        #[arg(long)]
        agent_cpu_weight: Option<u32>,
        /// Override: systemd `IOWeight=` applied to every agent container,
        /// the block-IO counterpart of `--agent-cpu-weight`. Set via
        /// `services.hyperhive.agentIoWeight`.
        #[arg(long)]
        agent_io_weight: Option<u32>,
        /// Override: per-model USD prices (per million tokens) for the
        /// hive-wide ST4TS cost estimate, as a JSON object mapping a
        /// model-family short name to `{input, output, cache_read,
        /// cache_write}`. Models not covered fall back to the built-in
        /// estimate. Set via the `services.hyperhive.modelPrices` NixOS
        /// option.
        #[arg(long)]
        model_prices: Option<String>,
        /// Override: number of concurrent nix-heavy job-queue nodes
        /// (prebuild / profile-swap / create / meta lock). Set via the
        /// `services.hyperhive.c0re.buildSlots` NixOS option.
        #[arg(long)]
        build_slots: Option<usize>,
    },
}

#[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")),
        )
        .init();

    let cli = Cli::parse();
    match cli.cmd {
        Cmd::Serve {
            config,
            db,
            hyperhive_flake,
            nixpkgs_flake,
            dashboard_port,
            operator_pronouns,
            context_window_tokens,
            agent_cpu_quota,
            agent_memory_max,
            agent_cpu_weight,
            agent_io_weight,
            model_prices,
            build_slots,
        } => {
            // Base config from the --config file (or the built-in
            // defaults), then apply any per-flag overrides — config
            // file is the base, explicit flags win.
            let mut sc = match &config {
                Some(p) => {
                    let s = std::fs::read_to_string(p)
                        .with_context(|| format!("read --config {}", p.display()))?;
                    serde_json::from_str::<ServeConfig>(&s)
                        .with_context(|| format!("parse --config {}", p.display()))?
                }
                None => ServeConfig::default(),
            };
            if let Some(v) = hyperhive_flake {
                sc.env.hyperhive_flake = v;
            }
            if let Some(v) = nixpkgs_flake {
                sc.env.nixpkgs_flake = v;
            }
            if let Some(v) = dashboard_port {
                sc.env.dashboard_port = v;
            }
            if let Some(v) = operator_pronouns {
                sc.env.operator_pronouns = v;
            }
            if let Some(v) = context_window_tokens {
                sc.env.context_window_tokens =
                    serde_json::from_str(&v).context("--context-window-tokens: invalid JSON")?;
            }
            if let Some(v) = agent_cpu_quota {
                sc.env.agent_cpu_quota = v;
            }
            if let Some(v) = agent_memory_max {
                sc.env.agent_memory_max = v;
            }
            // Passing the flag sets a weight; omitting it keeps whatever the
            // config file says (including `null` = don't emit the setting).
            // There's deliberately no flag spelling for "clear it" — that's
            // what the nix option's `null` is for.
            if let Some(v) = agent_cpu_weight {
                sc.env.agent_cpu_weight = Some(v);
            }
            if let Some(v) = agent_io_weight {
                sc.env.agent_io_weight = Some(v);
            }
            if let Some(v) = model_prices {
                sc.model_prices =
                    serde_json::from_str(&v).context("--model-prices: invalid JSON")?;
            }
            if let Some(v) = build_slots {
                sc.build_slots = v;
            }
            cmd_serve(sc.env, sc.model_prices, sc.build_slots, db, &cli.socket).await
        }
    }
}

/// Banner message for a failing matrix `ensure_all` sweep, shared by both
/// the initial and periodic `record_err` call sites in `cmd_serve` so the
/// wording can't drift between them.
fn matrix_sweep_banner(ctx: sweep_health::SweepFailure) -> String {
    let age = ctx.since_last_ok.map_or_else(
        || "no success this session".to_owned(),
        |d| format!("last ok {} ago", sweep_health::fmt_age(d)),
    );
    format!(
        "matrix user/space sweep failing ({} consecutive, {age}) \
         — some agents may be missing matrix accounts, space membership, \
         or chat-room invites",
        ctx.consecutive
    )
}

/// Start the coordinator daemon: open the broker, run migrations, spawn
/// background tasks (auto-update, vacuums, crash-watcher, scheduled-prompts,
/// dashboard), then serve the admin socket until a signal arrives.
#[allow(
    clippy::too_many_lines,
    reason = "startup orchestration: open the broker, run migrations, then spawn \
              the full set of background services (auto-update, vacuums, \
              crash-watch, schedulers, dashboard) before the serve loop — the \
              length is inherent to booting the daemon, not the arg list"
)]
async fn cmd_serve(
    env: HiveEnv,
    model_prices: crate::hive_stats::PriceTable,
    build_slots: usize,
    db: std::path::PathBuf,
    socket: &std::path::Path,
) -> Result<()> {
    // Move any host-side state still at the legacy flat layout into its
    // subdir (`db/`, `forge/`, `matrix/`, `run/`) BEFORE opening the
    // broker db — the broker + build-logs dbs are among the relocated
    // files. Idempotent; a no-op once migrated.
    crate::paths::relocate_legacy_state();
    // `dashboard_port` is consumed into the Coordinator below; capture the
    // Copy value first for the dashboard + knowledge-webhook tasks.
    let dashboard_port = env.dashboard_port;
    let coord = Arc::new(Coordinator::open(&db, env, model_prices, build_slots)?);
    socket_server::start_manager(coord.clone())?;
    // Idempotent pre-flight: rewrite pre-meta-layout applied
    // repos, ensure proposed repos carry the `applied`
    // remote, bootstrap the meta repo, repoint containers at
    // `meta#<name>` (one-shot, guarded by a marker file).
    // Runs before manager auto-spawn so the new manager is
    // built against meta from the first attempt.
    if let Err(e) = migrate::run(&coord).await {
        tracing::warn!(error = ?e, "startup migration failed");
    }
    // Auto-create the root agent container if it isn't there yet. Block
    // on this — without root the system has no manager harness.
    // Failures are logged but allowed: a broken auto-spawn shouldn't
    // make the dashboard unreachable for debugging.
    if let Err(e) = auto_update::ensure_root_agent(&coord).await {
        tracing::warn!(error = ?e, "auto-spawn root agent failed");
    }
    // Sync /etc/tmpfiles.d/hyperhive-agents.conf so agent runtime dirs are
    // pre-declared for the next boot. Best-effort background task — a failure
    // here must not block hive-c0re startup. See lifecycle::sync_tmpfiles.
    tokio::spawn(crate::lifecycle::sync_tmpfiles());
    // Auto-update in the background — don't block service start.
    // Sub-agent rebuilds can take tens of seconds; we want the admin
    // socket up immediately.
    let update_coord = coord.clone();
    tokio::spawn(async move {
        if let Err(e) = auto_update::run(update_coord).await {
            tracing::warn!(error = ?e, "auto-update task failed");
        }
    });
    // Forge user sweep: ensure every existing container has a
    // forgejo user + access token. No-op when the hive-forge
    // container isn't running. Backgrounded — touches the
    // forge state dir via `nixos-container run` which is slow.
    tokio::spawn(async move {
        forge::ensure_all().await;
    });
    // Webhook HMAC secret: load from state dir or generate on first run.
    // Used by both the webhook handlers (verification) and the Forgejo
    // hook registrations (so Forgejo signs deliveries with the same key).
    let webhook_secret: Option<String> = match crate::webhook_secret::load_or_generate() {
        Ok(s) => Some(s),
        Err(e) => {
            tracing::error!(
                error = ?e,
                "webhook secret load/generate failed; /webhook/* endpoints disabled and hooks not registered"
            );
            None
        }
    };
    // Webhook setup: ensure Forgejo webhooks are registered for both
    // `internal/knowledge` (push → git pull) and the `agent-configs` org
    // (pull_request → queue MergeConfigPr approval). Both run after
    // forge::ensure_all so the core token + repos + org are present.
    // URLs use the public hive domain (HYPERHIVE_HIVE_DOMAIN) so Forgejo
    // delivers through the gateway, bypassing the SSRF loopback guard.
    // No-op when the core token or domain are absent, or when the HMAC
    // secret is unavailable (load failure).
    let webhook_secret_reg = webhook_secret.clone();
    tokio::spawn(async move {
        let Some(webhook_secret_reg) = webhook_secret_reg else {
            tracing::debug!("webhook secret unavailable; skipping hook registration");
            return;
        };
        let Some(token) = forge::core_token() else {
            return;
        };
        let domain = std::env::var("HYPERHIVE_HIVE_DOMAIN")
            .ok()
            .filter(|v| !v.is_empty());
        let Some(domain) = domain else {
            tracing::debug!("HYPERHIVE_HIVE_DOMAIN unset; skipping webhook registration");
            return;
        };
        if let Err(e) = knowledge::ensure_webhook(&token, &domain, &webhook_secret_reg).await {
            tracing::warn!(error = ?e, "knowledge: ensure_webhook failed");
        }
        if let Err(e) = forge::ensure_config_pr_webhook(&token, &domain, &webhook_secret_reg).await
        {
            tracing::warn!(error = ?e, "forge: ensure_config_pr_webhook failed");
        }
    });
    // Config-PR polling fallback: scan agent-configs org every 5 minutes
    // for open PRs that have no pending MergeConfigPr approval. Catches
    // anything the webhook missed (c0re was down when PR opened, delivery
    // failed, etc.). First sweep fires immediately on startup.
    let poll_coord = coord.clone();
    let mut poll_shutdown = coord.shutdown_rx();
    tokio::spawn(async move {
        let interval = std::time::Duration::from_mins(5);
        loop {
            if let Some(token) = forge::core_token() {
                let result = Box::pin(forge::config_pr_poll::poll_open_config_prs(
                    &token,
                    &poll_coord,
                ))
                .await;
                if let Err(e) = result {
                    tracing::debug!(error = ?e, "config-pr poll: sweep failed (forge may be absent)");
                }
            }
            tokio::select! {
                () = tokio::time::sleep(interval) => {}
                _ = poll_shutdown.changed() => {
                    tracing::info!("config-pr poll: shutdown signal received");
                    break;
                }
            }
        }
    });
    // Knowledge periodic pull: hourly fallback in case the webhook is
    // missed (e.g. hive-c0re was down during a push). First fires at
    // startup (immediate pull after the clone is already present).
    let mut knowledge_shutdown = coord.shutdown_rx();
    let knowledge_coord = coord.clone();
    tokio::spawn(async move {
        // Initial pull — reconcile any commits that landed while c0re
        // was offline. Not fed to the health tracker: a startup miss is
        // expected (the clone may not exist yet) and is logged at debug.
        if let Err(e) = knowledge::pull(&knowledge_coord).await {
            tracing::debug!(error = ?e, "knowledge: startup pull skipped (no clone yet?)");
        }
        // Persistent-failure → banner. An hourly sweep that keeps failing for
        // several hours means the operator's `/knowledge` is drifting; raise a
        // warn banner after 3 consecutive misses so a one-off network blip
        // self-heals on the next tick without ever bannering. Cleared on the
        // next successful pull.
        let mut health = sweep_health::SweepHealth::new("knowledge_pull", "warn", 3);
        let interval = std::time::Duration::from_hours(1);
        loop {
            tokio::select! {
                () = tokio::time::sleep(interval) => {
                    match knowledge::pull(&knowledge_coord).await {
                        Ok(()) => health.record_ok(),
                        Err(e) => {
                            tracing::warn!(error = ?e, "knowledge: periodic pull failed");
                            let err = format!("{e:#}");
                            health.record_err(|ctx| {
                                let age = ctx.since_last_ok.map_or_else(
                                    || "no success this session".to_owned(),
                                    |d| format!("last ok {} ago", sweep_health::fmt_age(d)),
                                );
                                format!(
                                    "knowledge repo pull failing ({} consecutive, {age}) \
                                     — /knowledge is stale until it recovers: {err}",
                                    ctx.consecutive
                                )
                            });
                        }
                    }
                }
                _ = knowledge_shutdown.changed() => {
                    tracing::info!("knowledge pull: shutdown signal received");
                    break;
                }
            }
        }
    });
    // Disk-pressure watch: raise a `disk_pressure` banner warning while the
    // host nix store is over threshold and clear it when back under, via the
    // push-based warnings registry. Replaces the old per-`/api/state`
    // `statvfs`. ~60s cadence; first tick fires immediately. The held guard
    // lives for the task's lifetime — dropping it (on shutdown) clears the
    // banner.
    let mut disk_shutdown = coord.shutdown_rx();
    tokio::spawn(async move {
        let mut disk_guard: Option<warnings::WarningGuard> = None;
        let interval = std::time::Duration::from_mins(1);
        loop {
            host_stats::refresh_disk_warning(&mut disk_guard);
            tokio::select! {
                () = tokio::time::sleep(interval) => {}
                _ = disk_shutdown.changed() => {
                    tracing::info!("disk-pressure watch: shutdown signal received");
                    break;
                }
            }
        }
    });
    // Matrix user sweep: same shape — ensure every container has
    // an account on the local matrix-tuwunel homeserver with an
    // access_token persisted to `<state>/matrix-token`. No-op when
    // the hive-matrix container isn't running. Backgrounded because
    // UIAA is a two-roundtrip dance per agent.
    //
    // Runs once at startup AND periodically every 30 minutes so that
    // token files deleted by `hive-matrix-daemon` (stale-token
    // recovery — `M_UNKNOWN_TOKEN`) get re-provisioned without
    // requiring a hive-c0re restart.
    let mut matrix_shutdown = coord.shutdown_rx();
    tokio::spawn(async move {
        let interval = std::time::Duration::from_mins(30);
        // Debounced banner: a lone bad sweep (homeserver mid-restart, a
        // transient HTTP blip) shouldn't flap the dashboard, but a sweep
        // that's been failing for hours (missing agent invites, a broken
        // admin token) should surface. Cleared the moment a sweep is clean.
        let mut health = sweep_health::SweepHealth::new("matrix_ensure_all", "warn", 2);
        if matrix::ensure_all().await {
            health.record_ok();
        } else {
            health.record_err(matrix_sweep_banner);
        }
        loop {
            tokio::select! {
                () = tokio::time::sleep(interval) => {
                    if matrix::ensure_all().await {
                        health.record_ok();
                    } else {
                        health.record_err(matrix_sweep_banner);
                    }
                }
                _ = matrix_shutdown.changed() => {
                    tracing::info!("matrix ensure_all: shutdown signal received");
                    break;
                }
            }
        }
    });
    // Periodic broker vacuum: drop fully-acked messages older
    // than 30 days. Delivered-but-unacked rows (recoverable via
    // requeue_inflight) and undelivered rows are always kept.
    // Runs hourly; first sweep happens immediately.
    let vacuum_coord = coord.clone();
    let mut vacuum_shutdown = coord.shutdown_rx();
    tokio::spawn(async move {
        let interval = std::time::Duration::from_hours(1);
        let keep_secs: i64 = 30 * 24 * 3600;
        loop {
            match vacuum_coord.broker.vacuum_delivered(keep_secs) {
                Ok(0) => {}
                Ok(n) => tracing::info!(removed = n, "broker vacuum"),
                Err(e) => tracing::warn!(error = ?e, "broker vacuum failed"),
            }
            tokio::select! {
                () = tokio::time::sleep(interval) => {}
                _ = vacuum_shutdown.changed() => {
                    tracing::info!("broker vacuum: shutdown signal received");
                    break;
                }
            }
        }
    });
    // Per-agent events.sqlite + bash-tasks file cleanup now runs
    // agent-side in the harness (`hive_agent::vacuum`): the files are
    // agent-owned, so host-side deletes hit PermissionDenied / readonly-db
    // under privsep. See issue tracker "perms borked".
    // (turn-stats.sqlite has no vacuum — it's one tiny row per turn,
    // ~hundreds of KB, and the /stats + hive-stats views read it
    // directly; pruning it would just lose trend history for no gain.)
    // Slow per-container disk sampler: a `du` of each agent's state dir
    // + writable rootfs every ~5 min, cached so the 5s container-load
    // poll stays cheap cgroup-only reads. Feeds `disk_bytes` on the LOAD
    // tab. See container_stats::disk_sampler_loop.
    crate::container_stats::spawn_disk_sampler();
    // Per-agent container-resource OTEL export: rides the same
    // cgroup gauges out to the configured OTLP endpoint, reusing the hive
    // `services.hyperhive.otel` config (endpoint + LoadCredential auth).
    // No-op when OTEL isn't configured.
    crate::otel_metrics::spawn_exporter(&coord.hyperhive_flake);
    // build_logs.sqlite vacuum: c0re-side (single db). Failures kept
    // 30d, successes 24h — see `build_logs::vacuum` for the rule.
    crate::build_logs::spawn_vacuum(&coord);
    // audit_log.sqlite vacuum: agent-initiated privileged-action trail,
    // 90d retention — see `audit_log::vacuum`.
    crate::audit_log::spawn_vacuum(&coord);
    // Container crash watcher: emits HelperEvent::ContainerCrash
    // when a previously-running container goes away without an
    // operator-initiated transient state.
    crash_watch::spawn(coord.clone());
    // Agent-sockets marker poll: re-fires `agent_sockets::write`
    // and `gateway_nginx::write` every 10s so the JSON and nginx
    // config pick up newly-bound `.bound` markers after a rebuild.
    // Also retries any pending gateway nginx reload that failed on
    // the previous tick. write() is idempotent so steady-state cost
    // is one stat per agent per tick.
    // See `docs/gateway.md::Per-agent unix-socket upstream`.
    agent_sockets::spawn_poll();
    // MCP socket listener startup sync: one-shot sweep that re-registers any
    // running agent container whose MCP listener was lost when hive-c0re
    // restarted (Coordinator starts empty; /run/hyperhive/agents/ is tmpfs).
    // After this, listener registration is event-driven: run_create /
    // run_reconcile call register_agent on start; kill/destroy call
    // unregister_agent. No recurring poll needed — c0re owns the listeners.
    mcp_sockets::sync_on_start(coord.clone()).await;
    // Scheduled-prompts worker: drains due scheduled_prompts rows
    // and fans the body out to each active target's inbox. See
    // scheduled_prompts_worker.rs.
    scheduled_prompts_worker::spawn(coord.clone());
    // Job-queue scheduler: drives the global DAG queue (rebuild /
    // meta-update / spawn / power ops). Concurrency comes from the
    // build-slot count + per-agent leases inside the queue, not from
    // multiple workers — cheap nodes (graceful signals, drains,
    // reconciles) overlap nix-heavy ones structurally. Call sites
    // (auto_update, dashboard, manager, approval handler) submit DAGs
    // instead of awaiting lifecycle work inline. See `job_queue/`.
    {
        let q_coord = coord.clone();
        tokio::spawn(async move {
            job_queue::scheduler::run_worker(q_coord).await;
        });
    }
    // Forward every broker event onto the unified dashboard
    // channel with a freshly-stamped seq, so the dashboard SSE
    // sees broker messages + future mutation events on one
    // stream with one monotonic seq. The broker's intra-process
    // channel (used by `recv_blocking_batch`) stays untouched.
    spawn_broker_to_dashboard_forwarder(coord.clone());
    let dash_coord = coord.clone();
    let dash_secret = webhook_secret.clone();
    tokio::spawn(async move {
        if let Err(e) = dashboard::serve(dashboard_port, dash_coord, dash_secret).await {
            tracing::error!(error = ?e, "dashboard failed");
        }
    });
    // Run the admin socket until a signal arrives; then signal
    // all background tasks so they exit cleanly before the
    // process terminates.
    let coord_sig = coord.clone();
    tokio::select! {
        res = server::serve(socket, coord) => { res? }
        _ = tokio::signal::ctrl_c() => {
            tracing::info!("SIGINT received — requesting shutdown");
            coord_sig.request_shutdown();
        }
        () = async {
            let mut sig = tokio::signal::unix::signal(
                tokio::signal::unix::SignalKind::terminate()
            ).expect("failed to install SIGTERM handler");
            sig.recv().await;
        } => {
            tracing::info!("SIGTERM received — requesting shutdown");
            coord_sig.request_shutdown();
        }
    }
    Ok(())
}

/// Re-emit every broker `MessageEvent` onto the dashboard channel as
/// a `DashboardEvent::Sent` / `Delivered` with a freshly-stamped seq.
/// Background task; runs for the life of the process. On a lagged
/// broker subscription we just keep going — the dashboard channel is
/// best-effort presentation plumbing, the broker keeps its own sqlite
/// log for replay.
fn spawn_broker_to_dashboard_forwarder(coord: Arc<Coordinator>) {
    use broker::MessageEvent;
    use dashboard_events::DashboardEvent;
    let mut rx = coord.broker.subscribe();
    tokio::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(MessageEvent::Sent {
                    id,
                    from,
                    to,
                    body,
                    at,
                    in_reply_to,
                }) => {
                    let file_refs = dashboard::scan_validated_paths(&body);
                    coord.emit_dashboard_event(DashboardEvent::Sent {
                        seq: coord.next_seq(),
                        id,
                        from,
                        to,
                        body,
                        at: hive_sh4re::wire_time::from_secs(at),
                        in_reply_to,
                        file_refs,
                    });
                }
                Ok(MessageEvent::Delivered {
                    id,
                    from,
                    to,
                    body,
                    at,
                    in_reply_to,
                }) => {
                    let file_refs = dashboard::scan_validated_paths(&body);
                    coord.emit_dashboard_event(DashboardEvent::Delivered {
                        seq: coord.next_seq(),
                        id,
                        from,
                        to,
                        body,
                        at: hive_sh4re::wire_time::from_secs(at),
                        in_reply_to,
                        file_refs,
                    });
                }
                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                    tracing::warn!(skipped = n, "broker-to-dashboard forwarder lagged");
                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            }
        }
    });
}