feat(#1466): add sessions table + turn_stats.session_id capture for per-session stats

hive-ag3nt/src/bin/hive.rs

@@ -557,6 +557,14 @@ async fn handle_turn<S: Surface>(
         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;

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<AtomicBool>,
+    /// 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<Mutex<Option<i64>>>,
+    /// 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<AtomicBool>,
     /// 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<i64> {
+        *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<i64>) {
+        *self.session_id.lock().unwrap() = id;
+    }
+
     /// Currently-selected claude model name. Read on every turn so a
     /// `/model <name>` flip takes effect on the next turn.
     ///

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(),
     }
 }

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(),
         });

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<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).
@@ -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<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