feat(#2608): show session count in agent stats window

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.

frontend/packages/agent/src/stats.js

@@ -123,6 +123,12 @@ window.Chart = Chart;
         ['reminders pending',   fmtInt(s.reminder_stats.pending)],
       );
     }
+    // Session count: fresh claude sessions started in the window (each
+    // new-session or auto-compaction-fallback mints one). Omitted until
+    // the sessions table exists (older db).
+    if (typeof s.session_count === 'number') {
+      chips.push(['sessions', fmtInt(s.session_count)]);
+    }
     // First-turn ctx: input tokens of the most recent fresh session's
     // first turn — the cold system-prompt + CLAUDE.md cost, a sprawl
     // proxy. Omitted from the JSON (and so absent here) until the

hive-agent/src/stats.rs

@@ -149,6 +149,14 @@ pub struct Snapshot {
     /// 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)]
@@ -235,6 +243,56 @@ fn empty_snapshot(window: Window) -> Snapshot {
         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;
+            }
+        }
     }
 }
 
@@ -244,14 +302,12 @@ fn snapshot(path: &Path, window: Window) -> Result<Snapshot> {
         .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 —
+    // and blanks the whole stats page. Wait out the brief write instead.
-    // matches hive-c0re's host-side reader (`hive_stats::read_agent`).
     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 (`MIN(started_at)`, falling back to `now` on an empty
+    // recorded turn and sizes its buckets adaptively from that span.
-    // table) and sizes its buckets adaptively from that span.
     let (from, bucket_secs) = match window {
         Window::All => {
             let min_ts: Option<i64> =
@@ -291,51 +347,14 @@ fn snapshot(path: &Path, window: Window) -> Result<Snapshot> {
         })
     })?;
 
-    let mut by_bucket: HashMap<i64, BucketAcc> = HashMap::new();
+    let mut acc = TurnAccum::default();
-    let mut tool_totals: HashMap<String, u64> = HashMap::new();
-    let mut wake_totals: HashMap<String, u64> = HashMap::new();
-    let mut result_totals: HashMap<String, u64> = HashMap::new();
-    let mut model_set: HashSet<String> = HashSet::new();
-    let mut all_durations: Vec<i64> = Vec::new();
-    let mut turn_count: u64 = 0;
-
     for r in rows {
-        let r = r?;
+        acc.push(r?, bucket_secs);
-        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::<HashMap<String, u64>>(&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 buckets = fill_buckets(from, now, bucket_secs, &acc.by_bucket);
-    let duration_summary = summarize_durations(&mut all_durations);
+    let duration_summary = summarize_durations(&mut acc.all_durations);
-    let mut models: Vec<String> = model_set.into_iter().collect();
+    let mut models: Vec<String> = acc.model_set.into_iter().collect();
     models.sort_unstable();
 
     Ok(Snapshot {
@@ -343,19 +362,19 @@ fn snapshot(path: &Path, window: Window) -> Result<Snapshot> {
         bucket_seconds: bucket_secs,
         now,
         from,
-        turn_count,
+        turn_count: acc.turn_count,
         buckets,
-        tool_breakdown: top_n(tool_totals, 10),
+        tool_breakdown: top_n(acc.tool_totals, 10),
         bash_breakdown: read_bash_breakdown(&conn, from).unwrap_or_default(),
-        wake_mix: top_n(wake_totals, 20),
+        wake_mix: top_n(acc.wake_totals, 20),
-        result_mix: top_n(result_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 fresh session in the
+        // Inert-until-capture: `.ok()` maps both "no sessions in the window
-        // window yet" (QueryReturnedNoRows) and "sessions table absent on
+        // yet" and "sessions table absent on an older db" to None.
-        // an older db" (Err) to None, same decoupling as read_bash_breakdown.
         first_turn_ctx: read_first_turn_ctx(&conn, from).ok(),
+        session_count: read_session_count(&conn, from).ok(),
     })
 }
 
@@ -382,6 +401,23 @@ fn read_first_turn_ctx(conn: &Connection, from: i64) -> rusqlite::Result<u64> {
     )
 }
 
+/// 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.