feat(#1974): migrate-stats — add session.id/start_type/terminal.type + session.count (review)

hive-c0re/src/otel_migrate.rs

@@ -1,18 +1,27 @@
 //! One-shot backfill of historical per-agent turn stats into the OTEL
-//! collector as `claude_code.token.usage` metrics, so pre-OTEL history
+//! collector, so pre-OTEL history shows up alongside the live telemetry
-//! shows up alongside the live telemetry stream. Driven by
+//! stream. Driven by `hivectl migrate-stats`.
-//! `hivectl migrate-stats`.
 //!
 //! Reads each agent's `hyperhive-turn-stats.sqlite` (host path via
-//! `Coordinator::agent_harness_dir`) and turns the four per-turn token
+//! `Coordinator::agent_harness_dir`) and emits two **cumulative** counter
-//! columns into **cumulative** `claude_code.token.usage` counter samples:
+//! metrics, matching the live Claude Code shape so the backfill merges
-//! one series per `(type, model)`, value = the running total at each
+//! into the same series:
-//! turn's end time. Cumulative — not delta — because Prometheus/Mimir
+//!
-//! family backends silently drop delta sums (the same reason the live
+//! - `claude_code.token.usage` — the four per-turn token columns, one
-//! export forces cumulative temporality in harness-base.nix). The samples
+//!   running-total series per `(type, model, session.id)`, sampled at each
-//! are sent as OTLP/HTTP JSON to `<endpoint>/v1/metrics`. Every
+//!   turn's end time.
-//! datapoint carries `hyperhive-migration="true"` so the backfill is
+//! - `claude_code.session.count` — one increment per turn, split into
-//! distinguishable from the live stream.
+//!   `start_type` = `fresh` (a session's first turn) / `continue` (the
+//!   rest), mirroring the per-turn `--continue` model.
+//!
+//! Cumulative — not delta — because Prometheus/Mimir family backends
+//! silently drop delta sums (the same reason the live export forces
+//! cumulative temporality in harness-base.nix). Sent as OTLP/HTTP JSON to
+//! `<endpoint>/v1/metrics`. Every datapoint carries
+//! `hyperhive-migration="true"` (so the backfill is distinguishable from
+//! the live stream) and `terminal.type="non-interactive"`. `session.id` is
+//! the db's own session id verbatim when present, else a synthetic
+//! reproducible `<swarm>-<hive>-<agent>-migration-<n>`.
 //!
 //! Timestamps: `turn_stats.{started_at,ended_at}` are epoch *seconds*
 //! (see `hive-ag3nt` turn.rs), scaled to nanoseconds for OTLP.
@@ -20,12 +29,14 @@
 use anyhow::{Context, Result};
 use rusqlite::Connection;
 use serde_json::{Value, json};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::path::Path;
 
 use crate::coordinator::Coordinator;
 
-const METRIC: &str = "claude_code.token.usage";
+const TOKEN_METRIC: &str = "claude_code.token.usage";
+const SESSION_METRIC: &str = "claude_code.session.count";
+const TERMINAL_TYPE: &str = "non-interactive";
 
 /// Cap datapoints per POST so a long-lived agent's history is chunked
 /// into reasonably-sized OTLP requests rather than one huge body.
@@ -40,6 +51,12 @@ const TOKEN_TYPES: &[(&str, usize)] = &[
     ("cacheCreation", 6),
 ];
 
+/// Datapoints collected from one agent's stats db, split by metric.
+struct AgentPoints {
+    token: Vec<Value>,
+    session: Vec<Value>,
+}
+
 /// Run the migration. `endpoint` falls back to `OTEL_EXPORTER_OTLP_ENDPOINT`
 /// then `HYPERHIVE_OTEL_ENDPOINT`. `only_agent` limits to one agent.
 /// `dry_run` reports counts without sending.
@@ -74,23 +91,35 @@ pub async fn run(
         if !db.exists() {
             continue;
         }
-        let points =
+        let id_prefix = format!("{swarm}-{hive}-{agent}");
-            collect_agent_points(&db).with_context(|| format!("read turn_stats for {agent}"))?;
+        let pts = collect_agent_points(&db, &id_prefix)
-        if points.is_empty() {
+            .with_context(|| format!("read turn_stats for {agent}"))?;
+        let n = pts.token.len() + pts.session.len();
+        if n == 0 {
             println!("{agent}: no token rows, skipping");
             continue;
         }
         hit_agents += 1;
-        total_points += points.len();
+        total_points += n;
-        println!("{agent}: {} datapoints", points.len());
+        println!(
+            "{agent}: {} token + {} session datapoints",
+            pts.token.len(),
+            pts.session.len()
+        );
         if dry_run {
             continue;
         }
-        for chunk in points.chunks(MAX_DATAPOINTS_PER_POST) {
+        for chunk in pts.token.chunks(MAX_DATAPOINTS_PER_POST) {
-            let payload = build_payload(agent, &hive, &swarm, chunk);
+            let payload = build_payload(agent, &hive, &swarm, TOKEN_METRIC, "tokens", chunk);
             post(&client, &url, &payload)
                 .await
-                .with_context(|| format!("POST metrics for {agent}"))?;
+                .with_context(|| format!("POST token.usage for {agent}"))?;
+        }
+        for chunk in pts.session.chunks(MAX_DATAPOINTS_PER_POST) {
+            let payload = build_payload(agent, &hive, &swarm, SESSION_METRIC, "", chunk);
+            post(&client, &url, &payload)
+                .await
+                .with_context(|| format!("POST session.count for {agent}"))?;
         }
     }
     println!(
@@ -104,15 +133,16 @@ pub async fn run(
     Ok(())
 }
 
-/// Build cumulative `claude_code.token.usage` datapoints for one agent's
+/// Build cumulative datapoints for one agent's stats db. `id_prefix` is
-/// stats db. One running-total series per `(type, model)`; a datapoint is
+/// `<swarm>-<hive>-<agent>`, used to synthesise a `session.id` for rows
-/// emitted whenever that series' value changes (token delta > 0).
+/// whose `session_id` FK is NULL (pre-session-tracking turns).
-fn collect_agent_points(db: &Path) -> Result<Vec<Value>> {
+fn collect_agent_points(db: &Path, id_prefix: &str) -> Result<AgentPoints> {
     let conn = Connection::open_with_flags(db, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)
         .context("open stats db read-only")?;
+    // session_id last so the token column indices (3..=6) stay stable.
     let mut stmt = conn.prepare(
         "SELECT started_at, ended_at, model, input_tokens, output_tokens, \
-         cache_read_input_tokens, cache_creation_input_tokens \
+         cache_read_input_tokens, cache_creation_input_tokens, session_id \
          FROM turn_stats ORDER BY ended_at ASC",
     )?;
     let rows = stmt.query_map([], |r| {
@@ -124,45 +154,91 @@ fn collect_agent_points(db: &Path) -> Result<Vec<Value>> {
             r.get::<_, i64>(4)?,
             r.get::<_, i64>(5)?,
             r.get::<_, i64>(6)?,
+            r.get::<_, Option<i64>>(7)?,
         ))
     })?;
 
-    // (type, model) -> (running total, series start time in nanos string)
+    // token.usage: (type, model, session.id) -> (running total, series start nanos).
-    let mut series: HashMap<(&'static str, String), (u64, String)> = HashMap::new();
+    let mut token_series: HashMap<(&'static str, String, String), (u64, String)> = HashMap::new();
-    let mut points: Vec<Value> = Vec::new();
+    // session.count: start_type -> (running count, series start nanos).
+    let mut session_counts: HashMap<&'static str, (u64, String)> = HashMap::new();
+    let mut seen_sessions: HashSet<i64> = HashSet::new();
+    let mut migration_idx: u64 = 0;
+    let mut token = Vec::new();
+    let mut session = Vec::new();
+
     for row in rows {
-        let (started, ended, model, inp, out, cache_read, cache_create) = row?;
+        let (started, ended, model, inp, out, cache_read, cache_create, sid) = row?;
         let cols = [inp, out, cache_read, cache_create];
         let start_nanos = u64::try_from(started).unwrap_or(0) * 1_000_000_000;
         let time_nanos = (u64::try_from(ended).unwrap_or(0) * 1_000_000_000).to_string();
+
+        // session.id (verbatim db id, else synthetic) + start_type (a
+        // session's first turn is fresh; every later turn is continue).
+        let (session_id, is_fresh) = if let Some(id) = sid {
+            (id.to_string(), seen_sessions.insert(id))
+        } else {
+            let s = format!("{id_prefix}-migration-{migration_idx}");
+            migration_idx += 1;
+            (s, true)
+        };
+        let start_type = if is_fresh { "fresh" } else { "continue" };
+
+        // session.count: +1 this turn on the start_type series.
+        let scount = session_counts
+            .entry(start_type)
+            .or_insert_with(|| (0u64, start_nanos.to_string()));
+        scount.0 += 1;
+        session.push(json!({
+            "asInt": scount.0.to_string(),
+            "startTimeUnixNano": scount.1.clone(),
+            "timeUnixNano": time_nanos,
+            "attributes": [
+                {"key": "start_type", "value": {"stringValue": start_type}},
+                {"key": "terminal.type", "value": {"stringValue": TERMINAL_TYPE}},
+                {"key": "hyperhive-migration", "value": {"stringValue": "true"}},
+            ],
+        }));
+
+        // token.usage: per (type, model, session.id) running total.
         for &(type_, idx) in TOKEN_TYPES {
             let val = cols[idx - 3];
             if val <= 0 {
                 continue;
             }
-            let entry = series
+            let entry = token_series
-                .entry((type_, model.clone()))
+                .entry((type_, model.clone(), session_id.clone()))
                 .or_insert_with(|| (0u64, start_nanos.to_string()));
             entry.0 += u64::try_from(val).unwrap_or(0);
-            points.push(json!({
+            token.push(json!({
                 "asInt": entry.0.to_string(),
                 "startTimeUnixNano": entry.1.clone(),
                 "timeUnixNano": time_nanos,
                 "attributes": [
                     {"key": "type", "value": {"stringValue": type_}},
                     {"key": "model", "value": {"stringValue": model.clone()}},
+                    {"key": "session.id", "value": {"stringValue": session_id.clone()}},
+                    {"key": "terminal.type", "value": {"stringValue": TERMINAL_TYPE}},
                     {"key": "hyperhive-migration", "value": {"stringValue": "true"}},
                 ],
             }));
         }
     }
-    Ok(points)
+    Ok(AgentPoints { token, session })
 }
 
-/// Wrap a chunk of datapoints in an OTLP/HTTP JSON `ExportMetricsServiceRequest`.
+/// Wrap a chunk of datapoints in an OTLP/HTTP JSON `ExportMetricsServiceRequest`
-/// Resource attributes mirror the live export (`hive-serve-otel`): a fixed
+/// for one cumulative monotonic Sum metric. Resource attributes mirror the
-/// `service.name` plus this agent's name and the hive/swarm names.
+/// live export (`hive-serve-otel`): a fixed `service.name` plus this agent's
-fn build_payload(agent: &str, hive: &str, swarm: &str, points: &[Value]) -> Value {
+/// name and the hive/swarm names.
+fn build_payload(
+    agent: &str,
+    hive: &str,
+    swarm: &str,
+    metric: &str,
+    unit: &str,
+    points: &[Value],
+) -> Value {
     json!({
         "resourceMetrics": [{
             "resource": {"attributes": [
@@ -174,8 +250,8 @@ fn build_payload(agent: &str, hive: &str, swarm: &str, points: &[Value]) -> Valu
             "scopeMetrics": [{
                 "scope": {"name": "hyperhive-migration"},
                 "metrics": [{
-                    "name": METRIC,
+                    "name": metric,
-                    "unit": "tokens",
+                    "unit": unit,
                     "sum": {
                         // 2 = AGGREGATION_TEMPORALITY_CUMULATIVE
                         "aggregationTemporality": 2,