From c68706e899e81a9e7cb82d8f925e764e6eb8e41d Mon Sep 17 00:00:00 2001
From: damocles <damocles@hyperhive.local>
Date: Wed, 24 Jun 2026 21:33:25 +0200
Subject: [PATCH] fix(#1982): migrate-stats tolerates pre-session_id dbs +
 skips failed agents

---
 hive-c0re/src/otel_migrate.rs | 39 ++++++++++++++++++++++++++++-------
 1 file changed, 32 insertions(+), 7 deletions(-)

diff --git a/hive-c0re/src/otel_migrate.rs b/hive-c0re/src/otel_migrate.rs
index 0fb8d687..2a8acb78 100644
--- a/hive-c0re/src/otel_migrate.rs
+++ b/hive-c0re/src/otel_migrate.rs
@@ -86,14 +86,23 @@ pub async fn run(
     let client = reqwest::Client::new();
     let mut total_points = 0usize;
     let mut hit_agents = 0usize;
+    let mut failed = 0usize;
     for agent in &agents {
         let db = Coordinator::agent_harness_dir(agent).join("hyperhive-turn-stats.sqlite");
         if !db.exists() {
             continue;
         }
         let id_prefix = format!("{swarm}-{hive}-{agent}");
-        let pts = collect_agent_points(&db, &id_prefix)
-            .with_context(|| format!("read turn_stats for {agent}"))?;
+        // Skip + warn on a bad db (e.g. an old schema) rather than aborting
+        // the whole migration on the first failed agent.
+        let pts = match collect_agent_points(&db, &id_prefix) {
+            Ok(p) => p,
+            Err(e) => {
+                eprintln!("{agent}: skipped — {e:#}");
+                failed += 1;
+                continue;
+            }
+        };
         let n = pts.token.len() + pts.session.len();
         if n == 0 {
             println!("{agent}: no token rows, skipping");
@@ -123,11 +132,16 @@ pub async fn run(
         }
     }
     println!(
-        "migrate-stats: {hit_agents} agent(s) with stats, {total_points} datapoints {}",
+        "migrate-stats: {hit_agents} agent(s) with stats, {total_points} datapoints {}{}",
         if dry_run {
             "(dry-run, nothing sent)"
         } else {
             "sent"
+        },
+        if failed > 0 {
+            format!(" ({failed} agent(s) skipped on error)")
+        } else {
+            String::new()
         }
     );
     Ok(())
@@ -139,12 +153,23 @@ pub async fn run(
 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(
+    // Older dbs predate the `session_id` migration. Probe for the column and
+    // fall back to a `NULL AS session_id` placeholder so the row shape (col 7)
+    // is identical either way (those rows get a synthetic session.id below).
+    // session_id is selected last so the token column indices stay fixed.
+    let has_session = conn
+        .prepare("SELECT session_id FROM turn_stats LIMIT 1")
+        .is_ok();
+    let sql = if has_session {
         "SELECT started_at, ended_at, model, input_tokens, output_tokens, \
          cache_read_input_tokens, cache_creation_input_tokens, session_id \
-         FROM turn_stats ORDER BY ended_at ASC",
-    )?;
+         FROM turn_stats ORDER BY ended_at ASC"
+    } else {
+        "SELECT started_at, ended_at, model, input_tokens, output_tokens, \
+         cache_read_input_tokens, cache_creation_input_tokens, NULL AS session_id \
+         FROM turn_stats ORDER BY ended_at ASC"
+    };
+    let mut stmt = conn.prepare(sql)?;
     let rows = stmt.query_map([], |r| {
         Ok((
             r.get::<_, i64>(0)?,