From 17ddba0341248d10e2c7898e0b538ab821bfac1c Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 10 Jun 2026 19:37:48 +0200 Subject: [PATCH] feat(#1466): add sessions table + turn_stats.session_id capture for per-session stats --- hive-ag3nt/src/bin/hive.rs | 8 ++++++ hive-ag3nt/src/events.rs | 44 +++++++++++++++++++++++++++++++ hive-ag3nt/src/serve_common.rs | 1 + hive-ag3nt/src/turn.rs | 4 +++ hive-ag3nt/src/turn_stats.rs | 48 +++++++++++++++++++++++++++++++--- 5 files changed, 102 insertions(+), 3 deletions(-) diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index 04b4c39a..fe19e4f0 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -557,6 +557,14 @@ async fn handle_turn( S::send_to_parent(socket, format_turn_failure(e)).await; } if let Some(stats) = stats { + // Fresh session this turn → mint a `sessions` row and set its id on + // the bus so this turn (and subsequent ones until the next fresh + // start) stamp `turn_stats.session_id`. Takes the one-shot flag + // `run_claude` set when it suppressed `--continue`. + if bus.take_fresh_session() { + let sid = stats.start_session(started_at, &model_at_start); + bus.set_session_id(sid); + } let ended_at = serve_common::now_unix(); let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); let (open_threads, open_reminders) = S::post_turn_counts(socket).await; diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index a4e15a07..e4144324 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -618,6 +618,15 @@ pub struct Bus { /// behavior. Atomic so the consumer can take-and-clear without a /// lock. skip_continue_once: Arc, + /// Current fresh-claude-session id (FK to `sessions.id`). Set by the + /// bin loop after minting a session row on a fresh start; stamped onto + /// every `turn_stats` row until the next fresh session. `None` before + /// the first fresh turn or when the stats db is unavailable. + session_id: Arc>>, + /// One-shot: `run_claude` flips this true when it suppresses + /// `--continue` (a fresh session). The bin loop takes-and-clears it + /// after the turn to decide whether to mint a new `sessions` row. + fresh_session: Arc, /// Per-turn tool-call counter. Reset by the bin loop between /// turns via `take_tool_calls`. Populated by `observe_stream` as /// the stdout pump parses each stream-json line. Powers the @@ -691,6 +700,8 @@ impl Bus { last_cost_usage: Arc::new(Mutex::new(None)), rate_limited: Arc::new(AtomicBool::new(was_rate_limited)), skip_continue_once: Arc::new(AtomicBool::new(false)), + session_id: Arc::new(Mutex::new(None)), + fresh_session: Arc::new(AtomicBool::new(false)), tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())), last_turn_ended_unix: Arc::new(AtomicI64::new(0)), api_context_window: Arc::new(Mutex::new(None)), @@ -727,6 +738,39 @@ impl Bus { self.skip_continue_once.swap(false, Ordering::SeqCst) } + /// Mark that the current turn started a fresh claude session. + /// `run_claude` calls this when it suppresses `--continue`. + pub fn mark_fresh_session(&self) { + self.fresh_session.store(true, Ordering::SeqCst); + } + + /// Take + clear the fresh-session one-shot. The bin loop calls this + /// after the turn to decide whether to mint a new `sessions` row. + #[must_use] + pub fn take_fresh_session(&self) -> bool { + self.fresh_session.swap(false, Ordering::SeqCst) + } + + /// Currently-active session id (FK to `sessions.id`), or `None`. + /// + /// # Panics + /// + /// Panics if the internal lock is poisoned. + #[must_use] + pub fn current_session_id(&self) -> Option { + *self.session_id.lock().unwrap() + } + + /// Set the active session id after minting a `sessions` row on a + /// fresh start. + /// + /// # Panics + /// + /// Panics if the internal lock is poisoned. + pub fn set_session_id(&self, id: Option) { + *self.session_id.lock().unwrap() = id; + } + /// Currently-selected claude model name. Read on every turn so a /// `/model ` flip takes effect on the next turn. /// diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs index e401ef9e..7af95d68 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -114,5 +114,6 @@ pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow { open_reminders_count, result_kind, note, + session_id: bus.current_session_id(), } } diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 450fc3a8..0cd5d1bb 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -652,6 +652,10 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, let effort = bus.effort(); let resume = !bus.take_skip_continue(); if !resume { + // Flag the fresh session so the bin loop mints a new `sessions` + // row + stamps its id onto this turn's stats (and subsequent + // turns until the next fresh start). + bus.mark_fresh_session(); bus.emit(LiveEvent::Note { text: "fresh session (--continue suppressed for this turn)".into(), }); diff --git a/hive-ag3nt/src/turn_stats.rs b/hive-ag3nt/src/turn_stats.rs index 6a9238bc..d82123c0 100644 --- a/hive-ag3nt/src/turn_stats.rs +++ b/hive-ag3nt/src/turn_stats.rs @@ -46,10 +46,21 @@ CREATE TABLE IF NOT EXISTS turn_stats ( open_threads_count INTEGER, open_reminders_count INTEGER, result_kind TEXT NOT NULL, - note TEXT + note TEXT, + session_id INTEGER ); CREATE INDEX IF NOT EXISTS idx_turn_stats_started ON turn_stats (started_at DESC); +CREATE INDEX IF NOT EXISTS idx_turn_stats_session + ON turn_stats (session_id); +-- 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 @@ -61,6 +72,9 @@ const MIGRATIONS: &[&str] = &[ "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", ]; /// One row to be inserted. `Option`-wrapped fields default to NULL @@ -92,6 +106,10 @@ pub struct TurnStatRow { /// `"ok" | "failed" | "prompt_too_long"`. pub result_kind: &'static str, pub note: Option, + /// 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, } /// Thin sqlite wrapper. Cloning is cheap (Arc-shared connection). @@ -163,7 +181,7 @@ impl TurnStats { 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 + result_kind, note, session_id ) VALUES ( ?1, ?2, ?3, ?4, ?5, ?6, ?7, @@ -172,7 +190,7 @@ impl TurnStats { ?12, ?13, ?14, ?15, ?16, ?17, - ?18, ?19 + ?18, ?19, ?20 )", params![ row.started_at, @@ -196,6 +214,7 @@ impl TurnStats { .map(|n| i64::try_from(n).unwrap_or(i64::MAX)), row.result_kind, row.note, + row.session_id, ], ); if let Err(e) = res { @@ -203,6 +222,29 @@ impl TurnStats { } } + /// 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 { + 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