1use std::collections::HashMap;
32use std::path::{Path, PathBuf};
33
34use rusqlite::{Connection, OpenFlags};
35
36use super::schema::{NamedAgentRecord, NamedAgentRecordV1, MAX_SUPPORTED_SCHEMA};
37use super::store::{Registry, RegistryError};
38
39#[derive(Debug, Default, Clone, Copy)]
41pub struct MigrateStats {
42 pub dbs_scanned: usize,
45 pub dbs_skipped: usize,
49 pub rows_seen: usize,
50 pub records_written: usize,
51 pub records_skipped_existing: usize,
52 pub records_skipped_unmappable: usize,
53 pub complete: bool,
61}
62
63const MARKER: &str = ".migrated_from_sqlite";
66
67const MIGRATION_VERSION: u32 = 2;
77
78struct SqliteSource {
83 db_path: PathBuf,
84 agents_root: PathBuf,
85}
86
87pub fn migrate_from_sqlite_once(
96 home: &Path,
97 registry: &Registry,
98) -> Result<MigrateStats, RegistryError> {
99 let marker_path = registry.root().join(MARKER);
100 if marker_migration_version(&marker_path) >= MIGRATION_VERSION {
101 return Ok(MigrateStats {
106 complete: true,
107 ..MigrateStats::default()
108 });
109 }
110
111 let mut stats = MigrateStats::default();
112 let global_agents_root = home.join("agents");
126 let (sources, enum_incomplete) = enumerate_sources(home);
127
128 let mut latest_by_id: HashMap<String, RowSnapshot> = HashMap::new();
129 let mut any_db_failed = enum_incomplete;
134
135 for src in sources {
136 stats.dbs_scanned += 1;
137 match read_named_rows(&src.db_path, &src.agents_root) {
138 Ok(rows) => {
139 for row in rows {
140 stats.rows_seen += 1;
141 let key = row.id.clone();
142 match latest_by_id.get_mut(&key) {
143 Some(existing) if existing.started_at >= row.started_at => {
144 existing.display_hidden =
148 existing.display_hidden || row.display_hidden;
149 }
150 Some(existing) => {
151 let merged_hidden =
152 existing.display_hidden || row.display_hidden;
153 *existing = row;
154 existing.display_hidden = merged_hidden;
155 }
156 None => {
157 latest_by_id.insert(key, row);
158 }
159 }
160 }
161 }
162 Err(e) => {
163 tracing::warn!(
164 db = %src.db_path.display(),
165 error = %e,
166 "registry-migrate: DB unreadable — skipping this source; registry still attaches, marker deferred to retry"
167 );
168 stats.dbs_skipped += 1;
169 any_db_failed = true;
170 }
171 }
172 }
173
174 for (id, row) in latest_by_id {
175 if registry.exists_anywhere(&id) {
179 stats.records_skipped_existing += 1;
180 continue;
181 }
182 let display_hidden = row.display_hidden;
183 let Some(rec) = row_to_record(&row, &global_agents_root) else {
184 stats.records_skipped_unmappable += 1;
185 continue;
186 };
187 if let Err(e) = registry.upsert(&rec) {
188 tracing::warn!(
189 instance_id = %id,
190 error = %e,
191 "registry-migrate: upsert failed"
192 );
193 stats.records_skipped_unmappable += 1;
194 continue;
195 }
196 if display_hidden {
200 if let Err(e) = registry.retire(&id) {
201 tracing::warn!(
202 instance_id = %id,
203 error = %e,
204 "registry-migrate: failed to retire migrated tombstone — record may surface as active"
205 );
206 }
207 }
208 stats.records_written += 1;
209 }
210
211 stats.complete = !any_db_failed;
217 if stats.complete {
218 write_marker(&marker_path, &stats)?;
219 } else {
220 tracing::info!(
221 "registry-migrate: deferring marker write; one or more DBs were unreadable and will be retried next launch"
222 );
223 }
224 Ok(stats)
225}
226
227const SOURCE_BACKFILL_MARKER: &str = ".backfilled_source_bases";
231
232#[derive(Debug, Default, Clone, Copy)]
234pub struct SourceBackfillStats {
235 pub dbs_scanned: usize,
236 pub records_updated: usize,
237 pub records_unresolved: usize,
241 pub complete: bool,
242}
243
244pub fn backfill_source_bases_once(
258 home: &Path,
259 registry: &Registry,
260) -> Result<SourceBackfillStats, RegistryError> {
261 let marker = registry.root().join(SOURCE_BACKFILL_MARKER);
262 if marker.exists() {
263 return Ok(SourceBackfillStats {
264 complete: true,
265 ..Default::default()
266 });
267 }
268
269 let mut stats = SourceBackfillStats::default();
270
271 let mut pending: HashMap<String, NamedAgentRecord> = registry
276 .list_active()?
277 .into_iter()
278 .filter(|r| r.data.source_agents_base.is_none())
279 .map(|r| (r.data.instance_id.clone(), r))
280 .collect();
281
282 if pending.is_empty() {
283 std::fs::write(&marker, b"backfilled: 0\n")?;
285 stats.complete = true;
286 return Ok(stats);
287 }
288
289 let (sources, mut incomplete) = enumerate_sources(home);
290
291 let mut winners: HashMap<String, (i64, PathBuf)> = HashMap::new();
300 for src in sources {
301 stats.dbs_scanned += 1;
302 match read_named_rows(&src.db_path, &src.agents_root) {
303 Ok(rows) => {
304 for row in rows {
305 if !pending.contains_key(&row.id) {
306 continue;
307 }
308 match winners.get(&row.id) {
309 Some((ts, _)) if *ts >= row.started_at => {}
310 _ => {
311 winners.insert(row.id.clone(), (row.started_at, row.agents_root));
312 }
313 }
314 }
315 }
316 Err(e) => {
317 tracing::warn!(
318 db = %src.db_path.display(),
319 error = %e,
320 "source-base backfill: DB unreadable — will retry next launch"
321 );
322 incomplete = true;
323 }
324 }
325 }
326
327 for (id, (_, agents_root)) in winners {
330 let Some(mut rec) = pending.remove(&id) else {
331 continue;
332 };
333 rec.data.source_agents_base = Some(agents_root.to_string_lossy().to_string());
334 rec.schema_version = rec.data.min_schema_version();
335 if let Err(e) = registry.upsert(&rec) {
336 tracing::warn!(
337 instance_id = %id,
338 error = %e,
339 "source-base backfill: upsert failed — will retry next launch"
340 );
341 pending.insert(id, rec);
342 incomplete = true;
343 } else {
344 stats.records_updated += 1;
345 }
346 }
347
348 stats.records_unresolved = pending.len();
351
352 stats.complete = !incomplete;
355 if stats.complete {
356 std::fs::write(
357 &marker,
358 format!(
359 "backfilled: {}\nunresolved: {}\n",
360 stats.records_updated, stats.records_unresolved
361 ),
362 )?;
363 }
364 Ok(stats)
365}
366
367pub fn enumerate_objects_dbs(home: &Path) -> Vec<PathBuf> {
371 let (sources, _) = enumerate_sources(home);
372 sources.into_iter().map(|s| s.db_path).collect()
373}
374
375fn enumerate_sources(home: &Path) -> (Vec<SqliteSource>, bool) {
387 let mut out = Vec::new();
388 let mut incomplete = false;
389
390 if let Some(rd) = read_dir_tracking(&home.join("channels"), &mut incomplete) {
394 for ch in rd.flatten() {
395 let ch_dir = ch.path();
396 if !ch_dir.is_dir() {
397 continue;
398 }
399 let agents_root = ch_dir.join("agents");
400 if let Some(vrd) = read_dir_tracking(&ch_dir.join("versions"), &mut incomplete) {
401 for v in vrd.flatten() {
402 let db = v.path().join("data").join("db").join("objects.db");
403 if db.is_file() {
404 out.push(SqliteSource {
405 db_path: db,
406 agents_root: agents_root.clone(),
407 });
408 }
409 }
410 }
411 }
412 }
413
414 collect_dev_sources(&home.join("dev"), &mut out, &mut incomplete);
418
419 (out, incomplete)
420}
421
422fn read_dir_tracking(path: &Path, incomplete: &mut bool) -> Option<std::fs::ReadDir> {
426 match std::fs::read_dir(path) {
427 Ok(rd) => Some(rd),
428 Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
429 Err(_) => {
430 *incomplete = true;
431 None
432 }
433 }
434}
435
436fn collect_dev_sources(dev_root: &Path, out: &mut Vec<SqliteSource>, incomplete: &mut bool) {
446 let Some(branches) = read_dir_tracking(dev_root, incomplete) else {
447 return;
448 };
449 for branch in branches.flatten() {
450 let bdir = branch.path();
451 if !bdir.is_dir() {
452 continue;
453 }
454 if push_if_instance(&bdir, out) {
457 continue;
458 }
459 if let Some(subs) = read_dir_tracking(&bdir, incomplete) {
461 for sub in subs.flatten() {
462 let sdir = sub.path();
463 if sdir.is_dir() {
464 push_if_instance(&sdir, out);
465 }
466 }
467 }
468 }
469}
470
471fn push_if_instance(dir: &Path, out: &mut Vec<SqliteSource>) -> bool {
474 let db = dir.join("data").join("db").join("objects.db");
475 if db.is_file() {
476 out.push(SqliteSource {
477 db_path: db,
478 agents_root: dir.join("agents"),
479 });
480 true
481 } else {
482 false
483 }
484}
485
486fn marker_migration_version(path: &Path) -> u32 {
491 let Ok(body) = std::fs::read_to_string(path) else {
492 return 0;
493 };
494 for line in body.lines() {
495 if let Some(v) = line.strip_prefix("migration_version:") {
496 return v.trim().parse().unwrap_or(0);
497 }
498 }
499 0
500}
501
502fn write_marker(path: &Path, stats: &MigrateStats) -> std::io::Result<()> {
503 let now = chrono::Utc::now().to_rfc3339();
504 let body = format!(
505 "migration_version: {MIGRATION_VERSION}\n\
506 migrated_at: {now}\n\
507 dbs_scanned: {}\n\
508 dbs_skipped: {}\n\
509 rows_seen: {}\n\
510 records_written: {}\n\
511 records_skipped_existing: {}\n\
512 records_skipped_unmappable: {}\n",
513 stats.dbs_scanned,
514 stats.dbs_skipped,
515 stats.rows_seen,
516 stats.records_written,
517 stats.records_skipped_existing,
518 stats.records_skipped_unmappable,
519 );
520 std::fs::write(path, body)
521}
522
523struct RowSnapshot {
524 id: String,
525 instance_name: String,
526 definition_id: String,
527 identity_id: String,
528 memory_id: String,
529 working_directory: String,
530 agents_root: PathBuf,
537 started_at: i64,
538 created_at: i64,
539 display_hidden: bool,
540}
541
542fn is_missing_column_or_table(e: &rusqlite::Error) -> bool {
553 let msg = match e {
554 rusqlite::Error::SqliteFailure(_, Some(msg)) => msg.as_str(),
555 rusqlite::Error::SqlInputError { msg, .. } => msg.as_str(),
556 _ => return false,
557 };
558 msg.starts_with("no such column") || msg.starts_with("no such table")
559}
560
561fn read_named_rows(
562 db_path: &Path,
563 agents_root: &Path,
564) -> Result<Vec<RowSnapshot>, rusqlite::Error> {
565 let conn = Connection::open_with_flags(
566 db_path,
567 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
568 )?;
569 let mut stmt = match conn.prepare(
577 "SELECT id, instance_name, definition_id, identity_id, memory_id,
578 working_directory, started_at, created_at, display_hidden
579 FROM db_agent_instances
580 WHERE instance_name <> ''
581 AND parent_instance_id = ''",
582 ) {
583 Ok(s) => s,
584 Err(e) if is_missing_column_or_table(&e) => return Ok(Vec::new()),
585 Err(e) => return Err(e),
586 };
587 let iter = stmt.query_map([], |row| {
588 Ok(RowSnapshot {
589 id: row.get(0)?,
590 instance_name: row.get(1)?,
591 definition_id: row.get(2)?,
592 identity_id: row.get(3)?,
593 memory_id: row.get(4)?,
594 working_directory: row.get(5)?,
595 agents_root: agents_root.to_path_buf(),
596 started_at: row.get(6)?,
597 created_at: row.get(7)?,
598 display_hidden: row.get::<_, i64>(8)? != 0,
599 })
600 })?;
601 iter.collect()
602}
603
604fn row_to_record(row: &RowSnapshot, global_agents_root: &Path) -> Option<NamedAgentRecord> {
605 let abs = std::path::Path::new(&row.working_directory);
606 let (rel, base): (&Path, &Path) = abs
615 .strip_prefix(global_agents_root)
616 .ok()
617 .map(|r| (r, global_agents_root))
618 .or_else(|| {
619 abs.strip_prefix(row.agents_root.as_path())
620 .ok()
621 .map(|r| (r, row.agents_root.as_path()))
622 })?;
623 let rel_str = rel.to_string_lossy().to_string();
624 if rel_str.is_empty() || rel_str == "." {
625 return None;
626 }
627 let data = NamedAgentRecordV1 {
628 instance_id: row.id.clone(),
629 instance_name: row.instance_name.clone(),
630 definition_id: row.definition_id.clone(),
631 identity_id: empty_to_none(&row.identity_id),
632 memory_id: empty_to_none(&row.memory_id),
633 session_id: None,
639 working_dir: rel_str,
640 source_agents_base: Some(base.to_string_lossy().to_string()),
645 created_at_ms: row.created_at,
646 last_launched_at_ms: row.started_at,
647 created_by_version: "(legacy)".to_string(),
651 last_launched_by_version: "(legacy)".to_string(),
652 };
653 Some(NamedAgentRecord {
654 schema_version: data.min_schema_version(),
655 data,
656 })
657}
658
659fn empty_to_none(s: &str) -> Option<String> {
660 if s.is_empty() {
661 None
662 } else {
663 Some(s.to_string())
664 }
665}
666
667#[cfg(test)]
668mod tests {
669 use super::*;
670 use rusqlite::params;
671
672 fn make_db_at(version_dir: &Path, rows: &[(&str, &str, i64, &str, bool)]) {
675 let db_dir = version_dir.join("data").join("db");
676 std::fs::create_dir_all(&db_dir).unwrap();
677 let conn = Connection::open(db_dir.join("objects.db")).unwrap();
678 conn.execute_batch(
679 "CREATE TABLE db_agent_instances (
680 id TEXT PRIMARY KEY,
681 definition_id TEXT NOT NULL DEFAULT '',
682 parent_instance_id TEXT NOT NULL DEFAULT '',
683 block_id TEXT NOT NULL DEFAULT '',
684 session_id TEXT NOT NULL DEFAULT '',
685 status TEXT NOT NULL DEFAULT 'running',
686 github_context TEXT NOT NULL DEFAULT '',
687 started_at INTEGER NOT NULL DEFAULT 0,
688 ended_at INTEGER NOT NULL DEFAULT 0,
689 created_at INTEGER NOT NULL DEFAULT 0,
690 identity_id TEXT NOT NULL DEFAULT '',
691 memory_id TEXT NOT NULL DEFAULT '',
692 instance_name TEXT NOT NULL DEFAULT '',
693 working_directory TEXT NOT NULL DEFAULT '',
694 display_hidden INTEGER NOT NULL DEFAULT 0
695 );",
696 )
697 .unwrap();
698 for (id, name, started_at, working_directory, hidden) in rows {
699 conn.execute(
700 "INSERT INTO db_agent_instances
701 (id, definition_id, instance_name, working_directory, started_at, created_at, display_hidden)
702 VALUES (?1, 'claude-code', ?2, ?3, ?4, ?4, ?5)",
703 params![id, name, working_directory, started_at, if *hidden { 1_i64 } else { 0_i64 }],
704 )
705 .unwrap();
706 }
707 }
708
709 fn channel_agents(home: &Path, channel: &str) -> PathBuf {
711 home.join("channels").join(channel).join("agents")
712 }
713
714 fn make_channel_db(
716 home: &Path,
717 channel: &str,
718 version: &str,
719 rows: &[(&str, &str, i64, &str)],
720 ) {
721 let rows: Vec<_> = rows.iter().map(|(a, b, c, d)| (*a, *b, *c, *d, false)).collect();
722 let v_dir = home
723 .join("channels")
724 .join(channel)
725 .join("versions")
726 .join(version);
727 make_db_at(&v_dir, &rows);
728 }
729
730 fn make_channel_db_with_hidden(
731 home: &Path,
732 channel: &str,
733 version: &str,
734 rows: &[(&str, &str, i64, &str, bool)],
735 ) {
736 let v_dir = home
737 .join("channels")
738 .join(channel)
739 .join("versions")
740 .join(version);
741 make_db_at(&v_dir, rows);
742 }
743
744 fn fresh_home() -> (tempfile::TempDir, Registry) {
745 let home = tempfile::tempdir().unwrap();
746 let reg = Registry::open(
748 home.path()
749 .join("shared")
750 .join("agents")
751 .join("registry"),
752 )
753 .unwrap();
754 (home, reg)
755 }
756
757 #[test]
758 fn migrate_with_no_channels_writes_marker_and_no_rows() {
759 let (home, reg) = fresh_home();
760 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
761 assert_eq!(stats.dbs_scanned, 0);
762 assert_eq!(stats.records_written, 0);
763 assert!(reg.root().join(MARKER).exists());
764 }
765
766 #[test]
767 fn migrate_is_idempotent() {
768 let (home, reg) = fresh_home();
769 migrate_from_sqlite_once(home.path(), ®).unwrap();
771 let wd = channel_agents(home.path(), "stable").join("demo-1");
773 std::fs::create_dir_all(&wd).unwrap();
774 make_channel_db(
775 home.path(),
776 "stable",
777 "0.33.821",
778 &[("inst-1", "demo", 100, &wd.to_string_lossy())],
779 );
780 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
781 assert_eq!(
782 stats.records_written, 0,
783 "marker must short-circuit subsequent runs"
784 );
785 assert!(reg.list_active().unwrap().is_empty());
786 }
787
788 #[test]
789 fn migrate_writes_one_record_per_unique_id() {
790 let (home, reg) = fresh_home();
791 let agents = channel_agents(home.path(), "stable");
792 let wd_a = agents.join("demo-a");
793 let wd_b = agents.join("demo-b");
794 std::fs::create_dir_all(&wd_a).unwrap();
795 std::fs::create_dir_all(&wd_b).unwrap();
796 make_channel_db(
797 home.path(),
798 "stable",
799 "0.33.821",
800 &[
801 ("inst-a", "demoA", 100, &wd_a.to_string_lossy()),
802 ("inst-b", "demoB", 200, &wd_b.to_string_lossy()),
803 ],
804 );
805
806 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
807 assert_eq!(stats.rows_seen, 2);
808 assert_eq!(stats.records_written, 2);
809 assert_eq!(reg.list_active().unwrap().len(), 2);
810 }
811
812 #[test]
813 fn migrate_anchors_each_row_on_its_own_channel() {
814 let (home, reg) = fresh_home();
821 let wd_a = channel_agents(home.path(), "stable").join("alpha");
822 let wd_b = channel_agents(home.path(), "local-main-b28b7a").join("beta");
823 std::fs::create_dir_all(&wd_a).unwrap();
824 std::fs::create_dir_all(&wd_b).unwrap();
825 make_channel_db(
826 home.path(),
827 "stable",
828 "0.44.2",
829 &[("inst-a", "Alpha", 100, &wd_a.to_string_lossy())],
830 );
831 make_channel_db(
832 home.path(),
833 "local-main-b28b7a",
834 "0.44.2",
835 &[("inst-b", "Beta", 200, &wd_b.to_string_lossy())],
836 );
837
838 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
839 assert_eq!(stats.dbs_scanned, 2);
840 assert_eq!(stats.rows_seen, 2);
841 assert_eq!(stats.records_skipped_unmappable, 0, "per-channel anchoring");
842 assert_eq!(stats.records_written, 2);
843 let mut recs = reg.list_active().unwrap();
844 recs.sort_by(|a, b| a.data.instance_id.cmp(&b.data.instance_id));
845 assert_eq!(recs[0].data.working_dir, "alpha");
846 assert_eq!(recs[1].data.working_dir, "beta");
847 assert_eq!(
851 recs[0].data.source_agents_base.as_deref(),
852 Some(channel_agents(home.path(), "stable").to_string_lossy().as_ref())
853 );
854 assert_eq!(
855 recs[1].data.source_agents_base.as_deref(),
856 Some(
857 channel_agents(home.path(), "local-main-b28b7a")
858 .to_string_lossy()
859 .as_ref()
860 )
861 );
862 assert_eq!(recs[0].schema_version, 3);
863 }
864
865 #[test]
866 fn migrate_anchors_global_workspace_not_per_channel() {
867 let (home, reg) = fresh_home();
875 let global_agents = home.path().join("agents"); assert_ne!(
881 global_agents.as_path(),
882 reg.root().parent().unwrap(),
883 "workspace root must differ from the re-rooted registry's parent"
884 );
885 let wd = global_agents.join("qooma-0612g");
886 std::fs::create_dir_all(&wd).unwrap();
887 make_channel_db(
890 home.path(),
891 "stable",
892 "0.44.2",
893 &[("inst-q", "Qooma", 100, &wd.to_string_lossy())],
894 );
895
896 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
897 assert_eq!(stats.rows_seen, 1);
898 assert_eq!(
899 stats.records_skipped_unmappable, 0,
900 "a global workspace must NOT be unmappable"
901 );
902 assert_eq!(stats.records_written, 1);
903 let recs = reg.list_active().unwrap();
904 assert_eq!(recs.len(), 1);
905 assert_eq!(recs[0].data.working_dir, "qooma-0612g");
906 assert_eq!(
907 recs[0].data.source_agents_base.as_deref(),
908 Some(global_agents.to_string_lossy().as_ref()),
909 "anchored on the GLOBAL workspace root <home>/agents, not the channel"
910 );
911 assert_eq!(recs[0].schema_version, 3);
912 }
913
914 #[test]
915 fn migrate_legacy_marker_reruns_then_settles() {
916 let (home, reg) = fresh_home();
921 let global_agents = home.path().join("agents"); let wd = global_agents.join("naki");
923 std::fs::create_dir_all(&wd).unwrap();
924 make_channel_db(
925 home.path(),
926 "stable",
927 "0.44.2",
928 &[("inst-n", "Naki", 100, &wd.to_string_lossy())],
929 );
930 std::fs::write(
932 reg.root().join(MARKER),
933 b"migrated_at: 2026-06-10T00:00:00Z\nrecords_written: 0\n",
934 )
935 .unwrap();
936
937 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
939 assert_eq!(
940 stats.records_written, 1,
941 "legacy marker must trigger a one-time re-run"
942 );
943 assert_eq!(reg.list_active().unwrap().len(), 1);
944 assert_eq!(
945 marker_migration_version(®.root().join(MARKER)),
946 MIGRATION_VERSION,
947 "marker upgraded to the current version"
948 );
949
950 let again = migrate_from_sqlite_once(home.path(), ®).unwrap();
952 assert_eq!(again.records_written, 0, "settles after the single re-run");
953 }
954
955 #[test]
956 fn migrate_picks_latest_started_at_on_dedup() {
957 let (home, reg) = fresh_home();
958 let wd = channel_agents(home.path(), "stable").join("demo");
959 std::fs::create_dir_all(&wd).unwrap();
960 make_channel_db(
963 home.path(),
964 "stable",
965 "0.33.800",
966 &[("inst-1", "demo", 100, &wd.to_string_lossy())],
967 );
968 make_channel_db(
969 home.path(),
970 "stable",
971 "0.33.821",
972 &[("inst-1", "demo", 200, &wd.to_string_lossy())],
973 );
974
975 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
976 assert_eq!(stats.rows_seen, 2);
977 assert_eq!(stats.records_written, 1);
978 let recs = reg.list_active().unwrap();
979 assert_eq!(recs.len(), 1);
980 assert_eq!(recs[0].data.last_launched_at_ms, 200);
981 }
982
983 #[test]
984 fn migrate_handles_dev_layout() {
985 let (home, reg) = fresh_home();
990 let inst_dir = home.path().join("dev").join("mybranch").join("69d7a34a");
991 let wd = inst_dir.join("agents").join("devagent");
992 std::fs::create_dir_all(&wd).unwrap();
993 make_db_at(
994 &inst_dir,
995 &[("inst-dev", "DevAgent", 100, &wd.to_string_lossy(), false)],
996 );
997
998 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
999 assert_eq!(stats.dbs_scanned, 1);
1000 assert_eq!(stats.records_written, 1);
1001 let recs = reg.list_active().unwrap();
1002 assert_eq!(recs.len(), 1);
1003 assert_eq!(recs[0].data.working_dir, "devagent");
1004 }
1005
1006 #[test]
1007 fn migrate_dev_ignores_nested_agent_workspace_db() {
1008 let (home, reg) = fresh_home();
1013 let inst_dir = home.path().join("dev").join("mybranch").join("sub");
1014 let nested_inst = inst_dir.join("agents").join("nestedmux");
1016 let nested_wd = nested_inst.join("agents").join("inner");
1017 std::fs::create_dir_all(&nested_wd).unwrap();
1018 make_db_at(
1019 &nested_inst,
1020 &[("inst-nested", "Nested", 100, &nested_wd.to_string_lossy(), false)],
1021 );
1022
1023 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1024 assert_eq!(
1025 stats.dbs_scanned, 0,
1026 "must not descend into agent workspaces under dev/"
1027 );
1028 assert!(reg.list_active().unwrap().is_empty());
1029 }
1030
1031 #[test]
1032 fn migrate_dev_picks_instance_db_and_ignores_nested() {
1033 let (home, reg) = fresh_home();
1037 let inst_dir = home.path().join("dev").join("mybranch").join("sub");
1038 let wd = inst_dir.join("agents").join("realagent");
1039 std::fs::create_dir_all(&wd).unwrap();
1040 make_db_at(
1041 &inst_dir,
1042 &[("inst-real", "Real", 100, &wd.to_string_lossy(), false)],
1043 );
1044 let nested_inst = inst_dir.join("agents").join("nestedmux");
1045 let nested_wd = nested_inst.join("agents").join("inner");
1046 std::fs::create_dir_all(&nested_wd).unwrap();
1047 make_db_at(
1048 &nested_inst,
1049 &[("inst-nested", "Nested", 200, &nested_wd.to_string_lossy(), false)],
1050 );
1051
1052 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1053 assert_eq!(stats.dbs_scanned, 1, "only the instance-level dev DB is a source");
1054 let recs = reg.list_active().unwrap();
1055 assert_eq!(recs.len(), 1);
1056 assert_eq!(recs[0].data.instance_id, "inst-real");
1057 assert_eq!(recs[0].data.working_dir, "realagent");
1058 }
1059
1060 #[test]
1061 fn migrate_dev_branch_named_like_internal_dir_is_scanned() {
1062 let (home, reg) = fresh_home();
1066 let inst_dir = home.path().join("dev").join("data").join("sub");
1067 let wd = inst_dir.join("agents").join("a1");
1068 std::fs::create_dir_all(&wd).unwrap();
1069 make_db_at(
1070 &inst_dir,
1071 &[("inst-d", "D", 100, &wd.to_string_lossy(), false)],
1072 );
1073
1074 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1075 assert_eq!(stats.dbs_scanned, 1, "branch named 'data' must still be scanned");
1076 assert_eq!(reg.list_active().unwrap().len(), 1);
1077 }
1078
1079 #[test]
1080 fn migrate_defers_marker_when_versions_dir_is_unreadable() {
1081 let (home, reg) = fresh_home();
1086 let ch = home.path().join("channels").join("stable");
1087 std::fs::create_dir_all(&ch).unwrap();
1088 std::fs::write(ch.join("versions"), b"not a directory").unwrap();
1089
1090 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1091 assert!(
1092 !stats.complete,
1093 "an unreadable versions/ must defer the marker"
1094 );
1095 assert!(
1096 !reg.root().join(MARKER).exists(),
1097 "marker deferred so the channel is retried next launch"
1098 );
1099 }
1100
1101 #[test]
1102 fn migrate_skips_when_registry_already_has_record() {
1103 let (home, reg) = fresh_home();
1104 let wd = channel_agents(home.path(), "stable").join("demo");
1105 std::fs::create_dir_all(&wd).unwrap();
1106 reg.upsert(&NamedAgentRecord {
1108 schema_version: MAX_SUPPORTED_SCHEMA,
1109 data: NamedAgentRecordV1 {
1110 instance_id: "inst-1".to_string(),
1111 instance_name: "preexisting".to_string(),
1112 definition_id: "claude-code".to_string(),
1113 identity_id: None,
1114 memory_id: None,
1115 session_id: None,
1116 working_dir: "demo".to_string(),
1117 source_agents_base: None,
1118 created_at_ms: 50,
1119 last_launched_at_ms: 500,
1120 created_by_version: "0.33.823".to_string(),
1121 last_launched_by_version: "0.33.823".to_string(),
1122 },
1123 })
1124 .unwrap();
1125 make_channel_db(
1127 home.path(),
1128 "stable",
1129 "0.33.821",
1130 &[("inst-1", "legacyname", 100, &wd.to_string_lossy())],
1131 );
1132
1133 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1134 assert_eq!(stats.records_skipped_existing, 1);
1135 assert_eq!(stats.records_written, 0);
1136 let recs = reg.list_active().unwrap();
1138 assert_eq!(recs.len(), 1);
1139 assert_eq!(recs[0].data.instance_name, "preexisting");
1140 }
1141
1142 #[test]
1143 fn migrate_skips_when_record_is_retired() {
1144 let (home, reg) = fresh_home();
1148 let wd = channel_agents(home.path(), "stable").join("demo");
1149 std::fs::create_dir_all(&wd).unwrap();
1150
1151 let retired_record = NamedAgentRecord {
1152 schema_version: MAX_SUPPORTED_SCHEMA,
1153 data: NamedAgentRecordV1 {
1154 instance_id: "inst-1".to_string(),
1155 instance_name: "demo".to_string(),
1156 definition_id: "claude-code".to_string(),
1157 identity_id: None,
1158 memory_id: None,
1159 session_id: None,
1160 working_dir: "demo".to_string(),
1161 source_agents_base: None,
1162 created_at_ms: 50,
1163 last_launched_at_ms: 50,
1164 created_by_version: "0.33.823".to_string(),
1165 last_launched_by_version: "0.33.823".to_string(),
1166 },
1167 };
1168 reg.upsert(&retired_record).unwrap();
1169 reg.retire("inst-1").unwrap();
1170 assert!(reg.list_active().unwrap().is_empty());
1171 assert!(reg.exists_anywhere("inst-1"));
1172
1173 make_channel_db(
1174 home.path(),
1175 "stable",
1176 "0.33.821",
1177 &[("inst-1", "demo", 100, &wd.to_string_lossy())],
1178 );
1179
1180 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1181 assert_eq!(stats.rows_seen, 1);
1182 assert_eq!(stats.records_skipped_existing, 1);
1183 assert_eq!(stats.records_written, 0);
1184 assert!(reg.list_active().unwrap().is_empty());
1185 assert!(reg.root().join("retired").join("inst-1.json").exists());
1186 }
1187
1188 #[test]
1189 fn migrate_silently_skips_pre_v8_schema() {
1190 let (home, reg) = fresh_home();
1195 let db_dir = home
1196 .path()
1197 .join("channels")
1198 .join("stable")
1199 .join("versions")
1200 .join("0.33.643")
1201 .join("data")
1202 .join("db");
1203 std::fs::create_dir_all(&db_dir).unwrap();
1204 let conn = Connection::open(db_dir.join("objects.db")).unwrap();
1205 conn.execute_batch(
1206 "CREATE TABLE db_agent_instances (
1207 id TEXT PRIMARY KEY,
1208 definition_id TEXT NOT NULL DEFAULT '',
1209 parent_instance_id TEXT NOT NULL DEFAULT '',
1210 started_at INTEGER NOT NULL DEFAULT 0,
1211 created_at INTEGER NOT NULL DEFAULT 0
1212 );",
1213 )
1214 .unwrap();
1215 drop(conn);
1216
1217 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1218 assert_eq!(stats.dbs_scanned, 1);
1219 assert_eq!(stats.rows_seen, 0);
1220 assert!(stats.complete, "pre-v8 schema must not block the marker");
1221 assert!(reg.root().join(MARKER).exists());
1222 }
1223
1224 #[test]
1225 fn migrate_writes_legacy_hidden_row_as_tombstone() {
1226 let (home, reg) = fresh_home();
1229 let wd = channel_agents(home.path(), "stable").join("forgotten");
1230 std::fs::create_dir_all(&wd).unwrap();
1231 make_channel_db_with_hidden(
1232 home.path(),
1233 "stable",
1234 "0.33.821",
1235 &[("inst-1", "forgotten", 100, &wd.to_string_lossy(), true)],
1236 );
1237
1238 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1239 assert_eq!(stats.records_written, 1);
1240 assert!(
1241 reg.list_active().unwrap().is_empty(),
1242 "hidden legacy row must NOT appear active"
1243 );
1244 assert!(
1245 reg.root().join("retired").join("inst-1.json").exists(),
1246 "hidden legacy row must be migrated as retired tombstone"
1247 );
1248 }
1249
1250 #[test]
1251 fn migrate_preserves_forget_intent_across_versions() {
1252 let (home, reg) = fresh_home();
1255 let wd = channel_agents(home.path(), "stable").join("toggled");
1256 std::fs::create_dir_all(&wd).unwrap();
1257 make_channel_db_with_hidden(
1258 home.path(),
1259 "stable",
1260 "0.33.800",
1261 &[("inst-1", "toggled", 100, &wd.to_string_lossy(), false)],
1262 );
1263 make_channel_db_with_hidden(
1264 home.path(),
1265 "stable",
1266 "0.33.821",
1267 &[("inst-1", "toggled", 200, &wd.to_string_lossy(), true)],
1268 );
1269
1270 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1271 assert_eq!(stats.records_written, 1);
1272 assert!(
1273 reg.list_active().unwrap().is_empty(),
1274 "hidden intent in any version must propagate to registry tombstone"
1275 );
1276 assert!(reg.root().join("retired").join("inst-1.json").exists());
1277 }
1278
1279 #[test]
1280 fn migrate_defers_marker_on_unreadable_db() {
1281 let (home, reg) = fresh_home();
1284 let wd = channel_agents(home.path(), "stable").join("demo");
1285 std::fs::create_dir_all(&wd).unwrap();
1286 make_channel_db(
1287 home.path(),
1288 "stable",
1289 "0.33.821",
1290 &[("inst-good", "demo", 100, &wd.to_string_lossy())],
1291 );
1292 let bad_db_dir = home
1294 .path()
1295 .join("channels")
1296 .join("stable")
1297 .join("versions")
1298 .join("0.33.800")
1299 .join("data")
1300 .join("db");
1301 std::fs::create_dir_all(&bad_db_dir).unwrap();
1302 std::fs::write(bad_db_dir.join("objects.db"), b"not actually sqlite").unwrap();
1303
1304 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1305 assert_eq!(stats.records_written, 1, "good DB still migrated");
1306 assert_eq!(stats.dbs_skipped, 1, "bad DB counted, not fatal");
1307 assert!(
1308 !reg.root().join(MARKER).exists(),
1309 "marker deferred (retry) when a DB was unreadable — but the registry still attaches (see main.rs); good records are already written"
1310 );
1311 assert_eq!(reg.list_active().unwrap().len(), 1);
1314
1315 std::fs::remove_file(bad_db_dir.join("objects.db")).unwrap();
1318 make_channel_db(
1319 home.path(),
1320 "stable",
1321 "0.33.800",
1322 &[("inst-other", "demo", 50, &wd.to_string_lossy())],
1323 );
1324 let stats2 = migrate_from_sqlite_once(home.path(), ®).unwrap();
1325 assert!(stats2.complete, "complete flag set on clean retry");
1326 assert!(
1327 reg.root().join(MARKER).exists(),
1328 "marker written on the retry once all DBs read successfully"
1329 );
1330 assert_eq!(stats2.records_skipped_existing, 1);
1331 assert_eq!(stats2.records_written, 1);
1332 }
1333
1334 #[test]
1335 fn migrate_skips_unmappable_working_dirs() {
1336 let (home, reg) = fresh_home();
1337 let outside = home.path().join("not_under_agents").join("foo");
1339 make_channel_db(
1340 home.path(),
1341 "stable",
1342 "0.33.821",
1343 &[("inst-x", "demo", 100, &outside.to_string_lossy())],
1344 );
1345
1346 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1347 assert_eq!(stats.rows_seen, 1);
1348 assert_eq!(stats.records_skipped_unmappable, 1);
1349 assert_eq!(stats.records_written, 0);
1350 }
1351
1352 #[test]
1353 fn migrate_tolerates_missing_or_corrupt_dbs() {
1354 let (home, reg) = fresh_home();
1355 std::fs::create_dir_all(
1357 home.path()
1358 .join("channels")
1359 .join("stable")
1360 .join("versions")
1361 .join("0.33.700"),
1362 )
1363 .unwrap();
1364 let db_dir = home
1366 .path()
1367 .join("channels")
1368 .join("stable")
1369 .join("versions")
1370 .join("0.33.701")
1371 .join("data")
1372 .join("db");
1373 std::fs::create_dir_all(&db_dir).unwrap();
1374 std::fs::write(db_dir.join("objects.db"), b"not a sqlite file").unwrap();
1375
1376 let stats = migrate_from_sqlite_once(home.path(), ®).unwrap();
1377 assert_eq!(stats.dbs_scanned, 1);
1380 assert_eq!(stats.dbs_skipped, 1);
1381 assert_eq!(stats.records_written, 0);
1382 assert!(
1383 !reg.root().join(MARKER).exists(),
1384 "marker deferred on unreadable DB"
1385 );
1386 }
1387
1388 fn seed_pre_v3_record(reg: &Registry, id: &str, session: Option<&str>) {
1391 reg.upsert(&NamedAgentRecord {
1392 schema_version: if session.is_some() { 2 } else { 1 },
1393 data: NamedAgentRecordV1 {
1394 instance_id: id.to_string(),
1395 instance_name: "demo".to_string(),
1396 definition_id: "claude-code".to_string(),
1397 identity_id: Some("ident-1".to_string()),
1398 memory_id: None,
1399 session_id: session.map(|s| s.to_string()),
1400 working_dir: "demo".to_string(),
1401 source_agents_base: None,
1402 created_at_ms: 10,
1403 last_launched_at_ms: 20,
1404 created_by_version: "0.43.0".to_string(),
1405 last_launched_by_version: "0.43.0".to_string(),
1406 },
1407 })
1408 .unwrap();
1409 }
1410
1411 #[test]
1412 fn backfill_sets_source_base_preserving_session_and_identity() {
1413 let (home, reg) = fresh_home();
1414 std::fs::write(reg.root().join(MARKER), b"x").unwrap();
1416 seed_pre_v3_record(®, "inst-1", Some("sess-keep"));
1417 let wd = channel_agents(home.path(), "stable").join("demo");
1419 std::fs::create_dir_all(&wd).unwrap();
1420 make_channel_db(
1421 home.path(),
1422 "stable",
1423 "0.44.2",
1424 &[("inst-1", "demo", 100, &wd.to_string_lossy())],
1425 );
1426
1427 let stats = backfill_source_bases_once(home.path(), ®).unwrap();
1428 assert_eq!(stats.records_updated, 1);
1429 assert!(stats.complete);
1430 let recs = reg.list_active().unwrap();
1431 assert_eq!(recs.len(), 1);
1432 let r = &recs[0].data;
1433 assert_eq!(
1435 r.source_agents_base.as_deref(),
1436 Some(channel_agents(home.path(), "stable").to_string_lossy().as_ref())
1437 );
1438 assert_eq!(r.session_id.as_deref(), Some("sess-keep"));
1440 assert_eq!(r.identity_id.as_deref(), Some("ident-1"));
1441 assert_eq!(recs[0].schema_version, 3);
1442 assert!(reg.root().join(SOURCE_BACKFILL_MARKER).exists());
1443 }
1444
1445 #[test]
1446 fn backfill_is_idempotent_via_marker() {
1447 let (home, reg) = fresh_home();
1448 let s1 = backfill_source_bases_once(home.path(), ®).unwrap();
1450 assert_eq!(s1.records_updated, 0);
1451 assert!(s1.complete);
1452 assert!(reg.root().join(SOURCE_BACKFILL_MARKER).exists());
1453 seed_pre_v3_record(®, "inst-late", None);
1455 let s2 = backfill_source_bases_once(home.path(), ®).unwrap();
1456 assert_eq!(s2.records_updated, 0, "marker short-circuits subsequent runs");
1457 }
1458
1459 #[test]
1460 fn backfill_anchors_on_latest_started_at_channel() {
1461 let (home, reg) = fresh_home();
1465 std::fs::write(reg.root().join(MARKER), b"x").unwrap();
1466 seed_pre_v3_record(®, "inst-dup", None);
1467 let wda = channel_agents(home.path(), "chan-a").join("demo");
1468 let wdb = channel_agents(home.path(), "chan-b").join("demo");
1469 std::fs::create_dir_all(&wda).unwrap();
1470 std::fs::create_dir_all(&wdb).unwrap();
1471 make_channel_db(
1473 home.path(),
1474 "chan-a",
1475 "0.1",
1476 &[("inst-dup", "demo", 100, &wda.to_string_lossy())],
1477 );
1478 make_channel_db(
1479 home.path(),
1480 "chan-b",
1481 "0.1",
1482 &[("inst-dup", "demo", 200, &wdb.to_string_lossy())],
1483 );
1484
1485 let stats = backfill_source_bases_once(home.path(), ®).unwrap();
1486 assert_eq!(stats.records_updated, 1);
1487 let recs = reg.list_active().unwrap();
1488 assert_eq!(
1489 recs[0].data.source_agents_base.as_deref(),
1490 Some(channel_agents(home.path(), "chan-b").to_string_lossy().as_ref()),
1491 "anchors on the latest-started_at channel (chan-b)"
1492 );
1493 }
1494
1495 #[test]
1496 fn backfill_counts_unresolved_when_source_db_gone() {
1497 let (home, reg) = fresh_home();
1498 std::fs::write(reg.root().join(MARKER), b"x").unwrap();
1499 seed_pre_v3_record(®, "inst-orphan", None);
1500 let stats = backfill_source_bases_once(home.path(), ®).unwrap();
1502 assert_eq!(stats.records_updated, 0);
1503 assert_eq!(stats.records_unresolved, 1);
1504 assert!(stats.complete, "no DB failure → marker written");
1505 let recs = reg.list_active().unwrap();
1507 assert!(recs[0].data.source_agents_base.is_none());
1508 }
1509}