35611e9f0d
Adds a 'sessions' chip to the per-agent stats summary panel showing how many fresh claude sessions started within the selected time window. Backend (hive-agent/src/stats.rs): - New optional field `session_count: Option<u64>` on `Snapshot` (skip_serializing_if = None — inert-until-data, same pattern as first_turn_ctx). Counts rows in the `sessions` table whose started_at falls within the window; returns None when the table doesn't exist on an older db. - New `read_session_count(conn, from)` helper (rusqlite::Result so the caller maps Err to None). - Extracted per-row accumulation loop into `TurnAccum` struct + `push()` method to keep `snapshot()` under the too_many_lines limit. Frontend (frontend/packages/agent/src/stats.js): - New 'sessions' chip added to renderSummary, guarded by `typeof s.session_count === 'number'`, placed before the existing first-turn-ctx chip.
816 lines
30 KiB
Rust
816 lines
30 KiB
Rust
//! Read-side aggregations over the per-agent `turn_stats.sqlite` for
|
|
//! the agent's `/stats` web page. Owned by the agent (same process
|
|
//! that writes the sink) so per-MCP extensions can register more
|
|
//! providers without the host needing to know their schemas.
|
|
//!
|
|
//! Best-effort: any sqlite error returns an empty snapshot rather than
|
|
//! propagating — the stats page is decorative, not authoritative, and
|
|
//! a missing db on a brand-new agent shouldn't 500 the route.
|
|
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::path::{Path, PathBuf};
|
|
|
|
use anyhow::{Context, Result};
|
|
use rusqlite::{Connection, OpenFlags};
|
|
use serde::Serialize;
|
|
|
|
use hive_sh4re::ReminderStats;
|
|
use hive_sh4re::wire_time::now_unix;
|
|
|
|
/// Window param accepted by `/api/stats?window=`. Each maps to a
|
|
/// total span + the bucket width used to roll up trend series.
|
|
#[derive(Debug, Clone, Copy)]
|
|
pub enum Window {
|
|
Hour,
|
|
FourHour,
|
|
Day,
|
|
ThreeDay,
|
|
Week,
|
|
Month,
|
|
/// All available data: the range starts at the earliest recorded turn
|
|
/// (`MIN(started_at)`) rather than a fixed lookback, with an adaptive
|
|
/// bucket width so the trend series stays bounded at any span.
|
|
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,
|
|
// Default (incl. `label()`'s own canonical `"24h"`/`"1d"`).
|
|
_ => 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",
|
|
}
|
|
}
|
|
|
|
#[must_use]
|
|
pub 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 — its range is computed from
|
|
// `MIN(started_at)` in `snapshot()`. 0 is only a safe fallback
|
|
// (→ `from == now` → empty range) if ever reached generically.
|
|
Self::All => 0,
|
|
}
|
|
}
|
|
|
|
fn bucket_secs(self) -> i64 {
|
|
match self {
|
|
// 5-min buckets for 1h (12 buckets), 15-min for 4h (16 buckets),
|
|
// hourly for 24h + 3d, daily for 7d + 30d. `All` shares the daily
|
|
// fallback, but its real bucket width is chosen adaptively in
|
|
// `snapshot()` via `adaptive_bucket_secs` — this arm is only
|
|
// reached defensively.
|
|
Self::Hour => 300,
|
|
Self::FourHour => 900,
|
|
Self::Day | Self::ThreeDay => 3600,
|
|
Self::Week | Self::Month | Self::All => 24 * 3600,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Bucket width for the unbounded `all` window, laddered by the actual
|
|
/// span so the trend series stays bounded (≈ ≤100 buckets) at any range:
|
|
/// hourly ≤ 2d, daily ≤ 90d, weekly ≤ 2y, ~monthly (30d) beyond.
|
|
fn adaptive_bucket_secs(span_secs: i64) -> i64 {
|
|
const HOUR: i64 = 3600;
|
|
const DAY: i64 = 24 * HOUR;
|
|
match span_secs {
|
|
s if s <= 2 * DAY => HOUR,
|
|
s if s <= 90 * DAY => DAY,
|
|
s if s <= 730 * DAY => 7 * DAY,
|
|
_ => 30 * DAY,
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Serialize)]
|
|
pub struct Snapshot {
|
|
pub window: &'static str,
|
|
pub bucket_seconds: i64,
|
|
pub now: i64,
|
|
pub from: i64,
|
|
/// Total turns in the window.
|
|
pub turn_count: u64,
|
|
/// Time-bucketed trend series, oldest first. Always covers the
|
|
/// full window even for empty buckets (so charts paint a stable
|
|
/// x-axis instead of skipping gaps).
|
|
pub buckets: Vec<Bucket>,
|
|
/// Top tools by call count across the window. Capped to 10.
|
|
pub tool_breakdown: Vec<KeyCount>,
|
|
/// Top shell commands ("favorite tools") by invocation count across
|
|
/// the window, capped to 10. Normalised command heads recorded per
|
|
/// bash task into the `bash_commands` table by hive-bash-mcp. Empty
|
|
/// until that capture lands (or on any agent that hasn't run a bash
|
|
/// task) — the table is created lazily by the writer, so a read
|
|
/// before the first insert returns an empty list, not an error.
|
|
pub bash_breakdown: Vec<KeyCount>,
|
|
pub wake_mix: Vec<KeyCount>,
|
|
pub result_mix: Vec<KeyCount>,
|
|
/// Distinct models seen in the window, sorted. Each bucket's
|
|
/// `model_counts` keys into this set; the stats page uses it as
|
|
/// the stacked-bar series list (stable order + colours).
|
|
pub models: Vec<String>,
|
|
/// Across-window p50 / p95 / avg of `duration_ms`. Same numbers
|
|
/// as the per-bucket fields but aggregated over the whole window
|
|
/// for the headline summary chips.
|
|
pub duration_summary: DurationSummary,
|
|
/// Reminder activity stats: counts of scheduled, delivered, and
|
|
/// pending reminders over the window (fetched from the broker RPC).
|
|
/// None if the RPC call failed or hasn't been integrated yet.
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub reminder_stats: Option<ReminderStats>,
|
|
/// First-turn input tokens of the most recent fresh claude session
|
|
/// that started in the window — a proxy for system-prompt + CLAUDE.md
|
|
/// sprawl (a fresh session's first turn pays the full static prefix
|
|
/// uncached, so this is the current "cold context" cost). `None` until
|
|
/// the sessions capture (per-session `session_id`) has data; every
|
|
/// pre-capture `turn_stats` row has a NULL `session_id` and is excluded.
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub first_turn_ctx: Option<u64>,
|
|
/// Number of fresh claude sessions (i.e. rows in the `sessions` table)
|
|
/// that started within the window. A new session is minted whenever the
|
|
/// harness runs a turn without `--continue` (manual `/new-session`,
|
|
/// auto-compaction fallback, or first-ever turn). `None` when the
|
|
/// `sessions` table doesn't exist yet (older db) — same inert-until-data
|
|
/// pattern as `first_turn_ctx`.
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub session_count: Option<u64>,
|
|
}
|
|
|
|
#[derive(Debug, Serialize)]
|
|
pub struct Bucket {
|
|
/// Unix timestamp of the bucket start.
|
|
pub ts: i64,
|
|
pub turn_count: u64,
|
|
pub avg_duration_ms: f64,
|
|
pub p50_duration_ms: f64,
|
|
pub p95_duration_ms: f64,
|
|
/// Sums across the bucket. JS picks how to combine them
|
|
/// (input + output for cost, etc.) so we don't bake a policy in.
|
|
pub input_tokens: u64,
|
|
pub output_tokens: u64,
|
|
pub cache_read_input_tokens: u64,
|
|
pub cache_creation_input_tokens: u64,
|
|
/// Mean of `last_input_tokens` across the bucket (the context
|
|
/// size at turn-end — useful for spotting drift toward compaction).
|
|
pub avg_ctx_tokens: f64,
|
|
pub max_ctx_tokens: u64,
|
|
/// Turn count per model in this bucket. Model choice greatly
|
|
/// affects token cost, so this lets the operator line model usage
|
|
/// up against the cost series over time.
|
|
pub model_counts: HashMap<String, u64>,
|
|
/// Turn count per `result_kind` in this bucket. Lets the stats
|
|
/// page chart error / rate-limit / compaction outcomes *over time*
|
|
/// (the window-total lives in `Snapshot::result_mix`).
|
|
pub result_counts: HashMap<String, u64>,
|
|
}
|
|
|
|
#[derive(Debug, Serialize)]
|
|
pub struct KeyCount {
|
|
pub key: String,
|
|
pub count: u64,
|
|
}
|
|
|
|
// Field names drop the `_ms` unit suffix (satisfies `clippy::struct_field_names`
|
|
// once this crate is a bin — pub structs lose the lib API-name exemption), but
|
|
// the serialized keys keep `_ms` via `serde(rename)` so the `/api/stats` JSON
|
|
// contract the agent web UI reads (`frontend/packages/agent/src/stats.js`) is
|
|
// unchanged.
|
|
#[derive(Debug, Default, Serialize)]
|
|
pub struct DurationSummary {
|
|
#[serde(rename = "avg_ms")]
|
|
pub avg: f64,
|
|
#[serde(rename = "p50_ms")]
|
|
pub p50: f64,
|
|
#[serde(rename = "p95_ms")]
|
|
pub p95: f64,
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn snapshot_default(window: Window) -> Snapshot {
|
|
let path = default_path();
|
|
match snapshot(&path, window) {
|
|
Ok(s) => s,
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, path = %path.display(), "stats: snapshot failed");
|
|
empty_snapshot(window)
|
|
}
|
|
}
|
|
}
|
|
|
|
fn default_path() -> PathBuf {
|
|
crate::paths::harness_dir().join("hyperhive-turn-stats.sqlite")
|
|
}
|
|
|
|
fn empty_snapshot(window: Window) -> Snapshot {
|
|
let now = now_unix();
|
|
let from = now - window.span_secs();
|
|
let buckets = fill_buckets(from, now, window.bucket_secs(), &HashMap::new());
|
|
Snapshot {
|
|
window: window.label(),
|
|
bucket_seconds: window.bucket_secs(),
|
|
now,
|
|
from,
|
|
turn_count: 0,
|
|
buckets,
|
|
tool_breakdown: Vec::new(),
|
|
bash_breakdown: Vec::new(),
|
|
wake_mix: Vec::new(),
|
|
result_mix: Vec::new(),
|
|
models: Vec::new(),
|
|
duration_summary: DurationSummary::default(),
|
|
reminder_stats: None,
|
|
first_turn_ctx: None,
|
|
session_count: None,
|
|
}
|
|
}
|
|
|
|
/// Accumulated totals across all rows fetched for one stats window.
|
|
/// Built by iterating `turn_stats` rows; passed to `fill_buckets` /
|
|
/// `summarize_durations` / `top_n` to produce the final `Snapshot`.
|
|
#[derive(Default)]
|
|
struct TurnAccum {
|
|
by_bucket: HashMap<i64, BucketAcc>,
|
|
tool_totals: HashMap<String, u64>,
|
|
wake_totals: HashMap<String, u64>,
|
|
result_totals: HashMap<String, u64>,
|
|
model_set: HashSet<String>,
|
|
all_durations: Vec<i64>,
|
|
turn_count: u64,
|
|
}
|
|
|
|
impl TurnAccum {
|
|
fn push(&mut self, r: Row, bucket_secs: i64) {
|
|
self.turn_count += 1;
|
|
let bucket_ts = (r.started_at / bucket_secs) * bucket_secs;
|
|
let b = self.by_bucket.entry(bucket_ts).or_default();
|
|
b.turn_count += 1;
|
|
b.durations.push(r.duration_ms.max(0));
|
|
b.input_tokens = b.input_tokens.saturating_add(r.input_tokens);
|
|
b.output_tokens = b.output_tokens.saturating_add(r.output_tokens);
|
|
b.cache_read_input_tokens = b
|
|
.cache_read_input_tokens
|
|
.saturating_add(r.cache_read_input_tokens);
|
|
b.cache_creation_input_tokens = b
|
|
.cache_creation_input_tokens
|
|
.saturating_add(r.cache_creation_input_tokens);
|
|
b.ctx_sum = b.ctx_sum.saturating_add(r.last_input_tokens);
|
|
b.ctx_max = b.ctx_max.max(r.last_input_tokens);
|
|
*b.model_counts.entry(r.model.clone()).or_insert(0) += 1;
|
|
*b.result_counts.entry(r.result_kind.clone()).or_insert(0) += 1;
|
|
|
|
self.all_durations.push(r.duration_ms.max(0));
|
|
*self.wake_totals.entry(r.wake_from).or_insert(0) += 1;
|
|
*self.result_totals.entry(r.result_kind).or_insert(0) += 1;
|
|
self.model_set.insert(r.model);
|
|
|
|
if let Some(json) = r.tool_breakdown_json
|
|
&& let Ok(map) = serde_json::from_str::<HashMap<String, u64>>(&json)
|
|
{
|
|
for (k, v) in map {
|
|
*self.tool_totals.entry(k).or_insert(0) += v;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn snapshot(path: &Path, window: Window) -> Result<Snapshot> {
|
|
// Read-only open so we can't corrupt the db via a query bug.
|
|
let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)
|
|
.with_context(|| format!("open {} read-only", path.display()))?;
|
|
// turn_stats is rollback-journal (not WAL): a read landing while the
|
|
// harness's own sink is mid-INSERT gets SQLITE_BUSY, which propagates up
|
|
// and blanks the whole stats page. Wait out the brief write instead.
|
|
conn.busy_timeout(std::time::Duration::from_millis(500))
|
|
.with_context(|| format!("set busy_timeout on {}", path.display()))?;
|
|
let now = now_unix();
|
|
// Fixed windows look back a constant span; `all` starts at the earliest
|
|
// recorded turn and sizes its buckets adaptively from that span.
|
|
let (from, bucket_secs) = match window {
|
|
Window::All => {
|
|
let min_ts: Option<i64> =
|
|
conn.query_row("SELECT MIN(started_at) FROM turn_stats", [], |row| {
|
|
row.get(0)
|
|
})?;
|
|
let from = min_ts.unwrap_or(now);
|
|
(from, adaptive_bucket_secs(now - from))
|
|
}
|
|
_ => (now - window.span_secs(), window.bucket_secs()),
|
|
};
|
|
|
|
let mut stmt = conn.prepare(
|
|
"SELECT started_at, duration_ms,
|
|
input_tokens, output_tokens,
|
|
cache_read_input_tokens, cache_creation_input_tokens,
|
|
last_input_tokens,
|
|
tool_call_breakdown_json,
|
|
wake_from, result_kind, model
|
|
FROM turn_stats
|
|
WHERE started_at >= ?1
|
|
ORDER BY started_at ASC",
|
|
)?;
|
|
let rows = stmt.query_map([from], |row| {
|
|
Ok(Row {
|
|
started_at: row.get(0)?,
|
|
duration_ms: row.get::<_, i64>(1)?,
|
|
input_tokens: u64_from_i64(row.get::<_, i64>(2)?),
|
|
output_tokens: u64_from_i64(row.get::<_, i64>(3)?),
|
|
cache_read_input_tokens: u64_from_i64(row.get::<_, i64>(4)?),
|
|
cache_creation_input_tokens: u64_from_i64(row.get::<_, i64>(5)?),
|
|
last_input_tokens: u64_from_i64(row.get::<_, i64>(6)?),
|
|
tool_breakdown_json: row.get::<_, Option<String>>(7)?,
|
|
wake_from: row.get::<_, String>(8)?,
|
|
result_kind: row.get::<_, String>(9)?,
|
|
model: row.get::<_, String>(10)?,
|
|
})
|
|
})?;
|
|
|
|
let mut acc = TurnAccum::default();
|
|
for r in rows {
|
|
acc.push(r?, bucket_secs);
|
|
}
|
|
|
|
let buckets = fill_buckets(from, now, bucket_secs, &acc.by_bucket);
|
|
let duration_summary = summarize_durations(&mut acc.all_durations);
|
|
let mut models: Vec<String> = acc.model_set.into_iter().collect();
|
|
models.sort_unstable();
|
|
|
|
Ok(Snapshot {
|
|
window: window.label(),
|
|
bucket_seconds: bucket_secs,
|
|
now,
|
|
from,
|
|
turn_count: acc.turn_count,
|
|
buckets,
|
|
tool_breakdown: top_n(acc.tool_totals, 10),
|
|
bash_breakdown: read_bash_breakdown(&conn, from).unwrap_or_default(),
|
|
wake_mix: top_n(acc.wake_totals, 20),
|
|
result_mix: top_n(acc.result_totals, 20),
|
|
models,
|
|
duration_summary,
|
|
reminder_stats: None, // filled in by api_stats in web_ui.rs via fetch_reminder_stats RPC
|
|
// Inert-until-capture: `.ok()` maps both "no sessions in the window
|
|
// yet" and "sessions table absent on an older db" to None.
|
|
first_turn_ctx: read_first_turn_ctx(&conn, from).ok(),
|
|
session_count: read_session_count(&conn, from).ok(),
|
|
})
|
|
}
|
|
|
|
/// First-turn input tokens of the most recent fresh claude session that
|
|
/// started in `[from, now]`. Tracks system-prompt + CLAUDE.md sprawl:
|
|
/// the first turn of a fresh session (`--continue` suppressed) pays the
|
|
/// full static prefix as uncached input, so watching this over time
|
|
/// surfaces creep. Uses the agreed per-session derive — the first turn
|
|
/// (`ORDER BY started_at LIMIT 1`) of the latest session row.
|
|
///
|
|
/// Returns `Err` when the `sessions` table doesn't exist (older db) or no
|
|
/// fresh session in the window has a recorded turn yet; the caller maps
|
|
/// that to `None` (inert-until-capture), same as `read_bash_breakdown`.
|
|
fn read_first_turn_ctx(conn: &Connection, from: i64) -> rusqlite::Result<u64> {
|
|
conn.query_row(
|
|
"SELECT input_tokens FROM turn_stats
|
|
WHERE session_id = (
|
|
SELECT id FROM sessions WHERE started_at >= ?1 ORDER BY started_at DESC LIMIT 1
|
|
)
|
|
ORDER BY started_at ASC
|
|
LIMIT 1",
|
|
[from],
|
|
|row| row.get::<_, i64>(0).map(u64_from_i64),
|
|
)
|
|
}
|
|
|
|
/// Count of fresh claude sessions that started within `[from, now]`.
|
|
///
|
|
/// Each row in the `sessions` table represents one fresh `claude --print`
|
|
/// invocation (without `--continue`): a manual `/new-session`, an
|
|
/// auto-compaction fallback, or the very first turn after spawning.
|
|
///
|
|
/// Returns `Err` when the `sessions` table doesn't exist (older db) so the
|
|
/// caller can map to `None` — same inert-until-data pattern as
|
|
/// [`read_first_turn_ctx`].
|
|
fn read_session_count(conn: &Connection, from: i64) -> rusqlite::Result<u64> {
|
|
conn.query_row(
|
|
"SELECT COUNT(*) FROM sessions WHERE started_at >= ?1",
|
|
[from],
|
|
|row| row.get::<_, i64>(0).map(u64_from_i64),
|
|
)
|
|
}
|
|
|
|
/// Aggregate the top shell-command heads ("favorite tools") over
|
|
/// `[from, now]` from the `bash_commands` table — one row per bash task
|
|
/// (`ts INTEGER NOT NULL, head TEXT NOT NULL`), written by hive-bash-mcp.
|
|
///
|
|
/// Returns `Err` (which the caller maps to an empty list) when the
|
|
/// table doesn't exist yet — the writer creates it lazily on first
|
|
/// insert, so any agent that hasn't run a bash task since the capture
|
|
/// shipped simply has no table. Decoupling it this way means the read
|
|
/// side is inert-until-data and needs no schema coordination here.
|
|
fn read_bash_breakdown(conn: &Connection, from: i64) -> rusqlite::Result<Vec<KeyCount>> {
|
|
let mut stmt = conn.prepare(
|
|
"SELECT head, COUNT(*) AS n
|
|
FROM bash_commands
|
|
WHERE ts >= ?1
|
|
GROUP BY head",
|
|
)?;
|
|
let rows = stmt.query_map([from], |row| {
|
|
Ok((
|
|
row.get::<_, String>(0)?,
|
|
u64_from_i64(row.get::<_, i64>(1)?),
|
|
))
|
|
})?;
|
|
let mut totals: HashMap<String, u64> = HashMap::new();
|
|
for r in rows {
|
|
let (head, n) = r?;
|
|
*totals.entry(head).or_insert(0) += n;
|
|
}
|
|
Ok(top_n(totals, 10))
|
|
}
|
|
|
|
struct Row {
|
|
started_at: i64,
|
|
duration_ms: i64,
|
|
input_tokens: u64,
|
|
output_tokens: u64,
|
|
cache_read_input_tokens: u64,
|
|
cache_creation_input_tokens: u64,
|
|
last_input_tokens: u64,
|
|
tool_breakdown_json: Option<String>,
|
|
wake_from: String,
|
|
result_kind: String,
|
|
model: String,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct BucketAcc {
|
|
turn_count: u64,
|
|
durations: Vec<i64>,
|
|
input_tokens: u64,
|
|
output_tokens: u64,
|
|
cache_read_input_tokens: u64,
|
|
cache_creation_input_tokens: u64,
|
|
ctx_sum: u64,
|
|
ctx_max: u64,
|
|
model_counts: HashMap<String, u64>,
|
|
result_counts: HashMap<String, u64>,
|
|
}
|
|
|
|
fn fill_buckets(
|
|
from: i64,
|
|
now: i64,
|
|
bucket_secs: i64,
|
|
by_bucket: &HashMap<i64, BucketAcc>,
|
|
) -> Vec<Bucket> {
|
|
let start = (from / bucket_secs) * bucket_secs;
|
|
let mut out = Vec::new();
|
|
let mut ts = start;
|
|
while ts <= now {
|
|
let bucket = if let Some(acc) = by_bucket.get(&ts) {
|
|
let mut sorted = acc.durations.clone();
|
|
sorted.sort_unstable();
|
|
let avg = if sorted.is_empty() {
|
|
0.0
|
|
} else {
|
|
#[allow(
|
|
clippy::cast_precision_loss,
|
|
reason = "stat magnitudes (counts, token + duration sums) stay well under f64's 2^53 exact-integer range, so this averaging cast loses no precision in practice"
|
|
)]
|
|
let sum_f = sorted.iter().sum::<i64>() as f64;
|
|
#[allow(
|
|
clippy::cast_precision_loss,
|
|
reason = "stat magnitudes (counts, token + duration sums) stay well under f64's 2^53 exact-integer range, so this averaging cast loses no precision in practice"
|
|
)]
|
|
let len_f = sorted.len() as f64;
|
|
sum_f / len_f
|
|
};
|
|
let p50 = percentile(&sorted, 50);
|
|
let p95 = percentile(&sorted, 95);
|
|
let avg_ctx = if acc.turn_count == 0 {
|
|
0.0
|
|
} else {
|
|
#[allow(
|
|
clippy::cast_precision_loss,
|
|
reason = "stat magnitudes (counts, token + duration sums) stay well under f64's 2^53 exact-integer range, so this averaging cast loses no precision in practice"
|
|
)]
|
|
let sum_f = acc.ctx_sum as f64;
|
|
#[allow(
|
|
clippy::cast_precision_loss,
|
|
reason = "stat magnitudes (counts, token + duration sums) stay well under f64's 2^53 exact-integer range, so this averaging cast loses no precision in practice"
|
|
)]
|
|
let cnt_f = acc.turn_count as f64;
|
|
sum_f / cnt_f
|
|
};
|
|
Bucket {
|
|
ts,
|
|
turn_count: acc.turn_count,
|
|
avg_duration_ms: avg,
|
|
p50_duration_ms: p50,
|
|
p95_duration_ms: p95,
|
|
input_tokens: acc.input_tokens,
|
|
output_tokens: acc.output_tokens,
|
|
cache_read_input_tokens: acc.cache_read_input_tokens,
|
|
cache_creation_input_tokens: acc.cache_creation_input_tokens,
|
|
avg_ctx_tokens: avg_ctx,
|
|
max_ctx_tokens: acc.ctx_max,
|
|
model_counts: acc.model_counts.clone(),
|
|
result_counts: acc.result_counts.clone(),
|
|
}
|
|
} else {
|
|
Bucket {
|
|
ts,
|
|
turn_count: 0,
|
|
avg_duration_ms: 0.0,
|
|
p50_duration_ms: 0.0,
|
|
p95_duration_ms: 0.0,
|
|
input_tokens: 0,
|
|
output_tokens: 0,
|
|
cache_read_input_tokens: 0,
|
|
cache_creation_input_tokens: 0,
|
|
avg_ctx_tokens: 0.0,
|
|
max_ctx_tokens: 0,
|
|
model_counts: HashMap::new(),
|
|
result_counts: HashMap::new(),
|
|
}
|
|
};
|
|
out.push(bucket);
|
|
ts += bucket_secs;
|
|
}
|
|
out
|
|
}
|
|
|
|
fn summarize_durations(all: &mut [i64]) -> DurationSummary {
|
|
if all.is_empty() {
|
|
return DurationSummary::default();
|
|
}
|
|
all.sort_unstable();
|
|
#[allow(
|
|
clippy::cast_precision_loss,
|
|
reason = "stat magnitudes (counts, token + duration sums) stay well under f64's 2^53 exact-integer range, so this averaging cast loses no precision in practice"
|
|
)]
|
|
let sum_f = all.iter().sum::<i64>() as f64;
|
|
#[allow(
|
|
clippy::cast_precision_loss,
|
|
reason = "stat magnitudes (counts, token + duration sums) stay well under f64's 2^53 exact-integer range, so this averaging cast loses no precision in practice"
|
|
)]
|
|
let len_f = all.len() as f64;
|
|
DurationSummary {
|
|
avg: sum_f / len_f,
|
|
p50: percentile(all, 50),
|
|
p95: percentile(all, 95),
|
|
}
|
|
}
|
|
|
|
#[allow(
|
|
clippy::cast_precision_loss,
|
|
clippy::cast_possible_truncation,
|
|
clippy::cast_sign_loss
|
|
)]
|
|
fn percentile(sorted: &[i64], pct: u8) -> f64 {
|
|
if sorted.is_empty() {
|
|
return 0.0;
|
|
}
|
|
if sorted.len() == 1 {
|
|
return sorted[0] as f64;
|
|
}
|
|
// Nearest-rank, clamped.
|
|
let rank = ((f64::from(pct) / 100.0) * (sorted.len() as f64 - 1.0)).round() as usize;
|
|
sorted[rank.min(sorted.len() - 1)] as f64
|
|
}
|
|
|
|
fn top_n(map: HashMap<String, u64>, n: usize) -> Vec<KeyCount> {
|
|
let mut v: Vec<KeyCount> = map
|
|
.into_iter()
|
|
.map(|(key, count)| KeyCount { key, count })
|
|
.collect();
|
|
v.sort_unstable_by(|a, b| b.count.cmp(&a.count).then_with(|| a.key.cmp(&b.key)));
|
|
v.truncate(n);
|
|
v
|
|
}
|
|
|
|
fn u64_from_i64(v: i64) -> u64 {
|
|
u64::try_from(v).unwrap_or(0)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use rusqlite::params;
|
|
use std::sync::atomic::{AtomicU32, Ordering};
|
|
|
|
static SEQ: AtomicU32 = AtomicU32::new(0);
|
|
|
|
fn tmp_db() -> PathBuf {
|
|
let n = SEQ.fetch_add(1, Ordering::SeqCst);
|
|
let pid = std::process::id();
|
|
std::env::temp_dir().join(format!("hyperhive-stats-test-{pid}-{n}.sqlite"))
|
|
}
|
|
|
|
fn seed_db(path: &Path, rows: &[(i64, i64, &str, &str, &str, &str)]) {
|
|
let conn = Connection::open(path).unwrap();
|
|
conn.execute_batch(
|
|
"CREATE TABLE turn_stats (
|
|
id INTEGER PRIMARY KEY,
|
|
started_at INTEGER NOT NULL,
|
|
ended_at INTEGER NOT NULL,
|
|
duration_ms INTEGER NOT NULL,
|
|
model TEXT NOT NULL,
|
|
wake_from TEXT NOT NULL,
|
|
input_tokens INTEGER NOT NULL DEFAULT 0,
|
|
output_tokens INTEGER NOT NULL DEFAULT 0,
|
|
cache_read_input_tokens INTEGER NOT NULL DEFAULT 0,
|
|
cache_creation_input_tokens INTEGER NOT NULL DEFAULT 0,
|
|
last_input_tokens INTEGER NOT NULL DEFAULT 0,
|
|
last_output_tokens INTEGER NOT NULL DEFAULT 0,
|
|
last_cache_read_input_tokens INTEGER NOT NULL DEFAULT 0,
|
|
last_cache_creation_input_tokens INTEGER NOT NULL DEFAULT 0,
|
|
tool_call_count INTEGER NOT NULL DEFAULT 0,
|
|
tool_call_breakdown_json TEXT,
|
|
open_threads_count INTEGER,
|
|
open_reminders_count INTEGER,
|
|
result_kind TEXT NOT NULL,
|
|
note TEXT
|
|
);",
|
|
)
|
|
.unwrap();
|
|
for (started, dur, model, wake, result, tools_json) in rows {
|
|
conn.execute(
|
|
"INSERT INTO turn_stats
|
|
(started_at, ended_at, duration_ms, model, wake_from,
|
|
last_input_tokens, tool_call_breakdown_json, result_kind)
|
|
VALUES (?1, ?2, ?3, ?4, ?5, 1000, ?6, ?7)",
|
|
params![
|
|
started,
|
|
started + dur / 1000,
|
|
dur,
|
|
model,
|
|
wake,
|
|
tools_json,
|
|
result
|
|
],
|
|
)
|
|
.unwrap();
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn snapshot_aggregates_rows() {
|
|
let db = tmp_db();
|
|
let _ = std::fs::remove_file(&db);
|
|
let now = now_unix();
|
|
seed_db(
|
|
&db,
|
|
&[
|
|
(
|
|
now - 600,
|
|
5_000,
|
|
"opus",
|
|
"recv",
|
|
"ok",
|
|
r#"{"Read":2,"Bash":1}"#,
|
|
),
|
|
(now - 300, 10_000, "opus", "recv", "ok", r#"{"Read":3}"#),
|
|
(now - 100, 20_000, "sonnet", "operator", "failed", "{}"),
|
|
],
|
|
);
|
|
let s = snapshot(&db, Window::Day).unwrap();
|
|
assert_eq!(s.turn_count, 3);
|
|
assert_eq!(s.window, "24h");
|
|
assert_eq!(s.bucket_seconds, 3600);
|
|
let tool_map: HashMap<_, _> = s
|
|
.tool_breakdown
|
|
.iter()
|
|
.map(|kc| (kc.key.clone(), kc.count))
|
|
.collect();
|
|
assert_eq!(tool_map.get("Read").copied(), Some(5));
|
|
assert_eq!(tool_map.get("Bash").copied(), Some(1));
|
|
let wake_map: HashMap<_, _> = s
|
|
.wake_mix
|
|
.iter()
|
|
.map(|kc| (kc.key.clone(), kc.count))
|
|
.collect();
|
|
assert_eq!(wake_map.get("recv").copied(), Some(2));
|
|
assert_eq!(wake_map.get("operator").copied(), Some(1));
|
|
let result_map: HashMap<_, _> = s
|
|
.result_mix
|
|
.iter()
|
|
.map(|kc| (kc.key.clone(), kc.count))
|
|
.collect();
|
|
assert_eq!(result_map.get("ok").copied(), Some(2));
|
|
assert_eq!(result_map.get("failed").copied(), Some(1));
|
|
// Model breakdown: 2 opus + 1 sonnet, all in the same hour
|
|
// bucket given the 24h window.
|
|
assert_eq!(s.models, vec!["opus".to_string(), "sonnet".to_string()]);
|
|
let mut model_totals: HashMap<String, u64> = HashMap::new();
|
|
for b in &s.buckets {
|
|
for (k, v) in &b.model_counts {
|
|
*model_totals.entry(k.clone()).or_insert(0) += v;
|
|
}
|
|
}
|
|
assert_eq!(model_totals.get("opus").copied(), Some(2));
|
|
assert_eq!(model_totals.get("sonnet").copied(), Some(1));
|
|
// Durations: [5000, 10000, 20000] → avg ≈ 11666.67, p50 = 10000, p95 ~ 20000
|
|
assert!((s.duration_summary.avg - 11_666.666_666_666_666).abs() < 1.0);
|
|
assert!((s.duration_summary.p50 - 10_000.0).abs() < 1.0);
|
|
assert!((s.duration_summary.p95 - 20_000.0).abs() < 1.0);
|
|
}
|
|
|
|
#[test]
|
|
fn empty_window_still_paints_buckets() {
|
|
let db = tmp_db();
|
|
let _ = std::fs::remove_file(&db);
|
|
seed_db(&db, &[]);
|
|
let s = snapshot(&db, Window::Day).unwrap();
|
|
assert_eq!(s.turn_count, 0);
|
|
// 24h / 1h buckets = ~24-25 buckets covering the window.
|
|
assert!(s.buckets.len() >= 24);
|
|
assert!(s.buckets.iter().all(|b| b.turn_count == 0));
|
|
}
|
|
|
|
#[test]
|
|
fn week_uses_daily_buckets() {
|
|
let db = tmp_db();
|
|
let _ = std::fs::remove_file(&db);
|
|
seed_db(&db, &[]);
|
|
let s = snapshot(&db, Window::Week).unwrap();
|
|
assert_eq!(s.window, "7d");
|
|
assert_eq!(s.bucket_seconds, 86_400);
|
|
assert!(s.buckets.len() >= 7);
|
|
}
|
|
|
|
/// `bash_breakdown` degrades gracefully when the `bash_commands`
|
|
/// table hasn't been created yet (the capture side hasn't shipped /
|
|
/// run on this agent). A `seed_db` DB has no such table, so the read
|
|
/// must yield an empty list rather than erroring the whole snapshot.
|
|
#[test]
|
|
fn bash_breakdown_empty_without_table() {
|
|
let db = tmp_db();
|
|
let _ = std::fs::remove_file(&db);
|
|
seed_db(&db, &[(now_unix() - 100, 1000, "opus", "recv", "ok", "{}")]);
|
|
let s = snapshot(&db, Window::Day).unwrap();
|
|
assert!(s.bash_breakdown.is_empty());
|
|
}
|
|
|
|
/// With a populated `bash_commands` table, `bash_breakdown` rolls up
|
|
/// per-head counts (busiest first) and respects the window cutoff.
|
|
#[test]
|
|
fn bash_breakdown_aggregates_heads() {
|
|
let db = tmp_db();
|
|
let _ = std::fs::remove_file(&db);
|
|
seed_db(&db, &[]);
|
|
let now = now_unix();
|
|
let conn = Connection::open(&db).unwrap();
|
|
conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);")
|
|
.unwrap();
|
|
// 3x cargo + 2x git inside the window, 1x rg outside it.
|
|
for (ts, head) in [
|
|
(now - 100, "cargo"),
|
|
(now - 200, "cargo"),
|
|
(now - 300, "cargo"),
|
|
(now - 400, "git"),
|
|
(now - 500, "git"),
|
|
(now - (2 * 24 * 3600), "rg"), // older than the 24h window
|
|
] {
|
|
conn.execute(
|
|
"INSERT INTO bash_commands (ts, head) VALUES (?1, ?2)",
|
|
params![ts, head],
|
|
)
|
|
.unwrap();
|
|
}
|
|
let s = snapshot(&db, Window::Day).unwrap();
|
|
let map: HashMap<_, _> = s
|
|
.bash_breakdown
|
|
.iter()
|
|
.map(|kc| (kc.key.clone(), kc.count))
|
|
.collect();
|
|
assert_eq!(map.get("cargo").copied(), Some(3));
|
|
assert_eq!(map.get("git").copied(), Some(2));
|
|
// `rg` fell outside the 24h window — excluded.
|
|
assert_eq!(map.get("rg").copied(), None);
|
|
// top_n orders busiest first.
|
|
assert_eq!(
|
|
s.bash_breakdown.first().map(|kc| kc.key.as_str()),
|
|
Some("cargo")
|
|
);
|
|
}
|
|
}
|