diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 6082ead7..3739027f 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -67,6 +67,7 @@ pub async fn serve(port: u16, coord: Arc) -> 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) -> Response { } } +#[derive(Deserialize)] +struct StatsHiveQuery { + window: Option, +} + +/// 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) -> 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 diff --git a/hive-c0re/src/hive_stats.rs b/hive-c0re/src/hive_stats.rs new file mode 100644 index 00000000..c0d62d96 --- /dev/null +++ b/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, + /// 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)?; + 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 = 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, + } +} diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index 4ed8b412..74cd9ee2 100644 --- a/hive-c0re/src/lib.rs +++ b/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;