426 lines
18 KiB
Rust
426 lines
18 KiB
Rust
//! Per-turn analytics sink. One sqlite row per claude turn captures:
|
|
//! identity (`model`, `wake_from`, `result_kind`), timing (`started_at`,
|
|
//! `ended_at`, `duration_ms`), cost (token counts), and behaviour (tool-call
|
|
//! count + per-tool breakdown).
|
|
//!
|
|
//! **Captured but not yet read** (written every turn, no reader today —
|
|
//! kept for a future chart / backfill, not consumed by `stats::snapshot`
|
|
//! or the host rollup): `tool_call_count` (the snapshot recomputes tool
|
|
//! totals from `tool_call_breakdown_json` instead), `open_threads_count` +
|
|
//! `open_reminders_count` (planned: a loose-ends-over-time trend), and
|
|
//! `note` (failure detail for `result_kind = "failed"`).
|
|
//!
|
|
//! Lives next to `hyperhive-events.sqlite` in the agent's state dir
|
|
//! so the host-side state vacuum sweep can reach both. Schema is
|
|
//! intentionally append-only — every column has a default so future
|
|
//! additions don't break old readers; new columns land via
|
|
//! `ALTER TABLE ... ADD COLUMN ... DEFAULT ...` in the migration
|
|
//! block.
|
|
//!
|
|
//! Writes are best-effort: a failed insert logs a warning and lets
|
|
//! the turn loop continue. The next turn either succeeds or the
|
|
//! operator sees the journal trail.
|
|
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Mutex;
|
|
|
|
use anyhow::{Context, Result};
|
|
use rusqlite::{Connection, params};
|
|
|
|
/// SQL bootstrap. CREATE TABLE IF NOT EXISTS so first-boot agents
|
|
/// and existing ones converge on the same shape. The base table is
|
|
/// fresh-install only; additive migrations land via `MIGRATIONS`
|
|
/// below as try-and-ignore ALTERs so existing dbs catch up.
|
|
const SCHEMA: &str = "
|
|
CREATE TABLE IF NOT EXISTS turn_stats (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
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,
|
|
session_id INTEGER
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_turn_stats_started
|
|
ON turn_stats (started_at DESC);
|
|
-- NOTE: the index on session_id is created in MIGRATIONS, not here. On an
|
|
-- existing pre-session db the `CREATE TABLE IF NOT EXISTS` above is a no-op
|
|
-- (the old table has no session_id column), so indexing session_id in this
|
|
-- batch would fail with `no such column` and abort the whole SCHEMA apply —
|
|
-- which disables the stats sink. The column is added by MIGRATIONS first.
|
|
-- One row per fresh claude session (minted when --continue is suppressed).
|
|
-- turn_stats.session_id FKs here so per-session stats (first-turn tokens,
|
|
-- per-session totals, turn count, duration) are one GROUP BY away.
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
started_at INTEGER NOT NULL,
|
|
model TEXT NOT NULL
|
|
);
|
|
";
|
|
|
|
/// Additive column migrations. Each runs unconditionally and ignores
|
|
/// `duplicate column name` errors — sqlite < 3.35 lacks
|
|
/// `ADD COLUMN IF NOT EXISTS`, so try-and-ignore is the portable path.
|
|
/// New columns MUST carry a default so existing rows decode.
|
|
const MIGRATIONS: &[&str] = &[
|
|
"ALTER TABLE turn_stats ADD COLUMN last_input_tokens INTEGER NOT NULL DEFAULT 0",
|
|
"ALTER TABLE turn_stats ADD COLUMN last_output_tokens INTEGER NOT NULL DEFAULT 0",
|
|
"ALTER TABLE turn_stats ADD COLUMN last_cache_read_input_tokens INTEGER NOT NULL DEFAULT 0",
|
|
"ALTER TABLE turn_stats ADD COLUMN last_cache_creation_input_tokens INTEGER NOT NULL DEFAULT 0",
|
|
// Nullable FK to sessions.id — no default; pre-migration rows stay NULL
|
|
// (the surface treats NULL as "no session", inert until capture lands).
|
|
"ALTER TABLE turn_stats ADD COLUMN session_id INTEGER",
|
|
// Index on session_id — must run AFTER the column is added, so it lives
|
|
// here rather than in SCHEMA (see the note there). Idempotent.
|
|
"CREATE INDEX IF NOT EXISTS idx_turn_stats_session ON turn_stats (session_id)",
|
|
];
|
|
|
|
/// One row to be inserted. `Option`-wrapped fields default to NULL
|
|
/// when the harness couldn't gather them (e.g. socket roundtrip for
|
|
/// `open_threads` failed) so a partial row beats no row.
|
|
#[derive(Debug, Clone)]
|
|
pub struct TurnStatRow {
|
|
pub started_at: i64,
|
|
pub ended_at: i64,
|
|
pub duration_ms: i64,
|
|
pub model: String,
|
|
pub wake_from: String,
|
|
/// Cumulative across every inference in the turn (cost signal).
|
|
pub input_tokens: u64,
|
|
pub output_tokens: u64,
|
|
pub cache_read_input_tokens: u64,
|
|
pub cache_creation_input_tokens: u64,
|
|
/// Last inference's usage — the actual context size at turn end.
|
|
pub last_input_tokens: u64,
|
|
pub last_output_tokens: u64,
|
|
pub last_cache_read_input_tokens: u64,
|
|
pub last_cache_creation_input_tokens: u64,
|
|
/// Captured, not yet read — the snapshot recomputes tool totals from
|
|
/// `tool_call_breakdown_json` (see the module doc).
|
|
pub tool_call_count: u64,
|
|
/// Per-tool breakdown as JSON: `{"Read":12,"Bash":3,...}`. None
|
|
/// when no tools were called (saves a sqlite write of `"{}"`).
|
|
pub tool_call_breakdown_json: Option<String>,
|
|
/// Post-turn loose-ends snapshot. Captured, not yet read — planned to
|
|
/// feed a loose-ends-over-time trend on the stats page.
|
|
pub open_threads_count: Option<u64>,
|
|
pub open_reminders_count: Option<u64>,
|
|
/// `"ok" | "failed" | "prompt_too_long"`.
|
|
pub result_kind: &'static str,
|
|
/// Failure detail for `result_kind = "failed"`. Captured, not yet read.
|
|
pub note: Option<String>,
|
|
/// FK to `sessions.id` for the fresh claude session this turn belongs
|
|
/// to. `None` on pre-capture rows (and when the stats db couldn't mint
|
|
/// a session) so the read side degrades to empty.
|
|
pub session_id: Option<i64>,
|
|
}
|
|
|
|
/// Thin sqlite wrapper. Cloning is cheap (Arc-shared connection).
|
|
#[derive(Clone)]
|
|
pub struct TurnStats {
|
|
inner: std::sync::Arc<Mutex<Connection>>,
|
|
}
|
|
|
|
impl TurnStats {
|
|
/// Open the per-agent stats db, creating the file + schema if
|
|
/// missing. Returns `None` when the db can't be opened (read-only
|
|
/// fs in tests, missing state dir) — the harness logs and
|
|
/// continues without a sink rather than failing the turn loop.
|
|
#[must_use]
|
|
pub fn open_default() -> Option<Self> {
|
|
let path = default_path();
|
|
match Self::open(&path) {
|
|
Ok(s) => Some(s),
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
error = ?e,
|
|
path = %path.display(),
|
|
"turn_stats: open failed; per-turn analytics disabled"
|
|
);
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
fn open(path: &Path) -> Result<Self> {
|
|
if let Some(parent) = path.parent() {
|
|
let _ = std::fs::create_dir_all(parent);
|
|
}
|
|
let conn = Connection::open(path)
|
|
.with_context(|| format!("open turn_stats db {}", path.display()))?;
|
|
conn.execute_batch(SCHEMA)
|
|
.context("apply turn_stats schema")?;
|
|
for stmt in MIGRATIONS {
|
|
// Ignore "duplicate column name" — the migration already ran.
|
|
// Any other error is logged but doesn't fail open() because the
|
|
// base schema works and we'd rather keep the harness alive than
|
|
// crash on an upgrade hiccup.
|
|
if let Err(e) = conn.execute(stmt, []) {
|
|
let msg = e.to_string();
|
|
if !msg.contains("duplicate column name") {
|
|
tracing::warn!(error = %msg, stmt, "turn_stats migration failed");
|
|
}
|
|
}
|
|
}
|
|
Ok(Self {
|
|
inner: std::sync::Arc::new(Mutex::new(conn)),
|
|
})
|
|
}
|
|
|
|
/// Insert a row. Best-effort — logs + swallows errors so a sqlite
|
|
/// hiccup (locked db, full disk) doesn't crash the harness.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the internal lock is poisoned.
|
|
pub fn record(&self, row: &TurnStatRow) {
|
|
let conn = self.inner.lock().unwrap();
|
|
let res = conn.execute(
|
|
"INSERT INTO turn_stats (
|
|
started_at, ended_at, duration_ms, model, wake_from,
|
|
input_tokens, output_tokens,
|
|
cache_read_input_tokens, cache_creation_input_tokens,
|
|
last_input_tokens, last_output_tokens,
|
|
last_cache_read_input_tokens, last_cache_creation_input_tokens,
|
|
tool_call_count, tool_call_breakdown_json,
|
|
open_threads_count, open_reminders_count,
|
|
result_kind, note, session_id
|
|
) VALUES (
|
|
?1, ?2, ?3, ?4, ?5,
|
|
?6, ?7,
|
|
?8, ?9,
|
|
?10, ?11,
|
|
?12, ?13,
|
|
?14, ?15,
|
|
?16, ?17,
|
|
?18, ?19, ?20
|
|
)",
|
|
params![
|
|
row.started_at,
|
|
row.ended_at,
|
|
row.duration_ms,
|
|
row.model,
|
|
row.wake_from,
|
|
i64::try_from(row.input_tokens).unwrap_or(i64::MAX),
|
|
i64::try_from(row.output_tokens).unwrap_or(i64::MAX),
|
|
i64::try_from(row.cache_read_input_tokens).unwrap_or(i64::MAX),
|
|
i64::try_from(row.cache_creation_input_tokens).unwrap_or(i64::MAX),
|
|
i64::try_from(row.last_input_tokens).unwrap_or(i64::MAX),
|
|
i64::try_from(row.last_output_tokens).unwrap_or(i64::MAX),
|
|
i64::try_from(row.last_cache_read_input_tokens).unwrap_or(i64::MAX),
|
|
i64::try_from(row.last_cache_creation_input_tokens).unwrap_or(i64::MAX),
|
|
i64::try_from(row.tool_call_count).unwrap_or(i64::MAX),
|
|
row.tool_call_breakdown_json,
|
|
row.open_threads_count
|
|
.map(|n| i64::try_from(n).unwrap_or(i64::MAX)),
|
|
row.open_reminders_count
|
|
.map(|n| i64::try_from(n).unwrap_or(i64::MAX)),
|
|
row.result_kind,
|
|
row.note,
|
|
row.session_id,
|
|
],
|
|
);
|
|
if let Err(e) = res {
|
|
tracing::warn!(error = ?e, "turn_stats: insert failed");
|
|
}
|
|
}
|
|
|
|
/// Mint a new session row at fresh-session start, returning its `id`
|
|
/// for stamping onto this session's `turn_stats` rows. Best-effort —
|
|
/// returns `None` (and logs) on any sqlite error, so a hiccup degrades
|
|
/// to NULL `session_id` rows rather than crashing the turn loop.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the internal lock is poisoned.
|
|
#[must_use]
|
|
pub fn start_session(&self, started_at: i64, model: &str) -> Option<i64> {
|
|
let conn = self.inner.lock().unwrap();
|
|
match conn.execute(
|
|
"INSERT INTO sessions (started_at, model) VALUES (?1, ?2)",
|
|
params![started_at, model],
|
|
) {
|
|
Ok(_) => Some(conn.last_insert_rowid()),
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "turn_stats: start_session insert failed");
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Token counts from the most recently inserted row, if any.
|
|
/// Returns `(ctx, cost)` — both backfill `Bus` on startup so the
|
|
/// per-agent web UI's ctx + cost badges paint with real numbers on
|
|
/// cold load instead of waiting for the next `TokenUsageChanged`
|
|
/// SSE event. Best-effort: any sqlite error returns `(None, None)`.
|
|
///
|
|
/// Pre-migration rows (before the `last_*_tokens` columns existed)
|
|
/// have last-inference zeros — those rows yield `ctx = None` so the
|
|
/// badge stays empty until the next real turn rather than showing a
|
|
/// misleading 0.
|
|
/// # Panics
|
|
///
|
|
/// Panics if the internal lock is poisoned.
|
|
#[must_use]
|
|
pub fn last_usage(
|
|
&self,
|
|
) -> (
|
|
Option<hive_claude::TokenUsage>,
|
|
Option<hive_claude::TokenUsage>,
|
|
) {
|
|
let conn = self.inner.lock().unwrap();
|
|
conn.query_row(
|
|
"SELECT input_tokens, output_tokens,
|
|
cache_read_input_tokens, cache_creation_input_tokens,
|
|
last_input_tokens, last_output_tokens,
|
|
last_cache_read_input_tokens, last_cache_creation_input_tokens
|
|
FROM turn_stats
|
|
-- `id` (AUTOINCREMENT) is monotonic with insertion, so this is
|
|
-- the most-recently-inserted row even among same-second turns
|
|
-- (which `started_at DESC` would order arbitrarily).
|
|
ORDER BY id DESC
|
|
LIMIT 1",
|
|
[],
|
|
|row| {
|
|
let g = |i: usize| -> rusqlite::Result<u64> {
|
|
Ok(u64::try_from(row.get::<_, i64>(i)?).unwrap_or(0))
|
|
};
|
|
// `TokenUsage` is `#[non_exhaustive]` (hive-claude 0.1.0+) —
|
|
// struct-expression construction is blocked for downstream
|
|
// crates entirely, even with a `..base` (functional update
|
|
// doesn't carve out an exception, per the non_exhaustive
|
|
// reference semantics). Build off `Default` and assign the
|
|
// (all-`pub`) fields instead.
|
|
let mut cost = hive_claude::TokenUsage::default();
|
|
cost.input_tokens = g(0)?;
|
|
cost.output_tokens = g(1)?;
|
|
cost.cache_read_input_tokens = g(2)?;
|
|
cost.cache_creation_input_tokens = g(3)?;
|
|
let mut last = hive_claude::TokenUsage::default();
|
|
last.input_tokens = g(4)?;
|
|
last.output_tokens = g(5)?;
|
|
last.cache_read_input_tokens = g(6)?;
|
|
last.cache_creation_input_tokens = g(7)?;
|
|
let ctx = if last == hive_claude::TokenUsage::default() {
|
|
None
|
|
} else {
|
|
Some(last)
|
|
};
|
|
Ok((ctx, Some(cost)))
|
|
},
|
|
)
|
|
.unwrap_or((None, None))
|
|
}
|
|
}
|
|
|
|
fn default_path() -> PathBuf {
|
|
crate::paths::harness_dir().join("hyperhive-turn-stats.sqlite")
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// A `turn_stats` db in the *pre-session* shape: the original table with no
|
|
/// `session_id` column, plus the `started_at` index. This is what every
|
|
/// agent created before the sessions feature has on disk.
|
|
fn seed_pre_session_db(path: &Path) {
|
|
let conn = Connection::open(path).unwrap();
|
|
conn.execute_batch(
|
|
"CREATE TABLE turn_stats (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
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
|
|
);
|
|
CREATE INDEX idx_turn_stats_started ON turn_stats (started_at DESC);",
|
|
)
|
|
.unwrap();
|
|
}
|
|
|
|
fn sample_row() -> TurnStatRow {
|
|
TurnStatRow {
|
|
started_at: 100,
|
|
ended_at: 101,
|
|
duration_ms: 1_000,
|
|
model: "opus".to_owned(),
|
|
wake_from: "recv".to_owned(),
|
|
input_tokens: 10,
|
|
output_tokens: 5,
|
|
cache_read_input_tokens: 0,
|
|
cache_creation_input_tokens: 0,
|
|
last_input_tokens: 10,
|
|
last_output_tokens: 5,
|
|
last_cache_read_input_tokens: 0,
|
|
last_cache_creation_input_tokens: 0,
|
|
tool_call_count: 1,
|
|
tool_call_breakdown_json: None,
|
|
open_threads_count: None,
|
|
open_reminders_count: None,
|
|
result_kind: "ok",
|
|
note: None,
|
|
session_id: None,
|
|
}
|
|
}
|
|
|
|
/// Regression: a pre-session db must `open()` cleanly (the `session_id`
|
|
/// index used to live in `SCHEMA` and aborted the apply with
|
|
/// `no such column`, silently disabling the stats sink), get the column
|
|
/// added by `MIGRATIONS`, and then accept writes.
|
|
#[test]
|
|
fn open_upgrades_pre_session_db_and_writes() {
|
|
let path = std::env::temp_dir().join("hyperhive-turnstats-pre-session-regression.sqlite");
|
|
let _ = std::fs::remove_file(&path);
|
|
seed_pre_session_db(&path);
|
|
|
|
let stats = TurnStats::open(&path).expect("open() must succeed on a pre-session db");
|
|
stats.record(&sample_row());
|
|
|
|
let conn = Connection::open(&path).unwrap();
|
|
let rows: i64 = conn
|
|
.query_row("SELECT COUNT(*) FROM turn_stats", [], |r| r.get(0))
|
|
.unwrap();
|
|
assert_eq!(rows, 1, "the row must be written once the column is added");
|
|
let has_session_id: i64 = conn
|
|
.query_row(
|
|
"SELECT COUNT(*) FROM pragma_table_info('turn_stats') \
|
|
WHERE name = 'session_id'",
|
|
[],
|
|
|r| r.get(0),
|
|
)
|
|
.unwrap();
|
|
assert_eq!(has_session_id, 1, "MIGRATIONS must add session_id");
|
|
|
|
let _ = std::fs::remove_file(&path);
|
|
}
|
|
}
|