remove the hive dashboard's stats page and the price table behind it
The ST4TS page (/stats.html), GET /api/stats-hive, hive-c0re's hive_stats rollup and the modelPrices option that only fed its cost estimate are gone. Each agent's own /stats page and the turn-stats database it reads are unchanged. Setting modelPrices now fails evaluation with a message saying so.
This commit is contained in:
iris
mara
parent
1d3ab2fae5
commit
5e9a22cb38
10 changed files with 112 additions and 1207 deletions
|
|
@ -69,13 +69,6 @@ pub struct Coordinator {
|
|||
/// Same drop-in, same relative-share and `None` semantics as
|
||||
/// [`Self::agent_cpu_weight`].
|
||||
pub agent_io_weight: Option<u32>,
|
||||
/// Operator-tunable model→price table backing the hive-wide cost
|
||||
/// estimate on the ST4TS tab. Set via `services.hyperhive.modelPrices`
|
||||
/// and passed to `hive-c0re serve --model-prices <json>`. Models not
|
||||
/// in the table fall back to the built-in estimate in `hive_stats`.
|
||||
/// hive-c0re-local (read only by the `/api/stats-hive` handler), so it
|
||||
/// is *not* part of `HiveEnv` and is never injected into containers.
|
||||
pub model_prices: crate::hive_stats::PriceTable,
|
||||
agents: Mutex<HashMap<String, AgentSocket>>,
|
||||
/// Tombstone for transients that have JUST been cleared. The
|
||||
/// crash watcher polls every 10s and would race the
|
||||
|
|
@ -215,9 +208,9 @@ impl Default for HiveEnv {
|
|||
|
||||
/// On-disk shape of the `hive-c0re serve --config <file>` JSON: the
|
||||
/// container-injected [`HiveEnv`] (flattened) plus the hive-c0re-local
|
||||
/// `model_prices` table (read only by the `/api/stats-hive` handler, never
|
||||
/// injected into containers — hence kept out of `HiveEnv`). The NixOS
|
||||
/// module writes this so the `ExecStart` carries one `--config` flag
|
||||
/// settings that are never injected into containers — hence kept out of
|
||||
/// `HiveEnv`. The NixOS module writes this so the `ExecStart` carries one
|
||||
/// `--config` flag
|
||||
/// instead of every host-level setting as its own JSON-blob argument.
|
||||
/// `#[serde(default)]` lets any field be omitted and fall back to its
|
||||
/// canonical default.
|
||||
|
|
@ -226,10 +219,9 @@ impl Default for HiveEnv {
|
|||
pub struct ServeConfig {
|
||||
#[serde(flatten)]
|
||||
pub env: HiveEnv,
|
||||
pub model_prices: crate::hive_stats::PriceTable,
|
||||
/// Number of concurrent closure-fetching job-queue nodes (prebuild /
|
||||
/// profile-swap / create / meta lock). hive-c0re-local like
|
||||
/// `model_prices` — never injected into containers. Set via
|
||||
/// profile-swap / create / meta lock). hive-c0re-local — never
|
||||
/// injected into containers. Set via
|
||||
/// `services.hyperhive.deploy.hive-controller.buildSlots`.
|
||||
pub build_slots: usize,
|
||||
}
|
||||
|
|
@ -238,7 +230,6 @@ impl Default for ServeConfig {
|
|||
fn default() -> Self {
|
||||
Self {
|
||||
env: HiveEnv::default(),
|
||||
model_prices: crate::hive_stats::PriceTable::default(),
|
||||
build_slots: 1,
|
||||
}
|
||||
}
|
||||
|
|
@ -250,16 +241,14 @@ mod serve_config_tests {
|
|||
|
||||
#[test]
|
||||
fn deserialises_flat_with_per_field_defaults() {
|
||||
// The flattened HiveEnv fields + the local model_prices all live at
|
||||
// The flattened HiveEnv fields + the local build_slots all live at
|
||||
// the JSON top level. Present fields are read; omitted ones fall
|
||||
// back to their canonical defaults (so a partial config is valid).
|
||||
let sc: ServeConfig = serde_json::from_str(
|
||||
r#"{
|
||||
"hyperhive_flake": "/x",
|
||||
"dashboard_port": 1234,
|
||||
"model_prices": {
|
||||
"opus": {"input": 5.0, "output": 25.0, "cache_read": 0.5, "cache_write": 10.0}
|
||||
}
|
||||
"build_slots": 3
|
||||
}"#,
|
||||
)
|
||||
.expect("flat ServeConfig should deserialise");
|
||||
|
|
@ -267,7 +256,15 @@ mod serve_config_tests {
|
|||
assert_eq!(sc.env.dashboard_port, 1234);
|
||||
// omitted → HiveEnv::default()
|
||||
assert_eq!(sc.env.agent_cpu_quota, "200%");
|
||||
assert!(sc.model_prices.contains_key("opus"));
|
||||
assert_eq!(sc.build_slots, 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_leftover_model_prices_key_is_ignored() {
|
||||
// Configs written before the hive stats page went carry the key.
|
||||
let sc: ServeConfig = serde_json::from_str(r#"{"model_prices": {"opus": {"input": 5.0}}}"#)
|
||||
.expect("an unknown key must not break startup");
|
||||
assert_eq!(sc.build_slots, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -275,7 +272,7 @@ mod serve_config_tests {
|
|||
let sc: ServeConfig = serde_json::from_str("{}").expect("empty config valid");
|
||||
assert_eq!(sc.env.dashboard_port, 7000);
|
||||
assert_eq!(sc.env.hyperhive_flake, "/etc/hyperhive");
|
||||
assert!(sc.model_prices.is_empty());
|
||||
assert_eq!(sc.build_slots, 1);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -336,12 +333,7 @@ impl std::fmt::Display for SetPausedByNameError {
|
|||
}
|
||||
|
||||
impl Coordinator {
|
||||
pub fn open(
|
||||
db_path: &Path,
|
||||
env: HiveEnv,
|
||||
model_prices: crate::hive_stats::PriceTable,
|
||||
build_slots: usize,
|
||||
) -> Result<Self> {
|
||||
pub fn open(db_path: &Path, env: HiveEnv, build_slots: usize) -> Result<Self> {
|
||||
let HiveEnv {
|
||||
hyperhive_flake,
|
||||
claude_code_path,
|
||||
|
|
@ -386,7 +378,6 @@ impl Coordinator {
|
|||
agent_memory_max,
|
||||
agent_cpu_weight,
|
||||
agent_io_weight,
|
||||
model_prices,
|
||||
agents: Mutex::new(HashMap::new()),
|
||||
recent_transient: Mutex::new(HashMap::new()),
|
||||
graceful_stop_pending: Mutex::new(HashSet::new()),
|
||||
|
|
@ -1072,13 +1063,7 @@ mod open_tests {
|
|||
use super::{Coordinator, HiveEnv};
|
||||
|
||||
fn open(db: &std::path::Path) -> Coordinator {
|
||||
Coordinator::open(
|
||||
db,
|
||||
HiveEnv::default(),
|
||||
crate::hive_stats::PriceTable::new(),
|
||||
1,
|
||||
)
|
||||
.expect("open coordinator")
|
||||
Coordinator::open(db, HiveEnv::default(), 1).expect("open coordinator")
|
||||
}
|
||||
|
||||
/// A build log row the previous process left open is closed as failed
|
||||
|
|
|
|||
|
|
@ -1,45 +1,15 @@
|
|||
//! Remaining single-endpoint dashboard handlers: mark-all-read,
|
||||
//! spawn-request and hive-wide turn stats.
|
||||
//! Remaining single-endpoint dashboard handlers: mark-all-read and
|
||||
//! spawn-request.
|
||||
|
||||
use axum::{
|
||||
extract::{Path as AxumPath, State},
|
||||
http::StatusCode,
|
||||
response::{IntoResponse, Response},
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use utoipa::{IntoParams, ToSchema};
|
||||
use serde::Serialize;
|
||||
use utoipa::ToSchema;
|
||||
|
||||
use super::{AppState, Ident, error_response};
|
||||
use crate::hive_stats::HiveStats;
|
||||
|
||||
#[derive(Deserialize, IntoParams)]
|
||||
pub(super) struct StatsHiveQuery {
|
||||
/// Stats window; defaults to `24h`.
|
||||
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).
|
||||
#[utoipa::path(
|
||||
get,
|
||||
path = "/api/stats-hive",
|
||||
params(StatsHiveQuery),
|
||||
responses((status = 200, description = "hive-wide turn-stats rollup", body = HiveStats)),
|
||||
tag = "misc_api"
|
||||
)]
|
||||
pub(super) async fn api_stats_hive(
|
||||
State(state): State<AppState>,
|
||||
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,
|
||||
&state.coord.model_prices,
|
||||
))
|
||||
.into_response()
|
||||
}
|
||||
|
||||
#[derive(Serialize, ToSchema)]
|
||||
pub(super) struct MarkAllReadBody {
|
||||
|
|
|
|||
|
|
@ -103,7 +103,6 @@ fn app(coord: Arc<Coordinator>) -> axum::Router {
|
|||
.routes(routes!(health::get_health_live))
|
||||
.routes(routes!(health::get_health_ready))
|
||||
.routes(routes!(state_snapshot::api_state))
|
||||
.routes(routes!(misc_api::api_stats_hive))
|
||||
.routes(routes!(misc_api::post_mark_all_read))
|
||||
.routes(routes!(build_logs::get_build_logs_all))
|
||||
.routes(routes!(build_logs::get_build_log_for_node))
|
||||
|
|
@ -264,7 +263,6 @@ mod tests {
|
|||
let coord = Coordinator::open(
|
||||
&dir.path().join("broker.sqlite"),
|
||||
crate::coordinator::HiveEnv::default(),
|
||||
crate::hive_stats::PriceTable::new(),
|
||||
1,
|
||||
)
|
||||
.expect("open coordinator");
|
||||
|
|
@ -352,7 +350,6 @@ mod router_build_probe {
|
|||
.routes(routes!(health::get_health_live))
|
||||
.routes(routes!(health::get_health_ready))
|
||||
.routes(routes!(state_snapshot::api_state))
|
||||
.routes(routes!(misc_api::api_stats_hive))
|
||||
.routes(routes!(misc_api::post_mark_all_read))
|
||||
.routes(routes!(build_logs::get_build_logs_all))
|
||||
.routes(routes!(build_logs::get_build_log_for_node))
|
||||
|
|
|
|||
|
|
@ -280,13 +280,8 @@ fn closure(cached: bool) -> AgentClosure {
|
|||
|
||||
fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let coord = Coordinator::open(
|
||||
&dir.path().join("broker.sqlite"),
|
||||
HiveEnv::default(),
|
||||
crate::hive_stats::PriceTable::new(),
|
||||
1,
|
||||
)
|
||||
.expect("open coordinator");
|
||||
let coord = Coordinator::open(&dir.path().join("broker.sqlite"), HiveEnv::default(), 1)
|
||||
.expect("open coordinator");
|
||||
(dir, Arc::new(coord))
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -301,9 +301,9 @@ async fn systemd_daemon_reload() -> Result<()> {
|
|||
/// `harness` and `work` are absent too: they hold that agent's own runtime
|
||||
/// material (`bash-tasks/`, turn-stats and event sqlite dbs) and its, often
|
||||
/// very large, worktrees and build trees, none of which the holder has a stated
|
||||
/// reason to read, let alone write. hive-c0re still reads `harness` directly on
|
||||
/// the host (`stats::hive_stats`), which needs no bind mount into anyone else's
|
||||
/// container.
|
||||
/// reason to read, let alone write. hive-c0re itself reaches `harness` directly
|
||||
/// on the host (the paused marker), which needs no bind mount into anyone
|
||||
/// else's container.
|
||||
fn bind_child_agent_dirs(child: &str, binds: &mut Vec<BindMount>) {
|
||||
let Ok(child) = hive_types::Ident::parse(child) else {
|
||||
tracing::warn!(%child, "skipping child bind: invalid agent name");
|
||||
|
|
|
|||
|
|
@ -45,7 +45,7 @@ mod test_env;
|
|||
mod workers;
|
||||
|
||||
pub(crate) use agent_config::{archetypes, limits, permissions, resource_limits, topology};
|
||||
pub(crate) use stats::{container_stats, hive_stats, host_stats, otel_metrics, warnings};
|
||||
pub(crate) use stats::{container_stats, host_stats, otel_metrics, warnings};
|
||||
pub(crate) use stores::{broker, build_logs, db, power};
|
||||
pub(crate) use workers::{agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets};
|
||||
|
||||
|
|
@ -69,7 +69,7 @@ enum Cmd {
|
|||
/// Path to a JSON config file holding the host-level daemon config
|
||||
/// (the [`ServeConfig`] shape:
|
||||
/// the container-injected `HiveEnv` fields + the hive-c0re-local
|
||||
/// `model_prices` table). Used as the base; any per-flag override
|
||||
/// `build_slots`). 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
|
||||
|
|
@ -106,14 +106,6 @@ enum Cmd {
|
|||
/// `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 closure-fetching job-queue nodes
|
||||
/// (prebuild / profile-swap / create / meta lock). Set via the
|
||||
/// `services.hyperhive.deploy.hive-controller.buildSlots` NixOS option.
|
||||
|
|
@ -138,7 +130,6 @@ async fn main() -> Result<()> {
|
|||
agent_memory_max,
|
||||
agent_cpu_weight,
|
||||
agent_io_weight,
|
||||
model_prices,
|
||||
build_slots,
|
||||
} => {
|
||||
// Base config from the --config file (or the built-in
|
||||
|
|
@ -175,10 +166,6 @@ async fn main() -> Result<()> {
|
|||
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;
|
||||
}
|
||||
|
|
@ -194,7 +181,7 @@ async fn main() -> Result<()> {
|
|||
// The forge integration is mandatory (the NixOS module sets both
|
||||
// vars unconditionally), so check them before any worker starts.
|
||||
forge::validate_env()?;
|
||||
cmd_serve(sc.env, sc.model_prices, sc.build_slots, db, &cli.socket).await
|
||||
cmd_serve(sc.env, sc.build_slots, db, &cli.socket).await
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -225,7 +212,6 @@ fn submit_periodic_sweep(
|
|||
)]
|
||||
async fn cmd_serve(
|
||||
env: HiveEnv,
|
||||
model_prices: crate::hive_stats::PriceTable,
|
||||
build_slots: usize,
|
||||
db: std::path::PathBuf,
|
||||
socket: &std::path::Path,
|
||||
|
|
@ -238,7 +224,7 @@ async fn cmd_serve(
|
|||
// `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)?);
|
||||
let coord = Arc::new(Coordinator::open(&db, env, build_slots)?);
|
||||
// Idempotent pre-flight: rewrite pre-meta-layout applied
|
||||
// repos, refresh the files that follow the agent set.
|
||||
// Runs before the boot reconcile.
|
||||
|
|
@ -340,7 +326,7 @@ async fn cmd_serve(
|
|||
// 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
|
||||
// ~hundreds of KB, and the agent's own /stats page reads it
|
||||
// directly; pruning it would just lose trend history for no gain.)
|
||||
// Slow per-container disk sampler: hive-priv measures each agent's state/
|
||||
// + work/ + writable rootfs every 30 min, cached so a sample only reads
|
||||
|
|
|
|||
|
|
@ -794,7 +794,6 @@ mod tests {
|
|||
let coord = Coordinator::open(
|
||||
&dir.path().join("broker.sqlite"),
|
||||
crate::coordinator::HiveEnv::default(),
|
||||
crate::hive_stats::PriceTable::new(),
|
||||
1,
|
||||
)
|
||||
.expect("open coordinator");
|
||||
|
|
|
|||
|
|
@ -577,7 +577,6 @@ mod tests {
|
|||
let coord = Coordinator::open(
|
||||
&dir.path().join("broker.sqlite"),
|
||||
crate::coordinator::HiveEnv::default(),
|
||||
crate::hive_stats::PriceTable::new(),
|
||||
1,
|
||||
)
|
||||
.expect("open coordinator");
|
||||
|
|
|
|||
|
|
@ -1,514 +0,0 @@
|
|||
//! 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-agent`'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` can read them fine, but cannot write/delete (which is why
|
||||
//! retention sweeps run agent-side in the harness, not here). 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;
|
||||
|
||||
use rusqlite::{Connection, OpenFlags};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use utoipa::ToSchema;
|
||||
|
||||
use crate::coordinator::Coordinator;
|
||||
use chrono::Utc;
|
||||
|
||||
/// 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,
|
||||
/// All available data: aggregate over every recorded turn since the
|
||||
/// earliest (`from == 0`), with no fixed lookback.
|
||||
All,
|
||||
}
|
||||
|
||||
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,
|
||||
"all" => Self::All,
|
||||
_ => 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",
|
||||
Self::All => "all",
|
||||
}
|
||||
}
|
||||
|
||||
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,
|
||||
// `All` has no fixed lookback; `hive_snapshot` uses `from == 0`
|
||||
// (every recorded turn). 0 here is a consistent safe fallback.
|
||||
Self::All => 0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Approximate USD price per **million** tokens for one model class.
|
||||
/// The four fields map to the four token kinds the turn-stats schema
|
||||
/// records. This is a deliberately rough estimate — Anthropic list
|
||||
/// pricing drifts, so the dashboard labels the figure as an estimate.
|
||||
///
|
||||
/// Operators keep the table current without a code change via the
|
||||
/// `services.hyperhive.modelPrices` nix option (passed to
|
||||
/// `hive-c0re serve --model-prices <json>`, held on the `Coordinator`
|
||||
/// as a [`PriceTable`]). Any model not covered by the operator table
|
||||
/// falls back to [`builtin_prices`].
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
|
||||
pub struct Prices {
|
||||
pub input: f64,
|
||||
pub output: f64,
|
||||
pub cache_read: f64,
|
||||
pub cache_write: f64,
|
||||
}
|
||||
|
||||
/// Operator-tunable model→price map. Keys are matched
|
||||
/// case-insensitively as a substring of the model id (e.g. `"sonnet"`
|
||||
/// matches `"claude-sonnet-4-5"`); the longest matching key wins so a
|
||||
/// specific entry beats a generic family name. An empty map means
|
||||
/// every model uses [`builtin_prices`].
|
||||
pub type PriceTable = HashMap<String, Prices>;
|
||||
|
||||
/// Built-in fallback pricing, per **million** tokens. Used when the
|
||||
/// operator's [`PriceTable`] has no key matching the model. Figures are
|
||||
/// current Anthropic list prices for the Claude 4.x family (Opus 4.x,
|
||||
/// Sonnet 4.x, Haiku 4.5); `cache_write` uses the 1-hour cache-TTL price,
|
||||
/// which is the default through the Claude subscription the agents run
|
||||
/// on. Keep in sync with the `services.hyperhive.modelPrices` nix default
|
||||
/// (`nix/host-modules/hive-c0re/options.nix`).
|
||||
fn builtin_prices(model: &str) -> Prices {
|
||||
let m = model.to_ascii_lowercase();
|
||||
if m.contains("opus") {
|
||||
Prices {
|
||||
input: 5.0,
|
||||
output: 25.0,
|
||||
cache_read: 0.5,
|
||||
cache_write: 10.0,
|
||||
}
|
||||
} else if m.contains("haiku") {
|
||||
Prices {
|
||||
input: 1.0,
|
||||
output: 5.0,
|
||||
cache_read: 0.1,
|
||||
cache_write: 2.0,
|
||||
}
|
||||
} else {
|
||||
// sonnet + unknown fallback
|
||||
Prices {
|
||||
input: 3.0,
|
||||
output: 15.0,
|
||||
cache_read: 0.3,
|
||||
cache_write: 6.0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve the price for `model`: prefer the operator-provided
|
||||
/// `table` (longest matching key wins, so a specific
|
||||
/// `claude-3-5-sonnet` entry beats a generic `sonnet`), else fall back
|
||||
/// to [`builtin_prices`].
|
||||
fn resolve_prices(model: &str, table: &PriceTable) -> Prices {
|
||||
let m = model.to_ascii_lowercase();
|
||||
let mut best: Option<(usize, Prices)> = None;
|
||||
for (k, v) in table {
|
||||
let key = k.to_ascii_lowercase();
|
||||
if key.is_empty() || !m.contains(&key) {
|
||||
continue;
|
||||
}
|
||||
let better = match best {
|
||||
Some((blen, _)) => key.len() > blen,
|
||||
None => true,
|
||||
};
|
||||
if better {
|
||||
best = Some((key.len(), *v));
|
||||
}
|
||||
}
|
||||
match best {
|
||||
Some((_, p)) => p,
|
||||
None => builtin_prices(&m),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, ToSchema)]
|
||||
pub struct KeyCount {
|
||||
pub key: String,
|
||||
pub count: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, ToSchema)]
|
||||
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, ToSchema)]
|
||||
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>,
|
||||
/// Most-run normalised bash-command heads ("favorite tools") across
|
||||
/// the whole swarm, busiest first, capped to 10. Empty until the
|
||||
/// hive-bash-daemon capture has recorded `bash_commands` rows on at
|
||||
/// least one active agent.
|
||||
pub bash_mix: Vec<KeyCount>,
|
||||
/// Most-triggered skills (fully-qualified `plugin:skill-name`) across
|
||||
/// the whole swarm, busiest first, capped to 10. Sourced from each
|
||||
/// agent's `tool_call_breakdown_json`. Empty until at least one agent has
|
||||
/// actually invoked a skill.
|
||||
pub skill_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>,
|
||||
/// Normalised bash-command head → invocation count, from the agent's
|
||||
/// `bash_commands` table (written by hive-bash-daemon). Empty when that
|
||||
/// capture hasn't run for this agent (table absent).
|
||||
bash: HashMap<String, u64>,
|
||||
/// Skill (fully-qualified `plugin:skill-name`) → invocation count,
|
||||
/// unpacked from `turn_stats.tool_call_breakdown_json`'s `Skill:*`
|
||||
/// entries. Empty when no skill was invoked in the window.
|
||||
skills: HashMap<String, u64>,
|
||||
}
|
||||
|
||||
fn u64_from_i64(v: i64) -> u64 {
|
||||
u64::try_from(v).unwrap_or(0)
|
||||
}
|
||||
|
||||
/// 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, prices: &PriceTable) -> rusqlite::Result<AgentAgg> {
|
||||
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,
|
||||
tool_call_breakdown_json
|
||||
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)?),
|
||||
row.get::<_, Option<String>>(5)?,
|
||||
))
|
||||
})?;
|
||||
for r in rows {
|
||||
let (model, input, output, cache_read, cache_creation, breakdown_json) = 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 = resolve_prices(&model, prices);
|
||||
#[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;
|
||||
if let Some(json) = breakdown_json {
|
||||
for (skill, count) in skills_from_breakdown(&json) {
|
||||
*agg.skills.entry(skill).or_insert(0) += count;
|
||||
}
|
||||
}
|
||||
}
|
||||
agg.bash = read_bash_heads(&conn, from);
|
||||
Ok(agg)
|
||||
}
|
||||
|
||||
/// Pull the `Skill:<skill>` entries out of one turn's
|
||||
/// `tool_call_breakdown_json` blob (see `hive_agent::events::observe_stream`
|
||||
/// for how they're written), stripping the `Skill:` marker so the rollup
|
||||
/// keys by the bare (fully-qualified `plugin:skill-name`) skill identifier.
|
||||
/// Malformed/absent JSON yields nothing rather than erroring — one bad row
|
||||
/// shouldn't drop a whole agent from the rollup.
|
||||
fn skills_from_breakdown(json: &str) -> Vec<(String, u64)> {
|
||||
let Ok(map) = serde_json::from_str::<HashMap<String, u64>>(json) else {
|
||||
return Vec::new();
|
||||
};
|
||||
map.into_iter()
|
||||
.filter_map(|(k, v)| k.strip_prefix("Skill:").map(|s| (s.to_owned(), v)))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Tally normalised bash-command heads from the agent's `bash_commands`
|
||||
/// table (`ts INTEGER, head TEXT`, written by hive-bash-daemon) over
|
||||
/// `[from, now]`. Best-effort + isolated from `read_agent`'s error path:
|
||||
/// a missing table (capture hasn't run for this agent) or any read error
|
||||
/// yields an empty map rather than propagating, so the favorite-tools
|
||||
/// rollup simply omits that agent and never fails the whole endpoint.
|
||||
/// `ts` is unix seconds, matching the `from` cutoff.
|
||||
fn read_bash_heads(conn: &Connection, from: i64) -> HashMap<String, u64> {
|
||||
let mut out: HashMap<String, u64> = HashMap::new();
|
||||
let Ok(mut stmt) =
|
||||
conn.prepare("SELECT head, COUNT(*) FROM bash_commands WHERE ts >= ?1 GROUP BY head")
|
||||
else {
|
||||
return out;
|
||||
};
|
||||
let Ok(rows) = stmt.query_map([from], |row| {
|
||||
Ok((
|
||||
row.get::<_, String>(0)?,
|
||||
u64_from_i64(row.get::<_, i64>(1)?),
|
||||
))
|
||||
}) else {
|
||||
return out;
|
||||
};
|
||||
for (head, count) in rows.flatten() {
|
||||
*out.entry(head).or_insert(0) += count;
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// 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, prices: &PriceTable) -> HiveStats {
|
||||
let now = Utc::now().timestamp();
|
||||
// Fixed windows look back a constant span; `all` aggregates every
|
||||
// recorded turn (`from == 0`). The hive rollup isn't time-bucketed, so
|
||||
// unlike the per-agent snapshot it needs no adaptive bucket sizing.
|
||||
let from = match window {
|
||||
Window::All => 0,
|
||||
_ => now - window.span_secs(),
|
||||
};
|
||||
|
||||
let mut agents: Vec<AgentRollup> = Vec::new();
|
||||
let mut model_mix: HashMap<String, u64> = HashMap::new();
|
||||
let mut bash_mix: HashMap<String, u64> = HashMap::new();
|
||||
let mut skill_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, prices) {
|
||||
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;
|
||||
}
|
||||
for (h, c) in &agg.bash {
|
||||
*bash_mix.entry(h.clone()).or_insert(0) += c;
|
||||
}
|
||||
for (s, c) in &agg.skills {
|
||||
*skill_mix.entry(s.clone()).or_insert(0) += c;
|
||||
}
|
||||
agents.push(AgentRollup {
|
||||
name: name.into_string(),
|
||||
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)));
|
||||
|
||||
// Busiest commands first, capped to a top-10 "favorite tools" list.
|
||||
let mut bash_mix: Vec<KeyCount> = bash_mix
|
||||
.into_iter()
|
||||
.map(|(key, count)| KeyCount { key, count })
|
||||
.collect();
|
||||
bash_mix.sort_by(|a, b| b.count.cmp(&a.count).then_with(|| a.key.cmp(&b.key)));
|
||||
bash_mix.truncate(10);
|
||||
|
||||
// Busiest skills first, capped to a top-10 list — same shape as bash_mix.
|
||||
let mut skill_mix: Vec<KeyCount> = skill_mix
|
||||
.into_iter()
|
||||
.map(|(key, count)| KeyCount { key, count })
|
||||
.collect();
|
||||
skill_mix.sort_by(|a, b| b.count.cmp(&a.count).then_with(|| a.key.cmp(&b.key)));
|
||||
skill_mix.truncate(10);
|
||||
|
||||
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,
|
||||
bash_mix,
|
||||
skill_mix,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// `read_bash_heads` tallies per-head counts over the window and is
|
||||
/// isolated from a missing table (returns empty, never errors).
|
||||
#[test]
|
||||
fn bash_heads_tally_and_window() {
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);")
|
||||
.unwrap();
|
||||
let now = Utc::now().timestamp();
|
||||
for (ts, head) in [
|
||||
(now - 100, "cargo"),
|
||||
(now - 200, "cargo"),
|
||||
(now - 300, "git"),
|
||||
(now - 10_000, "rg"), // outside a 1h-ish cutoff below
|
||||
] {
|
||||
conn.execute(
|
||||
"INSERT INTO bash_commands (ts, head) VALUES (?1, ?2)",
|
||||
rusqlite::params![ts, head],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
let heads = read_bash_heads(&conn, now - 3600);
|
||||
assert_eq!(heads.get("cargo").copied(), Some(2));
|
||||
assert_eq!(heads.get("git").copied(), Some(1));
|
||||
assert_eq!(heads.get("rg").copied(), None); // window cutoff excludes it
|
||||
}
|
||||
|
||||
/// A missing `bash_commands` table degrades to an empty tally rather
|
||||
/// than erroring — the pre-capture window on a fresh agent.
|
||||
#[test]
|
||||
fn bash_heads_missing_table_is_empty() {
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
assert!(read_bash_heads(&conn, 0).is_empty());
|
||||
}
|
||||
|
||||
/// `Skill:<skill>` entries are unpacked and the marker stripped; a
|
||||
/// non-skill tool entry in the same blob is ignored.
|
||||
#[test]
|
||||
fn skills_from_breakdown_strips_marker_and_ignores_other_tools() {
|
||||
let json = r#"{"Skill:base:async-task-hygiene":3,"Read":7}"#;
|
||||
let mut got = skills_from_breakdown(json);
|
||||
got.sort();
|
||||
assert_eq!(got, vec![("base:async-task-hygiene".to_owned(), 3)]);
|
||||
}
|
||||
|
||||
/// Malformed JSON degrades to empty rather than erroring — one bad row
|
||||
/// shouldn't drop a whole agent from the rollup.
|
||||
#[test]
|
||||
fn skills_from_breakdown_malformed_json_is_empty() {
|
||||
assert!(skills_from_breakdown("not json").is_empty());
|
||||
}
|
||||
|
||||
/// No `Skill:*` keys at all (an agent that hasn't triggered one yet)
|
||||
/// is also empty, not an error.
|
||||
#[test]
|
||||
fn skills_from_breakdown_no_skills_is_empty() {
|
||||
assert!(skills_from_breakdown(r#"{"Read":1,"Edit":2}"#).is_empty());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,10 +1,9 @@
|
|||
//! Metrics aggregation for the dashboard: hive-wide turn-stats
|
||||
//! rollups, host-system probes / server warnings, and live
|
||||
//! per-container cgroup load. Each submodule is re-exported at the
|
||||
//! crate root, so `crate::hive_stats::…` etc. keep working unchanged.
|
||||
//! Metrics aggregation for the dashboard: host-system probes / server
|
||||
//! warnings, and live per-container cgroup load. Each submodule is
|
||||
//! re-exported at the crate root, so `crate::host_stats::…` etc. keep
|
||||
//! working unchanged.
|
||||
|
||||
pub mod container_stats;
|
||||
pub mod hive_stats;
|
||||
pub mod host_stats;
|
||||
pub mod otel_metrics;
|
||||
pub mod sweep_health;
|
||||
|
|
|
|||
Loading…
Reference in a new issue