//! 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; /// 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, _ => 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, /// Top tools by call count across the window. Capped to 10. pub tool_breakdown: Vec, /// 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, pub wake_mix: Vec, pub result_mix: Vec, /// 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, /// 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, /// 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, } #[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, /// 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, } #[derive(Debug, Serialize)] pub struct KeyCount { pub key: String, pub count: u64, } #[derive(Debug, Default, Serialize)] pub struct DurationSummary { pub avg_ms: f64, pub p50_ms: f64, pub p95_ms: 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_secs(); 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, } } fn snapshot(path: &Path, window: Window) -> Result { // Read-only open so an in-flight writer (the harness's own // turn_stats sink) never blocks us and 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()))?; let now = now_secs(); // Fixed windows look back a constant span; `all` starts at the earliest // recorded turn (`MIN(started_at)`, falling back to `now` on an empty // table) and sizes its buckets adaptively from that span. let (from, bucket_secs) = match window { Window::All => { let min_ts: Option = 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>(7)?, wake_from: row.get::<_, String>(8)?, result_kind: row.get::<_, String>(9)?, model: row.get::<_, String>(10)?, }) })?; let mut by_bucket: HashMap = HashMap::new(); let mut tool_totals: HashMap = HashMap::new(); let mut wake_totals: HashMap = HashMap::new(); let mut result_totals: HashMap = HashMap::new(); let mut model_set: HashSet = HashSet::new(); let mut all_durations: Vec = Vec::new(); let mut turn_count: u64 = 0; for r in rows { let r = r?; turn_count += 1; let bucket_ts = (r.started_at / bucket_secs) * bucket_secs; let acc = by_bucket.entry(bucket_ts).or_default(); acc.turn_count += 1; acc.durations.push(r.duration_ms.max(0)); acc.input_tokens = acc.input_tokens.saturating_add(r.input_tokens); acc.output_tokens = acc.output_tokens.saturating_add(r.output_tokens); acc.cache_read_input_tokens = acc .cache_read_input_tokens .saturating_add(r.cache_read_input_tokens); acc.cache_creation_input_tokens = acc .cache_creation_input_tokens .saturating_add(r.cache_creation_input_tokens); acc.ctx_sum = acc.ctx_sum.saturating_add(r.last_input_tokens); acc.ctx_max = acc.ctx_max.max(r.last_input_tokens); *acc.model_counts.entry(r.model.clone()).or_insert(0) += 1; *acc.result_counts.entry(r.result_kind.clone()).or_insert(0) += 1; all_durations.push(r.duration_ms.max(0)); *wake_totals.entry(r.wake_from).or_insert(0) += 1; *result_totals.entry(r.result_kind).or_insert(0) += 1; model_set.insert(r.model); if let Some(json) = r.tool_breakdown_json && let Ok(map) = serde_json::from_str::>(&json) { for (k, v) in map { *tool_totals.entry(k).or_insert(0) += v; } } } let buckets = fill_buckets(from, now, bucket_secs, &by_bucket); let duration_summary = summarize_durations(&mut all_durations); let mut models: Vec = model_set.into_iter().collect(); models.sort_unstable(); Ok(Snapshot { window: window.label(), bucket_seconds: bucket_secs, now, from, turn_count, buckets, tool_breakdown: top_n(tool_totals, 10), bash_breakdown: read_bash_breakdown(&conn, from).unwrap_or_default(), wake_mix: top_n(wake_totals, 20), result_mix: top_n(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 fresh session in the // window yet" (QueryReturnedNoRows) and "sessions table absent on // an older db" (Err) to None, same decoupling as read_bash_breakdown. first_turn_ctx: read_first_turn_ctx(&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 { 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), ) } /// 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> { 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 = 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, wake_from: String, result_kind: String, model: String, } #[derive(Default)] struct BucketAcc { turn_count: u64, durations: Vec, 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, result_counts: HashMap, } fn fill_buckets( from: i64, now: i64, bucket_secs: i64, by_bucket: &HashMap, ) -> Vec { 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::() 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::() 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_ms: sum_f / len_f, p50_ms: percentile(all, 50), p95_ms: 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, n: usize) -> Vec { let mut v: Vec = 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) } fn now_secs() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX)) } #[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_secs(); 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 = 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_ms - 11_666.666_666_666_666).abs() < 1.0); assert!((s.duration_summary.p50_ms - 10_000.0).abs() < 1.0); assert!((s.duration_summary.p95_ms - 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_secs() - 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_secs(); 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") ); } }