feat(c0re): /api/stats-hive — hive-wide turn-stats rollup

#1424 P2 backend. New hive_stats module aggregates every agent's
turn-stats.sqlite read-only (reusing Coordinator::kept_state_names +
agent_harness_dir, skipping missing/unreadable dbs) into swarm totals,
a busiest-first per-agent rollup, swarm model mix, and a labelled USD
cost estimate (rough model->price table; can move to a nix option
later). Exposed as GET /api/stats-hive?window=. Dashboard UI follows.

hive-c0re/src/dashboard.rs

@@ -67,6 +67,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
         .route("/api/approval-diff/{id}", get(get_approval_diff))
         .route("/api/state-file", get(get_state_file))
         .route("/api/reminders", get(api_reminders))
+        .route("/api/stats-hive", get(api_stats_hive))
         .route("/api/build-logs", get(get_build_logs_all))
         .route("/api/build-logs/{agent}", get(get_build_logs_agent))
         .route("/api/build-logs/id/{id}", get(get_build_log_full))
@@ -1734,6 +1735,19 @@ async fn api_reminders(State(state): State<AppState>) -> Response {
     }
 }
 
+#[derive(Deserialize)]
+struct StatsHiveQuery {
+    window: Option<String>,
+}
+
+/// Hive-wide turn-stats rollup for the dashboard swarm-stats view.
+/// Aggregates every agent's `hyperhive-turn-stats.sqlite` read-only
+/// (skips missing/unreadable ones). Window defaults to `24h`.
+async fn api_stats_hive(axum::extract::Query(q): axum::extract::Query<StatsHiveQuery>) -> Response {
+    let window = crate::hive_stats::Window::parse(q.window.as_deref().unwrap_or("24h"));
+    axum::Json(crate::hive_stats::hive_snapshot(window)).into_response()
+}
+
 #[derive(Deserialize)]
 struct BuildLogsQuery {
     /// Maximum number of rows to return. Capped server-side at 50

hive-c0re/src/hive_stats.rs

@@ -0,0 +1,293 @@
+//! Hive-wide turn-stats aggregation for the dashboard's swarm stats
+//! view. Reads every agent's per-agent
+//! `hyperhive-turn-stats.sqlite` read-only and rolls the rows up into
+//! swarm totals + a per-agent rollup + model mix + a *labelled*
+//! cost estimate.
+//!
+//! Why re-read the rows here instead of reusing `hive-ag3nt`'s
+//! `stats.rs`: that module lives in a different crate (the agent
+//! harness) which hive-c0re can't import. The stable contract is the
+//! turn-stats *schema*, so we run a focused query against it. If we
+//! ever want a single source of truth, the row-read + aggregation can
+//! be lifted into a shared crate — overkill for now.
+//!
+//! Privsep: the sqlite files are mode 0644 owned by the agent user;
+//! `hive-core` reads them fine (same as `stats_vacuum`). We open
+//! read-only so an in-flight harness writer never blocks us.
+
+use std::collections::HashMap;
+use std::path::Path;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use rusqlite::{Connection, OpenFlags};
+use serde::Serialize;
+
+use crate::coordinator::Coordinator;
+
+/// Window accepted by `/api/stats-hive?window=`. Maps to a lookback
+/// span; the hive view is a flat rollup (no per-bucket trend — the
+/// per-agent `/stats` page owns the trend charts).
+#[derive(Debug, Clone, Copy)]
+pub enum Window {
+    Hour,
+    FourHour,
+    Day,
+    ThreeDay,
+    Week,
+    Month,
+}
+
+impl Window {
+    #[must_use]
+    pub fn parse(s: &str) -> Self {
+        match s {
+            "1h" => Self::Hour,
+            "4h" => Self::FourHour,
+            "3d" => Self::ThreeDay,
+            "7d" => Self::Week,
+            "30d" => Self::Month,
+            _ => Self::Day,
+        }
+    }
+
+    fn label(self) -> &'static str {
+        match self {
+            Self::Hour => "1h",
+            Self::FourHour => "4h",
+            Self::Day => "24h",
+            Self::ThreeDay => "3d",
+            Self::Week => "7d",
+            Self::Month => "30d",
+        }
+    }
+
+    fn span_secs(self) -> i64 {
+        match self {
+            Self::Hour => 3600,
+            Self::FourHour => 4 * 3600,
+            Self::Day => 24 * 3600,
+            Self::ThreeDay => 3 * 24 * 3600,
+            Self::Week => 7 * 24 * 3600,
+            Self::Month => 30 * 24 * 3600,
+        }
+    }
+}
+
+/// Approximate USD price per **million** tokens, per model class.
+/// Matched by substring against the model id. This is a deliberately
+/// rough estimate — Anthropic list pricing drifts, so the dashboard
+/// labels the figure as an estimate. A follow-up can wire it from a
+/// nix option (like `contextWindowTokens`) instead of hard-coding.
+struct Prices {
+    input: f64,
+    output: f64,
+    cache_read: f64,
+    cache_write: f64,
+}
+
+fn model_prices(model: &str) -> Prices {
+    let m = model.to_ascii_lowercase();
+    if m.contains("opus") {
+        Prices {
+            input: 15.0,
+            output: 75.0,
+            cache_read: 1.5,
+            cache_write: 18.75,
+        }
+    } else if m.contains("haiku") {
+        Prices {
+            input: 0.8,
+            output: 4.0,
+            cache_read: 0.08,
+            cache_write: 1.0,
+        }
+    } else {
+        // sonnet + unknown fallback
+        Prices {
+            input: 3.0,
+            output: 15.0,
+            cache_read: 0.3,
+            cache_write: 3.75,
+        }
+    }
+}
+
+#[derive(Debug, Serialize)]
+pub struct KeyCount {
+    pub key: String,
+    pub count: u64,
+}
+
+#[derive(Debug, Serialize)]
+pub struct AgentRollup {
+    pub name: String,
+    pub turns: u64,
+    pub input_tokens: u64,
+    pub output_tokens: u64,
+    pub cache_read_tokens: u64,
+    pub cache_creation_tokens: u64,
+    /// Labelled estimate — see [`Prices`].
+    pub est_cost_usd: f64,
+}
+
+#[derive(Debug, Serialize)]
+pub struct HiveStats {
+    pub window: &'static str,
+    pub from: i64,
+    pub now: i64,
+    /// Number of agents that had at least one turn in the window.
+    pub active_agents: u64,
+    pub total_turns: u64,
+    pub total_input_tokens: u64,
+    pub total_output_tokens: u64,
+    pub total_cache_read_tokens: u64,
+    pub total_cache_creation_tokens: u64,
+    /// Labelled estimate — see [`Prices`].
+    pub est_cost_usd: f64,
+    /// Per-agent rollup, busiest (most turns) first.
+    pub agents: Vec<AgentRollup>,
+    /// Turns per model across the whole swarm, busiest first.
+    pub model_mix: Vec<KeyCount>,
+}
+
+#[derive(Default)]
+struct AgentAgg {
+    turns: u64,
+    input: u64,
+    output: u64,
+    cache_read: u64,
+    cache_creation: u64,
+    cost: f64,
+    models: HashMap<String, u64>,
+}
+
+fn now_secs() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
+}
+
+#[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)]
+fn u64_from_i64(v: i64) -> u64 {
+    v.max(0) as u64
+}
+
+/// Aggregate one agent's turn-stats over `[from, now]`. Errors bubble
+/// up so the caller can skip a single bad/locked db without failing
+/// the whole endpoint.
+fn read_agent(path: &Path, from: i64) -> rusqlite::Result<AgentAgg> {
+    let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
+    let mut stmt = conn.prepare(
+        "SELECT model, input_tokens, output_tokens,
+                cache_read_input_tokens, cache_creation_input_tokens
+         FROM turn_stats
+         WHERE started_at >= ?1",
+    )?;
+    let mut agg = AgentAgg::default();
+    let rows = stmt.query_map([from], |row| {
+        Ok((
+            row.get::<_, String>(0)?,
+            u64_from_i64(row.get::<_, i64>(1)?),
+            u64_from_i64(row.get::<_, i64>(2)?),
+            u64_from_i64(row.get::<_, i64>(3)?),
+            u64_from_i64(row.get::<_, i64>(4)?),
+        ))
+    })?;
+    for r in rows {
+        let (model, input, output, cache_read, cache_creation) = r?;
+        agg.turns += 1;
+        agg.input = agg.input.saturating_add(input);
+        agg.output = agg.output.saturating_add(output);
+        agg.cache_read = agg.cache_read.saturating_add(cache_read);
+        agg.cache_creation = agg.cache_creation.saturating_add(cache_creation);
+        let p = model_prices(&model);
+        #[allow(clippy::cast_precision_loss)]
+        {
+            agg.cost += (input as f64 * p.input
+                + output as f64 * p.output
+                + cache_read as f64 * p.cache_read
+                + cache_creation as f64 * p.cache_write)
+                / 1_000_000.0;
+        }
+        *agg.models.entry(model).or_insert(0) += 1;
+    }
+    Ok(agg)
+}
+
+/// Build the swarm-wide rollup. Best-effort: a missing or unreadable
+/// per-agent db is skipped (logged), never fatal.
+#[must_use]
+pub fn hive_snapshot(window: Window) -> HiveStats {
+    let now = now_secs();
+    let from = now - window.span_secs();
+
+    let mut agents: Vec<AgentRollup> = Vec::new();
+    let mut model_mix: HashMap<String, u64> = HashMap::new();
+    let mut total_turns = 0u64;
+    let mut total_input = 0u64;
+    let mut total_output = 0u64;
+    let mut total_cache_read = 0u64;
+    let mut total_cache_creation = 0u64;
+    let mut total_cost = 0.0f64;
+    let mut active_agents = 0u64;
+
+    for name in Coordinator::kept_state_names() {
+        let path = Coordinator::agent_harness_dir(&name).join("hyperhive-turn-stats.sqlite");
+        if !path.exists() {
+            continue;
+        }
+        let agg = match read_agent(&path, from) {
+            Ok(a) => a,
+            Err(e) => {
+                tracing::warn!(agent = %name, error = ?e, "hive-stats: read failed; skipping");
+                continue;
+            }
+        };
+        if agg.turns == 0 {
+            continue;
+        }
+        active_agents += 1;
+        total_turns += agg.turns;
+        total_input = total_input.saturating_add(agg.input);
+        total_output = total_output.saturating_add(agg.output);
+        total_cache_read = total_cache_read.saturating_add(agg.cache_read);
+        total_cache_creation = total_cache_creation.saturating_add(agg.cache_creation);
+        total_cost += agg.cost;
+        for (m, c) in &agg.models {
+            *model_mix.entry(m.clone()).or_insert(0) += c;
+        }
+        agents.push(AgentRollup {
+            name,
+            turns: agg.turns,
+            input_tokens: agg.input,
+            output_tokens: agg.output,
+            cache_read_tokens: agg.cache_read,
+            cache_creation_tokens: agg.cache_creation,
+            est_cost_usd: agg.cost,
+        });
+    }
+
+    // Busiest agents first.
+    agents.sort_by(|a, b| b.turns.cmp(&a.turns).then_with(|| a.name.cmp(&b.name)));
+
+    let mut model_mix: Vec<KeyCount> = model_mix
+        .into_iter()
+        .map(|(key, count)| KeyCount { key, count })
+        .collect();
+    model_mix.sort_by(|a, b| b.count.cmp(&a.count).then_with(|| a.key.cmp(&b.key)));
+
+    HiveStats {
+        window: window.label(),
+        from,
+        now,
+        active_agents,
+        total_turns,
+        total_input_tokens: total_input,
+        total_output_tokens: total_output,
+        total_cache_read_tokens: total_cache_read,
+        total_cache_creation_tokens: total_cache_creation,
+        est_cost_usd: total_cost,
+        agents,
+        model_mix,
+    }
+}

hive-c0re/src/lib.rs

@@ -32,6 +32,7 @@ pub mod events_vacuum;
 pub mod flake_check;
 pub mod forge;
 pub mod gateway_nginx;
+pub mod hive_stats;
 pub mod knowledge;
 pub mod lifecycle;
 pub mod limits;