//! 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 `events_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::{Duration, 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, /// Turns per model across the whole swarm, busiest first. pub model_mix: Vec, } #[derive(Default)] struct AgentAgg { turns: u64, input: u64, output: u64, cache_read: u64, cache_creation: u64, cost: f64, models: HashMap, } 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 { let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?; // turn_stats is rollback-journal (not WAL): a read landing while the // harness is mid-INSERT would get SQLITE_BUSY and drop that active // agent from the rollup. Wait out the brief write instead. conn.busy_timeout(Duration::from_millis(500))?; 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, reason = "token counts stay well under f64's 2^53 exact-integer range, so this cost computation loses no precision" )] { 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 = Vec::new(); let mut model_mix: HashMap = 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 = 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, } }