1use std::collections::BTreeMap;
64use std::fs;
65use std::path::{Path, PathBuf};
66use std::time::Duration;
67
68use meerkat_core::storage_diagnostics::{DiagnoseScope, FindingSeverity, StorageFinding};
69pub use meerkat_store::migrate::{DivergenceStatus, MigrateMode, PruneAction, PruneArtifactKind};
70use meerkat_store::migrate::{
71 archive_path_read_only, backup_artifact_name, remove_maintenance_artifact,
72};
73use serde::{Deserialize, Serialize};
74use sha2::{Digest, Sha256};
75
76use crate::console_aggregator::SqliteConsoleLogStore;
77use crate::identity_first::LocalContinuityStore;
78use crate::memory::sqlite_store::SqliteAgentMemoryStore;
79use crate::runtime::SqliteMetadataStore;
80use crate::storage_doctor::{self, DATABASE_FAMILIES, MEMORY_LEDGER_DOMAIN, MEMORY_ROOT_SPELLINGS};
81use crate::storage_layout::{
82 DatabaseProvenance, DatabaseSlot, MobKitStorageLayout, RUNTIME_REGISTRY_FILE_NAME,
83 StorageLayoutError,
84};
85use crate::workgraph_admission::WORKGRAPH_ADMISSION_SIDECAR_FILE;
86
87const FENCE_DRAIN_DEADLINE: Duration = Duration::from_secs(10);
90
91pub const FINDING_DEAD_RUNTIME_REGISTRY_ENTRY: &str = "dead-runtime-registry-entry";
94
95pub fn is_registered_backup_artifact_name(name: &str) -> bool {
108 let Some(index) = name.rfind(".pre-") else {
109 return false;
110 };
111 if index == 0 {
112 return false; }
114 let rest = &name[index + ".pre-".len()..];
115 let Some((version, tail)) = rest.split_once('-') else {
116 return false;
117 };
118 let version_components: Vec<&str> = version.split('.').collect();
119 if version_components.len() < 2
120 || version_components
121 .iter()
122 .any(|part| part.is_empty() || !part.bytes().all(|byte| byte.is_ascii_digit()))
123 {
124 return false;
125 }
126 let (timestamp, purpose) = match tail.split_once('.') {
127 Some((timestamp, purpose)) => (timestamp, Some(purpose)),
128 None => (tail, None),
129 };
130 if timestamp.is_empty() || !timestamp.bytes().all(|byte| byte.is_ascii_digit()) {
131 return false;
132 }
133 purpose != Some("")
134}
135
136pub fn is_registered_quarantine_artifact_name(name: &str) -> bool {
141 let Some(index) = name.rfind(".corrupt-") else {
142 return false;
143 };
144 if index == 0 {
145 return false;
146 }
147 let digits = &name[index + ".corrupt-".len()..];
148 !digits.is_empty() && digits.bytes().all(|byte| byte.is_ascii_digit())
149}
150
151const LEFTOVER_FINDING_CODES: &[&str] = &[
154 storage_doctor::FINDING_LEGACY_FS_BLOBS,
155 storage_doctor::FINDING_BACKUP_ARTIFACT,
156 storage_doctor::FINDING_QUARANTINE_ARTIFACT,
157 storage_doctor::FINDING_WORKGRAPH_ADMISSION_SIDECAR,
158 storage_doctor::FINDING_MAINTENANCE_FENCE_LOCK,
159];
160
161pub fn enumerate_state_dir_databases(state_dir: &Path) -> Vec<PathBuf> {
175 let mut files: Vec<PathBuf> = Vec::new();
176 for slot in DatabaseSlot::ALL {
177 if slot == DatabaseSlot::AgentMemory {
178 continue; }
180 let mut names: Vec<&str> = vec![slot.canonical_name()];
181 names.extend(slot.legacy_names());
182 for name in names {
183 let path = state_dir.join(name);
184 if path.is_file() {
185 files.push(path);
186 }
187 }
188 }
189 for spelling in MEMORY_ROOT_SPELLINGS {
190 let root = state_dir.join(spelling);
191 let Ok(entries) = fs::read_dir(&root) else {
192 continue;
193 };
194 for entry in entries.filter_map(Result::ok) {
195 let path = entry.path();
196 if path.is_file() && path.extension().and_then(|ext| ext.to_str()) == Some("sqlite3") {
197 files.push(path);
198 }
199 }
200 }
201 let jobs = MobKitStorageLayout::with_injected_roots(state_dir.to_path_buf(), None).jobs_db();
202 if jobs.is_file() {
203 files.push(jobs);
204 }
205 files.sort();
206 files.dedup();
207 files
208}
209
210#[derive(Debug)]
220pub struct MobKitMaintenanceFence {
221 fences: Vec<meerkat_sqlite::ExclusiveFence>,
222 databases: Vec<PathBuf>,
223}
224
225impl MobKitMaintenanceFence {
226 pub fn acquire(
230 state_dir: &Path,
231 deadline: Duration,
232 ) -> Result<Self, meerkat_sqlite::SqliteStoreError> {
233 let started = std::time::Instant::now();
234 let databases = enumerate_state_dir_databases(state_dir);
235 let mut fences = Vec::with_capacity(databases.len());
236 for database in &databases {
237 let remaining = deadline.saturating_sub(started.elapsed());
238 let fence = meerkat_sqlite::ExclusiveFence::acquire(database, remaining)?;
240 fences.push(fence);
241 }
242 Ok(Self { fences, databases })
243 }
244
245 pub fn fenced_databases(&self) -> &[PathBuf] {
247 &self.databases
248 }
249
250 pub fn len(&self) -> usize {
252 self.fences.len()
253 }
254
255 pub fn is_empty(&self) -> bool {
257 self.fences.is_empty()
258 }
259
260 fn cover_renamed(&mut self, path: &Path) -> Result<(), meerkat_sqlite::SqliteStoreError> {
266 let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
271 if self
272 .databases
273 .iter()
274 .any(|held| std::fs::canonicalize(held).unwrap_or_else(|_| held.clone()) == canonical)
275 {
276 return Ok(());
277 }
278 match meerkat_sqlite::ExclusiveFence::try_acquire(path)? {
279 Some(fence) => {
280 self.fences.push(fence);
281 self.databases.push(path.to_path_buf());
282 Ok(())
283 }
284 None => Err(meerkat_sqlite::SqliteStoreError::MaintenanceFenceHeld {
285 path: path.to_path_buf(),
286 }),
287 }
288 }
289}
290
291#[non_exhaustive]
298#[derive(Debug, Clone, Default, Serialize, Deserialize)]
299pub struct MobKitMigrateReport {
300 #[serde(default)]
302 pub mode: MigrateMode,
303 #[serde(default)]
305 pub state_dir: PathBuf,
306 #[serde(default)]
309 pub fenced_databases: Vec<PathBuf>,
310 #[serde(default)]
312 pub ledger: Vec<LedgerBaselineEntry>,
313 #[serde(default)]
316 pub renames: Vec<FileRenameEntry>,
317 #[serde(default)]
319 pub twins: Vec<TwinReport>,
320 #[serde(default)]
323 pub findings: Vec<StorageFinding>,
324 #[serde(default)]
326 pub notes: Vec<String>,
327 #[serde(default)]
329 pub errors: Vec<String>,
330}
331
332impl MobKitMigrateReport {
333 pub fn new(mode: MigrateMode, state_dir: &Path) -> Self {
335 Self {
336 mode,
337 state_dir: state_dir.to_path_buf(),
338 ..Self::default()
339 }
340 }
341
342 pub fn has_errors(&self) -> bool {
345 !self.errors.is_empty()
346 }
347}
348
349#[non_exhaustive]
351#[derive(Debug, Clone, Serialize, Deserialize)]
352pub struct LedgerBaselineEntry {
353 pub database: PathBuf,
355 pub domain: String,
357 #[serde(default, skip_serializing_if = "Option::is_none")]
359 pub before: Option<i64>,
360 #[serde(default, skip_serializing_if = "Option::is_none")]
362 pub after: Option<i64>,
363 pub action: LedgerBaselineAction,
365}
366
367#[non_exhaustive]
369#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
370#[serde(rename_all = "kebab-case")]
371pub enum LedgerBaselineAction {
372 WouldStamp,
375 Recorded,
378 Stamped,
380 AlreadyCurrent,
382 ReportOnly,
386 Exempt,
389}
390
391#[non_exhaustive]
393#[derive(Debug, Clone, Serialize, Deserialize)]
394pub struct FileRenameEntry {
395 pub slot: String,
397 pub from: PathBuf,
399 pub to: PathBuf,
401 #[serde(default)]
403 pub siblings: Vec<SiblingRename>,
404 #[serde(default)]
406 pub wal_checkpointed: bool,
407 #[serde(default, skip_serializing_if = "Option::is_none")]
410 pub marker: Option<PathBuf>,
411 pub action: RenameAction,
413}
414
415#[non_exhaustive]
417#[derive(Debug, Clone, Serialize, Deserialize)]
418pub struct SiblingRename {
419 pub from: PathBuf,
420 pub to: PathBuf,
421}
422
423#[non_exhaustive]
425#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
426#[serde(rename_all = "kebab-case")]
427pub enum RenameAction {
428 WouldRename,
430 Renamed,
432 Refused,
435}
436
437#[non_exhaustive]
440#[derive(Debug, Clone, Serialize, Deserialize)]
441pub struct TwinReport {
442 pub slot: String,
444 pub paths: Vec<PathBuf>,
446 #[serde(default)]
449 pub byte_identical: bool,
450 #[serde(default)]
452 pub rows_equal: usize,
453 #[serde(default)]
457 pub rows: Vec<RowDivergenceEntry>,
458 pub resolution: TwinResolution,
460 #[serde(default)]
462 pub notes: Vec<String>,
463 #[serde(default)]
466 pub errors: Vec<String>,
467}
468
469#[non_exhaustive]
471#[derive(Debug, Clone, Serialize, Deserialize)]
472pub struct RowDivergenceEntry {
473 pub key: String,
475 pub status: DivergenceStatus,
477}
478
479#[non_exhaustive]
481#[derive(Debug, Clone, Serialize, Deserialize)]
482#[serde(rename_all = "snake_case", tag = "kind")]
483pub enum TwinResolution {
484 Refused {
486 reason: String,
488 },
489 Deduped {
493 kept: PathBuf,
495 archived: Vec<PathBuf>,
497 },
498 Adopted {
502 adopted: PathBuf,
504 archived: Vec<PathBuf>,
506 },
507}
508
509#[non_exhaustive]
514#[derive(Debug, Clone, Default, Serialize, Deserialize)]
515pub struct MobKitPruneReport {
516 #[serde(default)]
518 pub mode: MigrateMode,
519 #[serde(default)]
521 pub state_dir: PathBuf,
522 #[serde(default)]
525 pub older_than_days: u64,
526 #[serde(default)]
528 pub artifacts: Vec<MobKitPruneArtifact>,
529 #[serde(default)]
531 pub errors: Vec<String>,
532}
533
534impl MobKitPruneReport {
535 pub fn new(mode: MigrateMode, state_dir: &Path, older_than_days: u64) -> Self {
537 Self {
538 mode,
539 state_dir: state_dir.to_path_buf(),
540 older_than_days,
541 ..Self::default()
542 }
543 }
544
545 pub fn has_errors(&self) -> bool {
547 !self.errors.is_empty()
548 }
549}
550
551#[non_exhaustive]
553#[derive(Debug, Clone, Serialize, Deserialize)]
554pub struct MobKitPruneArtifact {
555 pub path: PathBuf,
557 pub kind: PruneArtifactKind,
559 #[serde(default)]
561 pub bytes: u64,
562 #[serde(default)]
564 pub age_days: u64,
565 pub action: PruneAction,
567}
568
569pub fn migrate_state_dir(
581 state_dir: &Path,
582 mode: MigrateMode,
583 adopt: Option<&Path>,
584) -> MobKitMigrateReport {
585 let mut report = MobKitMigrateReport::new(mode, state_dir);
586 if !state_dir.is_dir() {
587 report.errors.push(format!(
588 "state directory {} does not exist",
589 state_dir.display()
590 ));
591 return report;
592 }
593 let apply = mode == MigrateMode::Apply;
594 let layout = MobKitStorageLayout::with_injected_roots(state_dir.to_path_buf(), None);
595
596 let mut fence = if apply {
599 match MobKitMaintenanceFence::acquire(state_dir, FENCE_DRAIN_DEADLINE) {
600 Ok(fence) => {
601 report.fenced_databases = fence.fenced_databases().to_vec();
602 Some(fence)
603 }
604 Err(error) => {
605 report
606 .errors
607 .push(format!("maintenance fence not acquirable: {error}"));
608 return report;
609 }
610 }
611 } else {
612 report.fenced_databases = enumerate_state_dir_databases(state_dir);
613 None
614 };
615
616 let mut unfenced: Vec<PathBuf> = Vec::new();
620
621 let mut unresolved_twin = false;
623 for slot in DatabaseSlot::ALL {
624 let Err(StorageLayoutError::FileNameTwins { paths, .. }) = layout.resolve_database(slot)
625 else {
626 continue;
627 };
628 let twin = reconcile_twin(
629 slot,
630 &paths,
631 mode,
632 adopt,
633 &mut report.renames,
634 fence.as_mut(),
635 &mut report.errors,
636 &mut unfenced,
637 );
638 if let TwinResolution::Refused { reason } = &twin.resolution {
639 unresolved_twin = true;
640 report
641 .errors
642 .push(format!("file-name twins for the {slot} store: {reason}"));
643 }
644 report.twins.push(twin);
645 }
646 if unresolved_twin {
647 return report;
650 }
651
652 for slot in DatabaseSlot::ALL {
654 let Ok(resolved) = layout.resolve_database(slot) else {
655 continue; };
657 let DatabaseProvenance::LegacySpelling(_) = resolved.provenance else {
658 continue;
659 };
660 let to = state_dir.join(slot.canonical_name());
661 let entry = rename_to_canonical(slot, &resolved.path, &to, apply, &mut report.errors);
662 if entry.action == RenameAction::Renamed
663 && slot != DatabaseSlot::AgentMemory
664 && let Some(fence) = fence.as_mut()
665 && let Err(error) = fence.cover_renamed(&to)
666 {
667 report.errors.push(format!(
668 "renamed {slot} store is not fenced at its canonical path {}: {error}; \
669 skipping further mutation of this database",
670 to.display()
671 ));
672 unfenced.push(to.clone());
673 }
674 report.renames.push(entry);
675 }
676
677 let before = read_ledger_matrix(state_dir);
679 if apply {
680 open_stores_through_ledgered_constructors(&layout, &unfenced, &mut report.errors);
681 let after = read_ledger_matrix(state_dir);
682 let before_of = |database: &Path, domain: &str| -> Option<i64> {
683 before
684 .iter()
685 .find(|(db, dom, _)| db == database && dom == domain)
686 .and_then(|(_, _, version)| *version)
687 };
688 for (database, domain, after_version) in after {
689 let class = ledger_domain_class(&domain);
690 let before_version = before_of(&database, &domain);
691 let (action, after_field) = match class {
692 LedgerDomainClass::Stampable => {
693 if after_version == before_version {
694 (LedgerBaselineAction::AlreadyCurrent, after_version)
695 } else {
696 (LedgerBaselineAction::Stamped, after_version)
697 }
698 }
699 LedgerDomainClass::Exempt => (LedgerBaselineAction::Exempt, None),
700 LedgerDomainClass::ReportOnly => (LedgerBaselineAction::ReportOnly, None),
701 };
702 report.ledger.push(LedgerBaselineEntry {
703 database,
704 domain,
705 before: before_version,
706 after: after_field,
707 action,
708 });
709 }
710 } else {
711 for (database, domain, version) in before {
712 let action = match ledger_domain_class(&domain) {
713 LedgerDomainClass::Stampable => {
714 if version.is_some() {
715 LedgerBaselineAction::Recorded
716 } else {
717 LedgerBaselineAction::WouldStamp
718 }
719 }
720 LedgerDomainClass::Exempt => LedgerBaselineAction::Exempt,
721 LedgerDomainClass::ReportOnly => LedgerBaselineAction::ReportOnly,
722 };
723 report.ledger.push(LedgerBaselineEntry {
724 database,
725 domain,
726 before: version,
727 after: None,
728 action,
729 });
730 }
731 }
732 if report.ledger.iter().any(|entry| {
736 entry.domain == "mobkit-continuity"
737 && entry.before.is_some_and(|version| {
738 version < crate::identity_first::HEAD_CANONICAL_CONTINUITY_SCHEMA_VERSION
739 })
740 }) {
741 report.notes.push(format!(
742 "continuity database is below the head-canonical schema version \
743 (mobkit-continuity v{}); {} — this bump is ONE-WAY: binaries older than this \
744 release refuse a stamped file at open (SchemaFromTheFuture). Back up continuity.* \
745 before applying.",
746 crate::identity_first::HEAD_CANONICAL_CONTINUITY_SCHEMA_VERSION,
747 if apply {
748 "--apply committed it under the exclusive maintenance fence"
749 } else {
750 "--apply would commit it under the exclusive maintenance fence, and the first \
751 incremental session write would otherwise commit it lazily"
752 }
753 ));
754 }
755 if state_dir.join(WORKGRAPH_ADMISSION_SIDECAR_FILE).is_file() {
756 report.notes.push(
757 "workgraph admission sidecar is ledger-exempt by design (M3): the lock database \
758 deliberately carries no tables; stamping a ledger row on open would contend the \
759 cross-process admission lock"
760 .to_string(),
761 );
762 }
763 if report
764 .ledger
765 .iter()
766 .any(|entry| entry.action == LedgerBaselineAction::ReportOnly)
767 {
768 report.notes.push(
769 "sessions / runtime / schedule / workgraph are meerkat-owned stores: their ledgers \
770 are reported read-only here and converge through the owning meerkat store's next \
771 open (normal gateway boot)"
772 .to_string(),
773 );
774 }
775
776 let scope = DiagnoseScope::new(vec![state_dir.to_path_buf()]);
785 let diagnosis = storage_doctor::diagnose_state_dir_blocking(&scope, None);
786 report.findings.extend(
787 diagnosis
788 .findings
789 .into_iter()
790 .filter(|finding| LEFTOVER_FINDING_CODES.contains(&finding.code.as_str())),
791 );
792 sweep_runtime_registry(state_dir, &mut report.findings, &mut report.notes);
793
794 drop(fence);
795 report
796}
797
798enum LedgerDomainClass {
800 Stampable,
802 Exempt,
804 ReportOnly,
806}
807
808fn ledger_domain_class(domain: &str) -> LedgerDomainClass {
809 match domain {
810 "mobkit-continuity" | "mobkit-metadata" | "mobkit-console" | "jobs" => {
811 LedgerDomainClass::Stampable
812 }
813 "mobkit-workgraph-admission" => LedgerDomainClass::Exempt,
814 _ if domain == MEMORY_LEDGER_DOMAIN => LedgerDomainClass::Stampable,
815 _ => LedgerDomainClass::ReportOnly,
816 }
817}
818
819fn read_domain_versions(db_path: &Path) -> Result<Option<Vec<(String, i64)>>, String> {
822 let conn = meerkat_sqlite::open(db_path, meerkat_sqlite::ConnectionProfile::ReadOnly)
823 .map_err(|error| error.to_string())?;
824 if !table_exists(&conn, "meerkat_schema").map_err(|error| error.to_string())? {
825 return Ok(None);
826 }
827 let result = (|| -> Result<Vec<(String, i64)>, rusqlite::Error> {
828 let mut statement =
829 conn.prepare("SELECT domain, version FROM meerkat_schema ORDER BY domain")?;
830 let rows = statement
831 .query_map([], |row| {
832 Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
833 })?
834 .collect::<Result<Vec<_>, _>>()?;
835 Ok(rows)
836 })();
837 result.map(Some).map_err(|error| error.to_string())
838}
839
840fn table_exists(conn: &rusqlite::Connection, table: &str) -> Result<bool, rusqlite::Error> {
841 use rusqlite::OptionalExtension;
842 Ok(conn
843 .query_row(
844 "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1",
845 [table],
846 |_| Ok(()),
847 )
848 .optional()?
849 .is_some())
850}
851
852fn read_ledger_matrix(state_dir: &Path) -> Vec<(PathBuf, String, Option<i64>)> {
856 let mut matrix = Vec::new();
857 let mut push_db = |db_path: PathBuf, expected: &[&str]| {
858 let rows = read_domain_versions(&db_path)
859 .ok()
860 .flatten()
861 .unwrap_or_default();
862 for (domain, version) in &rows {
863 matrix.push((db_path.clone(), domain.clone(), Some(*version)));
864 }
865 for domain in expected {
866 if !rows.iter().any(|(name, _)| name == domain) {
867 matrix.push((db_path.clone(), (*domain).to_string(), None));
868 }
869 }
870 };
871 for family in DATABASE_FAMILIES {
872 for spelling in family.spellings {
873 let db_path = state_dir.join(spelling);
874 if db_path.is_file() {
875 push_db(db_path, family.ledger_domains);
876 }
877 }
878 }
879 for spelling in MEMORY_ROOT_SPELLINGS {
880 let root = state_dir.join(spelling);
881 let Ok(entries) = fs::read_dir(&root) else {
882 continue;
883 };
884 let mut realm_dbs: Vec<PathBuf> = entries
885 .filter_map(Result::ok)
886 .map(|entry| entry.path())
887 .filter(|path| {
888 path.is_file() && path.extension().and_then(|ext| ext.to_str()) == Some("sqlite3")
889 })
890 .collect();
891 realm_dbs.sort();
892 for db_path in realm_dbs {
893 push_db(db_path, &[MEMORY_LEDGER_DOMAIN]);
894 }
895 }
896 let jobs = MobKitStorageLayout::with_injected_roots(state_dir.to_path_buf(), None).jobs_db();
897 if jobs.is_file() {
898 push_db(jobs, &["jobs"]);
899 }
900 matrix
901}
902
903fn open_stores_through_ledgered_constructors(
911 layout: &MobKitStorageLayout,
912 unfenced: &[PathBuf],
913 errors: &mut Vec<String>,
914) {
915 match layout.continuity_db() {
916 Ok(resolved) if resolved.path.is_file() && !unfenced.contains(&resolved.path) => {
917 if let Err(error) = LocalContinuityStore::open(&resolved.path) {
918 errors.push(format!("continuity store open failed: {error}"));
919 } else if let Err(error) =
920 LocalContinuityStore::apply_head_canonical_schema_at(&resolved.path)
921 {
922 errors.push(format!(
930 "continuity head-canonical schema migration failed: {error}"
931 ));
932 }
933 }
934 Ok(_) => {}
935 Err(error) => errors.push(format!("continuity locator unresolved: {error}")),
936 }
937 match layout.metadata_db() {
938 Ok(resolved) if resolved.path.is_file() && !unfenced.contains(&resolved.path) => {
939 if let Err(error) = SqliteMetadataStore::open(&resolved.path) {
940 errors.push(format!("metadata store open failed: {error}"));
941 }
942 }
943 Ok(_) => {}
944 Err(error) => errors.push(format!("metadata locator unresolved: {error}")),
945 }
946 match layout.console_db() {
947 Ok(resolved) if resolved.path.is_file() && !unfenced.contains(&resolved.path) => {
948 if let Err(error) = SqliteConsoleLogStore::open(&resolved.path) {
949 errors.push(format!("console store open failed: {error}"));
950 }
951 }
952 Ok(_) => {}
953 Err(error) => errors.push(format!("console locator unresolved: {error}")),
954 }
955 match layout.agent_memory_root() {
956 Ok(resolved) if resolved.path.is_dir() => {
957 match SqliteAgentMemoryStore::open(&resolved.path) {
958 Ok(store) => match store.known_realms() {
959 Ok(realms) => {
960 for realm in realms {
961 if let Err(error) = store.open_realm_ledgered(&realm) {
962 errors.push(format!(
963 "agent-memory realm '{realm}' open failed: {error}"
964 ));
965 }
966 }
967 }
968 Err(error) => {
969 errors.push(format!("agent-memory realm listing failed: {error}"));
970 }
971 },
972 Err(error) => errors.push(format!("agent-memory store open failed: {error}")),
973 }
974 }
975 Ok(_) => {}
976 Err(error) => errors.push(format!("agent-memory locator unresolved: {error}")),
977 }
978 let jobs = layout.jobs_db();
979 if jobs.is_file()
980 && !unfenced.contains(&jobs)
981 && let Err(error) = meerkat::SqliteDetachedJobStore::open(&jobs)
982 {
983 errors.push(format!("detached-job store open failed: {error}"));
984 }
985}
986
987fn path_with_suffix(path: &Path, suffix: &str) -> PathBuf {
992 let mut os = path.as_os_str().to_os_string();
993 os.push(suffix);
994 PathBuf::from(os)
995}
996
997fn checkpoint_wal_if_nonempty(db: &Path) -> Result<bool, String> {
1003 let wal = path_with_suffix(db, "-wal");
1004 let wal_len = fs::metadata(&wal).map(|meta| meta.len()).unwrap_or(0);
1005 if wal_len == 0 {
1006 return Ok(false);
1007 }
1008 let conn = meerkat_sqlite::open(
1009 db,
1010 meerkat_sqlite::ConnectionProfile::Maintenance { write: true },
1011 )
1012 .map_err(|error| format!("open for WAL checkpoint failed: {error}"))?;
1013 let busy: i64 = conn
1014 .query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |row| row.get(0))
1015 .map_err(|error| format!("wal_checkpoint failed: {error}"))?;
1016 if busy != 0 {
1017 return Err(format!(
1018 "WAL for {} cannot be checkpointed (readers active); refusing to move it",
1019 db.display()
1020 ));
1021 }
1022 Ok(true)
1023}
1024
1025fn move_database_file(from: &Path, to: &Path) -> Result<(Vec<SiblingRename>, bool), String> {
1030 if to.exists() {
1031 return Err(format!(
1032 "cannot move {} to {}: target already exists",
1033 from.display(),
1034 to.display()
1035 ));
1036 }
1037 let checkpointed = checkpoint_wal_if_nonempty(from)?;
1038 fs::rename(from, to)
1039 .map_err(|error| format!("rename {} -> {}: {error}", from.display(), to.display()))?;
1040 let mut siblings = Vec::new();
1041 for suffix in ["-wal", "-shm"] {
1042 let src = path_with_suffix(from, suffix);
1043 if !src.exists() {
1044 continue;
1045 }
1046 let dst = path_with_suffix(to, suffix);
1047 fs::rename(&src, &dst)
1048 .map_err(|error| format!("rename {} -> {}: {error}", src.display(), dst.display()))?;
1049 siblings.push(SiblingRename { from: src, to: dst });
1050 }
1051 Ok((siblings, checkpointed))
1052}
1053
1054fn write_rename_marker(from: &Path, to: &Path) -> Result<PathBuf, String> {
1059 let from_name = from
1060 .file_name()
1061 .and_then(|name| name.to_str())
1062 .ok_or_else(|| format!("{} has no UTF-8 file name", from.display()))?;
1063 let to_name = to
1064 .file_name()
1065 .and_then(|name| name.to_str())
1066 .unwrap_or_default();
1067 let marker = from.with_file_name(backup_artifact_name(from_name, "renamed"));
1068 let body = serde_json::json!({
1069 "renamed_from": from_name,
1070 "renamed_to": to_name,
1071 });
1072 let bytes = serde_json::to_vec_pretty(&body).map_err(|error| error.to_string())?;
1073 fs::write(&marker, bytes)
1074 .map_err(|error| format!("write rename marker {}: {error}", marker.display()))?;
1075 Ok(marker)
1076}
1077
1078fn rename_to_canonical(
1081 slot: DatabaseSlot,
1082 from: &Path,
1083 to: &Path,
1084 apply: bool,
1085 errors: &mut Vec<String>,
1086) -> FileRenameEntry {
1087 let mut entry = FileRenameEntry {
1088 slot: slot.to_string(),
1089 from: from.to_path_buf(),
1090 to: to.to_path_buf(),
1091 siblings: Vec::new(),
1092 wal_checkpointed: false,
1093 marker: None,
1094 action: RenameAction::WouldRename,
1095 };
1096 if !apply {
1097 return entry;
1098 }
1099 let moved = if slot == DatabaseSlot::AgentMemory {
1100 if to.exists() {
1103 Err(format!(
1104 "cannot move {} to {}: target already exists",
1105 from.display(),
1106 to.display()
1107 ))
1108 } else {
1109 fs::rename(from, to)
1110 .map_err(|error| format!("rename {} -> {}: {error}", from.display(), to.display()))
1111 .map(|()| (Vec::new(), false))
1112 }
1113 } else {
1114 move_database_file(from, to)
1115 };
1116 match moved {
1117 Ok((siblings, checkpointed)) => {
1118 entry.siblings = siblings;
1119 entry.wal_checkpointed = checkpointed;
1120 entry.action = RenameAction::Renamed;
1121 match write_rename_marker(from, to) {
1122 Ok(marker) => entry.marker = Some(marker),
1123 Err(error) => errors.push(error),
1124 }
1125 }
1126 Err(error) => {
1127 entry.action = RenameAction::Refused;
1128 errors.push(format!("rename of the {slot} store refused: {error}"));
1129 }
1130 }
1131 entry
1132}
1133
1134fn file_digest(path: &Path) -> Result<[u8; 32], String> {
1142 let mut hasher = Sha256::new();
1143 hasher.update(fs::read(path).map_err(|error| format!("{}: {error}", path.display()))?);
1144 let wal = path_with_suffix(path, "-wal");
1145 if wal.is_file() {
1146 hasher.update(fs::read(&wal).map_err(|error| format!("{}: {error}", wal.display()))?);
1147 }
1148 Ok(hasher.finalize().into())
1149}
1150
1151fn quote_ident(name: &str) -> String {
1152 format!("\"{}\"", name.replace('"', "\"\""))
1153}
1154
1155type RowDigests = BTreeMap<String, [u8; 32]>;
1156
1157fn row_digests(db_path: &Path) -> Result<RowDigests, String> {
1164 let conn = meerkat_sqlite::open(db_path, meerkat_sqlite::ConnectionProfile::ReadOnly)
1165 .map_err(|error| error.to_string())?;
1166 let inner = || -> Result<RowDigests, rusqlite::Error> {
1167 let tables: Vec<String> = {
1168 let mut statement = conn.prepare(
1169 "SELECT name FROM sqlite_master WHERE type = 'table' \
1170 AND name NOT LIKE 'sqlite_%' AND name <> 'meerkat_schema' ORDER BY name",
1171 )?;
1172 statement
1173 .query_map([], |row| row.get::<_, String>(0))?
1174 .collect::<Result<Vec<_>, _>>()?
1175 };
1176 let mut digests = RowDigests::new();
1177 for table in tables {
1178 let mut pk_columns: Vec<(i64, usize)> = Vec::new();
1180 conn.pragma(None, "table_info", &table, |row| {
1181 let cid: i64 = row.get(0)?;
1182 let pk: i64 = row.get(5)?;
1183 if pk > 0 {
1184 pk_columns.push((pk, cid as usize));
1185 }
1186 Ok(())
1187 })?;
1188 pk_columns.sort_unstable();
1189
1190 let (sql, key_indexes, data_start): (String, Vec<usize>, usize) =
1191 if pk_columns.is_empty() {
1192 (
1193 format!(
1194 "SELECT rowid, * FROM {} ORDER BY rowid",
1195 quote_ident(&table)
1196 ),
1197 vec![0],
1198 1,
1199 )
1200 } else {
1201 let order: Vec<String> = pk_columns
1202 .iter()
1203 .map(|(_, cid)| format!("{}", cid + 1))
1204 .collect();
1205 (
1206 format!(
1207 "SELECT * FROM {} ORDER BY {}",
1208 quote_ident(&table),
1209 order.join(", ")
1210 ),
1211 pk_columns.iter().map(|(_, cid)| *cid).collect(),
1212 0,
1213 )
1214 };
1215 let mut statement = conn.prepare(&sql)?;
1216 let column_count = statement.column_count();
1217 let mut rows = statement.query([])?;
1218 while let Some(row) = rows.next()? {
1219 let mut key = format!("{table}/");
1220 for (position, index) in key_indexes.iter().enumerate() {
1221 if position > 0 {
1222 key.push('|');
1223 }
1224 key.push_str(&value_key(row.get_ref(*index)?));
1225 }
1226 let mut hasher = Sha256::new();
1227 for index in data_start..column_count {
1228 feed_value(&mut hasher, row.get_ref(index)?);
1229 }
1230 digests.insert(key, hasher.finalize().into());
1231 }
1232 }
1233 Ok(digests)
1234 };
1235 inner().map_err(|error| format!("{}: {error}", db_path.display()))
1236}
1237
1238fn value_key(value: rusqlite::types::ValueRef<'_>) -> String {
1239 use rusqlite::types::ValueRef;
1240 match value {
1241 ValueRef::Null => "null".to_string(),
1242 ValueRef::Integer(int) => int.to_string(),
1243 ValueRef::Real(real) => real.to_string(),
1244 ValueRef::Text(text) => String::from_utf8_lossy(text).into_owned(),
1245 ValueRef::Blob(blob) => {
1246 use std::fmt::Write as _;
1247 blob.iter().fold(String::new(), |mut hex, byte| {
1248 let _ = write!(hex, "{byte:02x}");
1249 hex
1250 })
1251 }
1252 }
1253}
1254
1255fn feed_value(hasher: &mut Sha256, value: rusqlite::types::ValueRef<'_>) {
1256 use rusqlite::types::ValueRef;
1257 let (tag, bytes): (u8, Vec<u8>) = match value {
1258 ValueRef::Null => (0, Vec::new()),
1259 ValueRef::Integer(int) => (1, int.to_be_bytes().to_vec()),
1260 ValueRef::Real(real) => (2, real.to_be_bytes().to_vec()),
1261 ValueRef::Text(text) => (3, text.to_vec()),
1262 ValueRef::Blob(blob) => (4, blob.to_vec()),
1263 };
1264 hasher.update([tag]);
1265 hasher.update((bytes.len() as u64).to_be_bytes());
1266 hasher.update(&bytes);
1267}
1268
1269fn dir_file_digests(dir: &Path) -> Result<RowDigests, String> {
1272 let mut digests = RowDigests::new();
1273 let entries = fs::read_dir(dir).map_err(|error| format!("{}: {error}", dir.display()))?;
1274 let mut files: Vec<PathBuf> = entries
1275 .filter_map(Result::ok)
1276 .map(|entry| entry.path())
1277 .filter(|path| path.is_file())
1278 .collect();
1279 files.sort();
1280 for file in files {
1281 let Some(name) = file.file_name().and_then(|name| name.to_str()) else {
1282 continue;
1283 };
1284 if name.ends_with("-wal") || name.ends_with("-shm") {
1286 continue;
1287 }
1288 digests.insert(name.to_string(), file_digest(&file)?);
1289 }
1290 Ok(digests)
1291}
1292
1293fn classify_divergence(
1296 per_copy: &[(PathBuf, RowDigests)],
1297 rows_equal: &mut usize,
1298 rows: &mut Vec<RowDivergenceEntry>,
1299) {
1300 let mut all_keys: Vec<String> = per_copy
1301 .iter()
1302 .flat_map(|(_, digests)| digests.keys().cloned())
1303 .collect();
1304 all_keys.sort();
1305 all_keys.dedup();
1306 for key in all_keys {
1307 let holders: Vec<(&PathBuf, &[u8; 32])> = per_copy
1308 .iter()
1309 .filter_map(|(path, digests)| digests.get(&key).map(|digest| (path, digest)))
1310 .collect();
1311 if holders.len() == 1 {
1312 rows.push(RowDivergenceEntry {
1313 key,
1314 status: DivergenceStatus::OnlyIn {
1315 location: holders[0].0.clone(),
1316 },
1317 });
1318 } else if holders.len() == per_copy.len()
1319 && holders.iter().all(|(_, digest)| *digest == holders[0].1)
1320 {
1321 *rows_equal += 1;
1322 } else {
1323 rows.push(RowDivergenceEntry {
1324 key,
1325 status: DivergenceStatus::Divergent,
1326 });
1327 }
1328 }
1329}
1330
1331fn archive_twin_copy(
1336 path: &Path,
1337 purpose: &str,
1338 apply_checkpoint: bool,
1339) -> Result<Vec<PathBuf>, String> {
1340 if path.is_file() && apply_checkpoint {
1341 checkpoint_wal_if_nonempty(path)?;
1342 }
1343 let siblings: Vec<PathBuf> = ["-wal", "-shm"]
1344 .iter()
1345 .map(|suffix| path_with_suffix(path, suffix))
1346 .filter(|sibling| sibling.exists())
1347 .collect();
1348 let archive = archive_path_read_only(path, purpose).map_err(|error| error.to_string())?;
1349 let mut archived = vec![archive.clone()];
1350 for sibling in siblings {
1351 let Some(suffix) = sibling
1352 .file_name()
1353 .and_then(|name| name.to_str())
1354 .and_then(|name| name.rsplit_once('-').map(|(_, tail)| format!("-{tail}")))
1355 else {
1356 continue;
1357 };
1358 let dst = path_with_suffix(&archive, &suffix);
1359 fs::rename(&sibling, &dst)
1360 .map_err(|error| format!("archive sibling {}: {error}", sibling.display()))?;
1361 if let Ok(metadata) = fs::symlink_metadata(&dst) {
1362 let mut permissions = metadata.permissions();
1363 permissions.set_readonly(true);
1364 let _ = fs::set_permissions(&dst, permissions);
1365 }
1366 archived.push(dst);
1367 }
1368 Ok(archived)
1369}
1370
1371#[allow(clippy::too_many_arguments)]
1376fn reconcile_twin(
1377 slot: DatabaseSlot,
1378 paths: &[PathBuf],
1379 mode: MigrateMode,
1380 adopt: Option<&Path>,
1381 renames: &mut Vec<FileRenameEntry>,
1382 fence: Option<&mut MobKitMaintenanceFence>,
1383 errors: &mut Vec<String>,
1384 unfenced: &mut Vec<PathBuf>,
1385) -> TwinReport {
1386 let mut twin = TwinReport {
1387 slot: slot.to_string(),
1388 paths: paths.to_vec(),
1389 byte_identical: false,
1390 rows_equal: 0,
1391 rows: Vec::new(),
1392 resolution: TwinResolution::Refused {
1393 reason: "unresolved".to_string(),
1394 },
1395 notes: Vec::new(),
1396 errors: Vec::new(),
1397 };
1398 match slot {
1399 DatabaseSlot::Continuity => twin.notes.push(
1400 "no synthesis: continuity fencing tokens are per-database sequences; merging twin \
1401 histories would corrupt CAS — adopt one copy, archive the rest"
1402 .to_string(),
1403 ),
1404 DatabaseSlot::Console => twin.notes.push(
1405 "no synthesis: console cursor_seq is a per-database AUTOINCREMENT sequence; merging \
1406 twin timelines would corrupt cursor replay — adopt one copy, archive the rest"
1407 .to_string(),
1408 ),
1409 _ => {}
1410 }
1411
1412 let row_level = matches!(
1415 slot,
1416 DatabaseSlot::Continuity | DatabaseSlot::Metadata | DatabaseSlot::Console
1417 );
1418 if row_level || slot == DatabaseSlot::AgentMemory {
1419 let mut per_copy: Vec<(PathBuf, RowDigests)> = Vec::new();
1420 for path in paths {
1421 let digests = if row_level {
1422 row_digests(path)
1423 } else {
1424 dir_file_digests(path)
1425 };
1426 match digests {
1427 Ok(digests) => per_copy.push((path.clone(), digests)),
1428 Err(error) => {
1429 twin.errors.push(format!(
1430 "divergence unavailable for {}: {error}",
1431 path.display()
1432 ));
1433 per_copy.push((path.clone(), RowDigests::new()));
1434 }
1435 }
1436 }
1437 classify_divergence(&per_copy, &mut twin.rows_equal, &mut twin.rows);
1438 } else {
1439 twin.notes
1440 .push("divergence computed at file-digest level for this slot".to_string());
1441 }
1442
1443 let byte_identical = if slot == DatabaseSlot::AgentMemory {
1446 let mut maps = Vec::new();
1447 let mut readable = true;
1448 for path in paths {
1449 match dir_file_digests(path) {
1450 Ok(map) => maps.push(map),
1451 Err(_) => {
1452 readable = false;
1453 break;
1454 }
1455 }
1456 }
1457 readable && maps.windows(2).all(|pair| pair[0] == pair[1])
1458 } else {
1459 let mut digests = Vec::new();
1460 let mut readable = true;
1461 for path in paths {
1462 match file_digest(path) {
1463 Ok(digest) => digests.push(digest),
1464 Err(error) => {
1465 twin.errors.push(format!(
1466 "file digest unavailable for {}: {error}",
1467 path.display()
1468 ));
1469 readable = false;
1470 break;
1471 }
1472 }
1473 }
1474 readable && digests.windows(2).all(|pair| pair[0] == pair[1])
1475 };
1476 twin.byte_identical = byte_identical;
1477
1478 if mode != MigrateMode::Apply {
1479 twin.resolution = TwinResolution::Refused {
1480 reason: if byte_identical {
1481 "twin copies are byte-identical; rerun with `--apply` to dedup (keep the \
1482 canonical spelling, archive the redundant copy read-only)"
1483 .to_string()
1484 } else {
1485 "divergent twin copies; rerun with `--apply --adopt <path>` to adopt one copy \
1486 and archive the rest read-only (no synthesis)"
1487 .to_string()
1488 },
1489 };
1490 return twin;
1491 }
1492
1493 let canonical_of = |path: &Path| fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
1495 let adopted_member = adopt.and_then(|adopt_path| {
1496 let adopt_canonical = canonical_of(adopt_path);
1497 paths
1498 .iter()
1499 .find(|path| canonical_of(path) == adopt_canonical)
1500 .cloned()
1501 });
1502 let (kept, purpose, dedup) = if let Some(adopted) = adopted_member {
1503 (adopted, "twin", false)
1504 } else if byte_identical {
1505 let canonical_name = slot.canonical_name();
1506 let kept = paths
1507 .iter()
1508 .find(|path| path.file_name().and_then(|name| name.to_str()) == Some(canonical_name))
1509 .unwrap_or(&paths[0])
1510 .clone();
1511 (kept, "twin-dedup", true)
1512 } else {
1513 twin.resolution = TwinResolution::Refused {
1514 reason: if adopt.is_some() {
1515 format!(
1516 "--adopt does not name one of this slot's twin copies (candidates: {})",
1517 paths
1518 .iter()
1519 .map(|path| path.display().to_string())
1520 .collect::<Vec<_>>()
1521 .join(", ")
1522 )
1523 } else {
1524 "divergent twin copies; rerun with `--apply --adopt <path>` to adopt one copy \
1525 and archive the rest read-only (no synthesis)"
1526 .to_string()
1527 },
1528 };
1529 return twin;
1530 };
1531
1532 let mut archived = Vec::new();
1535 for path in paths {
1536 if *path == kept {
1537 continue;
1538 }
1539 match archive_twin_copy(path, purpose, true) {
1540 Ok(mut paths) => archived.append(&mut paths),
1541 Err(error) => {
1542 twin.resolution = TwinResolution::Refused {
1543 reason: format!("archive of {} failed: {error}", path.display()),
1544 };
1545 return twin;
1546 }
1547 }
1548 }
1549 let canonical_path = kept.parent().map_or_else(
1550 || PathBuf::from(slot.canonical_name()),
1551 |parent| parent.join(slot.canonical_name()),
1552 );
1553 let final_path = if kept == canonical_path {
1554 kept
1555 } else {
1556 let mut rename_errors = Vec::new();
1557 let entry = rename_to_canonical(slot, &kept, &canonical_path, true, &mut rename_errors);
1558 let renamed = entry.action == RenameAction::Renamed;
1559 renames.push(entry);
1560 if renamed {
1561 if slot != DatabaseSlot::AgentMemory
1562 && let Some(fence) = fence
1563 && let Err(error) = fence.cover_renamed(&canonical_path)
1564 {
1565 errors.push(format!(
1566 "adopted {slot} store is not fenced at its canonical path {}: {error}; \
1567 skipping further mutation of this database",
1568 canonical_path.display()
1569 ));
1570 unfenced.push(canonical_path.clone());
1571 }
1572 canonical_path
1573 } else {
1574 twin.resolution = TwinResolution::Refused {
1575 reason: format!(
1576 "adopted copy could not be renamed to its canonical name: {}",
1577 rename_errors.join("; ")
1578 ),
1579 };
1580 return twin;
1581 }
1582 };
1583 twin.resolution = if dedup {
1584 TwinResolution::Deduped {
1585 kept: final_path,
1586 archived,
1587 }
1588 } else {
1589 TwinResolution::Adopted {
1590 adopted: final_path,
1591 archived,
1592 }
1593 };
1594 twin
1595}
1596
1597fn pid_alive(pid: u64) -> Option<bool> {
1605 #[cfg(unix)]
1606 {
1607 std::process::Command::new("ps")
1608 .args(["-p", &pid.to_string()])
1609 .stdout(std::process::Stdio::null())
1610 .stderr(std::process::Stdio::null())
1611 .status()
1612 .ok()
1613 .map(|status| status.success())
1614 }
1615 #[cfg(not(unix))]
1616 {
1617 let _ = pid;
1618 None
1619 }
1620}
1621
1622fn sweep_runtime_registry(
1626 state_dir: &Path,
1627 findings: &mut Vec<StorageFinding>,
1628 notes: &mut Vec<String>,
1629) {
1630 let registry_path = state_dir.join(RUNTIME_REGISTRY_FILE_NAME);
1631 if !registry_path.is_file() {
1632 return;
1633 }
1634 let Ok(bytes) = fs::read(®istry_path) else {
1635 return;
1636 };
1637 let Ok(value) = serde_json::from_slice::<serde_json::Value>(&bytes) else {
1638 notes.push(format!(
1639 "runtime registry {} is unparseable; dead-entry census skipped",
1640 registry_path.display()
1641 ));
1642 return;
1643 };
1644 let Some(entries) = value.get("entries").and_then(serde_json::Value::as_array) else {
1645 return;
1646 };
1647 for entry in entries {
1648 let runtime_id = entry
1649 .get("runtime_id")
1650 .and_then(serde_json::Value::as_str)
1651 .unwrap_or("<unknown>");
1652 let Some(pid) = entry.get("pid").and_then(serde_json::Value::as_u64) else {
1653 continue;
1654 };
1655 match pid_alive(pid) {
1656 Some(false) => findings.push(
1657 StorageFinding::new(
1658 FindingSeverity::Info,
1659 FINDING_DEAD_RUNTIME_REGISTRY_ENTRY,
1660 format!(
1661 "runtime registry entry '{runtime_id}' records pid {pid}, which is no \
1662 longer alive (report-only; the gateway prunes dead entries on next boot)"
1663 ),
1664 )
1665 .with_path(registry_path.clone()),
1666 ),
1667 Some(true) => {}
1668 None => notes.push(format!(
1669 "runtime registry entry '{runtime_id}': pid liveness not determinable on this \
1670 platform"
1671 )),
1672 }
1673 }
1674}
1675
1676fn recursive_size(path: &Path) -> u64 {
1681 let Ok(metadata) = fs::symlink_metadata(path) else {
1682 return 0;
1683 };
1684 if metadata.is_dir() {
1685 let Ok(entries) = fs::read_dir(path) else {
1686 return 0;
1687 };
1688 entries
1689 .filter_map(Result::ok)
1690 .map(|entry| recursive_size(&entry.path()))
1691 .sum()
1692 } else {
1693 metadata.len()
1694 }
1695}
1696
1697fn age_days(path: &Path) -> u64 {
1698 fs::symlink_metadata(path)
1699 .and_then(|metadata| metadata.modified())
1700 .ok()
1701 .and_then(|modified| std::time::SystemTime::now().duration_since(modified).ok())
1702 .map(|age| age.as_secs() / 86_400)
1703 .unwrap_or(0)
1704}
1705
1706fn artifact_kind(name: &str) -> Option<PruneArtifactKind> {
1707 if is_registered_backup_artifact_name(name) {
1708 Some(PruneArtifactKind::BackupArtifact)
1709 } else if is_registered_quarantine_artifact_name(name) {
1710 Some(PruneArtifactKind::QuarantinedIndex)
1711 } else {
1712 None
1713 }
1714}
1715
1716fn push_artifacts_in(dir: &Path, dirs_too: bool, artifacts: &mut Vec<MobKitPruneArtifact>) {
1717 let Ok(entries) = fs::read_dir(dir) else {
1718 return;
1719 };
1720 let mut paths: Vec<PathBuf> = entries
1721 .filter_map(Result::ok)
1722 .map(|entry| entry.path())
1723 .collect();
1724 paths.sort();
1725 for path in paths {
1726 if path.is_dir() && !dirs_too {
1727 continue;
1728 }
1729 let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
1730 continue;
1731 };
1732 let Some(kind) = artifact_kind(name) else {
1733 continue;
1734 };
1735 artifacts.push(MobKitPruneArtifact {
1736 bytes: recursive_size(&path),
1737 age_days: age_days(&path),
1738 path,
1739 kind,
1740 action: PruneAction::Kept,
1741 });
1742 }
1743}
1744
1745pub fn enumerate_state_dir_artifacts(state_dir: &Path) -> Vec<MobKitPruneArtifact> {
1752 let mut artifacts = Vec::new();
1753 push_artifacts_in(state_dir, true, &mut artifacts);
1754 for spelling in MEMORY_ROOT_SPELLINGS {
1755 let root = state_dir.join(spelling);
1756 if root.is_dir() {
1757 push_artifacts_in(&root, false, &mut artifacts);
1758 }
1759 }
1760 artifacts
1761}
1762
1763pub fn prune_state_dir(
1770 state_dir: &Path,
1771 older_than_days: u64,
1772 mode: MigrateMode,
1773) -> MobKitPruneReport {
1774 let mut report = MobKitPruneReport::new(mode, state_dir, older_than_days);
1775 let mut artifacts = enumerate_state_dir_artifacts(state_dir);
1776 for artifact in &mut artifacts {
1777 if artifact.age_days < older_than_days {
1778 artifact.action = PruneAction::Kept;
1779 continue;
1780 }
1781 if mode != MigrateMode::Apply {
1782 artifact.action = PruneAction::WouldDelete;
1783 continue;
1784 }
1785 match remove_maintenance_artifact(&artifact.path) {
1786 Ok(()) => artifact.action = PruneAction::Deleted,
1787 Err(error) => {
1788 artifact.action = PruneAction::DeleteFailed;
1789 report.errors.push(format!(
1790 "failed to delete {}: {error}",
1791 artifact.path.display()
1792 ));
1793 }
1794 }
1795 }
1796 report.artifacts = artifacts;
1797 report
1798}
1799
1800#[cfg(test)]
1801#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
1802mod tests {
1803 use super::*;
1804 use rusqlite::Connection;
1805
1806 const LEGACY_CONTINUITY_DDL: &str = "CREATE TABLE continuity_records (
1808 identity TEXT PRIMARY KEY,
1809 agent_runtime_id TEXT NOT NULL,
1810 session_id TEXT NOT NULL,
1811 generation INTEGER NOT NULL,
1812 checkpoint_version INTEGER NOT NULL,
1813 fencing_token INTEGER NOT NULL
1814 );
1815 CREATE TABLE session_snapshots (
1816 session_id TEXT PRIMARY KEY,
1817 identity TEXT NOT NULL,
1818 generation INTEGER NOT NULL,
1819 checkpoint_version INTEGER NOT NULL,
1820 fencing_token INTEGER NOT NULL,
1821 data BLOB NOT NULL
1822 );";
1823
1824 fn create_legacy_continuity(path: &Path) -> Connection {
1828 let conn = Connection::open(path).expect("create fixture db");
1829 conn.execute_batch(LEGACY_CONTINUITY_DDL)
1830 .expect("apply legacy ddl");
1831 conn.execute_batch(
1832 "CREATE TABLE IF NOT EXISTS meerkat_schema (
1833 domain TEXT PRIMARY KEY,
1834 version INTEGER NOT NULL
1835 );
1836 INSERT INTO meerkat_schema (domain, version) VALUES ('mobkit-continuity', 1);",
1837 )
1838 .expect("stamp fixture ledger at the v1 floor");
1839 conn
1840 }
1841
1842 fn insert_snapshot(conn: &Connection, session_id: &str, identity: &str, data: &[u8]) {
1843 conn.execute(
1844 "INSERT INTO session_snapshots \
1845 (session_id, identity, generation, checkpoint_version, fencing_token, data) \
1846 VALUES (?1, ?2, 3, 4, 7, ?3)",
1847 rusqlite::params![session_id, identity, data],
1848 )
1849 .expect("insert snapshot");
1850 }
1851
1852 fn insert_record(conn: &Connection, identity: &str, session_id: &str) {
1853 conn.execute(
1854 "INSERT INTO continuity_records \
1855 (identity, agent_runtime_id, session_id, generation, checkpoint_version, \
1856 fencing_token) VALUES (?1, 'rt-1', ?2, 3, 4, 7)",
1857 rusqlite::params![identity, session_id],
1858 )
1859 .expect("insert record");
1860 }
1861
1862 fn legacy_session_bytes() -> (String, Vec<u8>) {
1863 let session = meerkat_core::Session::new();
1864 let id = session.id().to_string();
1865 let bytes = serde_json::to_vec(&session).expect("serialize legacy session");
1866 (id, bytes)
1867 }
1868
1869 fn file_digest_hex(path: &Path) -> String {
1870 format!("{:x}", Sha256::digest(fs::read(path).expect("read db")))
1871 }
1872
1873 fn ledger_entry<'a>(
1874 report: &'a MobKitMigrateReport,
1875 file_name: &str,
1876 domain: &str,
1877 ) -> &'a LedgerBaselineEntry {
1878 report
1879 .ledger
1880 .iter()
1881 .find(|entry| entry.database.ends_with(file_name) && entry.domain == domain)
1882 .unwrap_or_else(|| panic!("no ledger entry for {file_name} [{domain}]"))
1883 }
1884
1885 #[test]
1886 fn fence_enumerates_layout_slots_and_memory_realms_sorted() {
1887 let temp = tempfile::tempdir().expect("tempdir");
1888 let state = temp.path();
1889 drop(create_legacy_continuity(&state.join("continuity.db")));
1890 drop(Connection::open(state.join("sessions.sqlite3")).expect("sessions"));
1891 drop(Connection::open(state.join("runtime.sqlite")).expect("runtime"));
1892 fs::create_dir_all(state.join("agent-memory")).expect("memory root");
1893 drop(Connection::open(state.join("agent-memory/alpha.sqlite3")).expect("realm"));
1894 let jobs = MobKitStorageLayout::with_injected_roots(state.to_path_buf(), None).jobs_db();
1895 drop(meerkat::SqliteDetachedJobStore::open(&jobs).expect("jobs"));
1896 drop(Connection::open(state.join(WORKGRAPH_ADMISSION_SIDECAR_FILE)).expect("sidecar"));
1899 fs::write(state.join("notes.txt"), b"x").expect("notes");
1900
1901 let fence = MobKitMaintenanceFence::acquire(state, Duration::from_secs(1)).expect("fence");
1902 let mut expected = vec![
1903 state.join("agent-memory/alpha.sqlite3"),
1904 state.join("continuity.db"),
1905 jobs,
1906 state.join("runtime.sqlite"),
1907 state.join("sessions.sqlite3"),
1908 ];
1909 expected.sort();
1910 assert_eq!(fence.fenced_databases(), expected.as_slice());
1911 assert_eq!(fence.len(), 5);
1912 assert!(!fence.is_empty());
1913 for database in fence.fenced_databases() {
1914 assert!(meerkat_sqlite::fence_lock_path(database).is_file());
1915 }
1916 }
1917
1918 #[test]
1919 fn fence_foreign_holder_fails_typed_and_releases_partial_acquisition() {
1920 let temp = tempfile::tempdir().expect("tempdir");
1921 let state = temp.path();
1922 drop(create_legacy_continuity(&state.join("continuity.db")));
1923 drop(Connection::open(state.join("sessions.sqlite3")).expect("sessions"));
1924
1925 let foreign_lock = meerkat_sqlite::fence_lock_path(&state.join("sessions.sqlite3"));
1928 let foreign = fs::OpenOptions::new()
1929 .read(true)
1930 .write(true)
1931 .create(true)
1932 .truncate(false)
1933 .open(&foreign_lock)
1934 .expect("open foreign lock");
1935 foreign.try_lock().expect("foreign exclusive lock");
1936
1937 let error = MobKitMaintenanceFence::acquire(state, Duration::from_millis(100))
1938 .expect_err("foreign holder must refuse acquisition");
1939 assert!(
1940 matches!(error, meerkat_sqlite::SqliteStoreError::MaintenanceFenceHeld { ref path }
1941 if path.ends_with("sessions.sqlite3")),
1942 "{error:?}"
1943 );
1944
1945 let reacquired = meerkat_sqlite::ExclusiveFence::try_acquire(&state.join("continuity.db"))
1947 .expect("try acquire");
1948 assert!(reacquired.is_some(), "partial acquisition must be released");
1949 drop(reacquired);
1950 drop(foreign);
1951
1952 let foreign = fs::OpenOptions::new()
1954 .read(true)
1955 .write(true)
1956 .create(true)
1957 .truncate(false)
1958 .open(&foreign_lock)
1959 .expect("reopen foreign lock");
1960 foreign.try_lock().expect("foreign exclusive lock");
1961 let report = migrate_state_dir(state, MigrateMode::Apply, None);
1962 assert!(report.has_errors());
1963 assert!(
1964 report.errors[0].contains("maintenance fence"),
1965 "{:?}",
1966 report.errors
1967 );
1968 drop(foreign);
1969 }
1970
1971 #[test]
1972 fn failed_canonical_fence_after_rename_fails_closed() {
1973 let temp = tempfile::tempdir().expect("tempdir");
1974 let state = temp.path();
1975 let legacy_db = state.join("continuity.db");
1976 let (sid, bytes) = legacy_session_bytes();
1977 {
1978 let conn = create_legacy_continuity(&legacy_db);
1979 insert_record(&conn, "test:alice", &sid);
1980 insert_snapshot(&conn, &sid, "test:alice", &bytes);
1981 }
1982
1983 let canonical = state.join("continuity.sqlite3");
1988 let foreign_lock = meerkat_sqlite::fence_lock_path(&canonical);
1989 let foreign = fs::OpenOptions::new()
1990 .read(true)
1991 .write(true)
1992 .create(true)
1993 .truncate(false)
1994 .open(&foreign_lock)
1995 .expect("open foreign lock");
1996 foreign.try_lock().expect("foreign exclusive lock");
1997
1998 let report = migrate_state_dir(state, MigrateMode::Apply, None);
1999
2000 assert!(canonical.is_file(), "rename itself must have happened");
2001 assert!(report.has_errors(), "{:?}", report.errors);
2002 assert!(
2003 report
2004 .errors
2005 .iter()
2006 .any(|error| error.contains("not fenced at its canonical path")),
2007 "{:?}",
2008 report.errors
2009 );
2010
2011 {
2014 let conn = Connection::open(&canonical).expect("open canonical");
2015 assert_eq!(
2016 meerkat_sqlite::domain_version(&conn, "mobkit-continuity").expect("ledger"),
2017 Some(1),
2018 "constructor must not run against an unfenced database"
2019 );
2020 }
2021 drop(foreign);
2022 }
2023
2024 #[test]
2029 fn head_canonical_continuity_bump_is_operator_gated() {
2030 let temp = tempfile::tempdir().expect("tempdir");
2031 let state = temp.path();
2032 let db = state.join("continuity.sqlite3");
2033 drop(create_legacy_continuity(&db));
2034
2035 LocalContinuityStore::open(&db).expect("gateway-style open");
2037 let probe = Connection::open(&db).expect("probe");
2038 assert_eq!(
2039 meerkat_sqlite::domain_version(&probe, "mobkit-continuity").expect("ledger"),
2040 Some(1),
2041 "launching a new gateway must not lock the previous release out of the file"
2042 );
2043 drop(probe);
2044
2045 let before = file_digest_hex(&db);
2046 let dry = migrate_state_dir(state, MigrateMode::DryRun, None);
2047 assert!(!dry.has_errors(), "{:?}", dry.errors);
2048 assert!(
2049 dry.notes
2050 .iter()
2051 .any(|note| note.contains("head-canonical") && note.contains("ONE-WAY")),
2052 "dry-run must announce the pending one-way bump: {:?}",
2053 dry.notes
2054 );
2055 assert_eq!(
2056 file_digest_hex(&db),
2057 before,
2058 "dry-run must not mutate the database"
2059 );
2060
2061 let applied = migrate_state_dir(state, MigrateMode::Apply, None);
2062 assert!(!applied.has_errors(), "{:?}", applied.errors);
2063 let entry = ledger_entry(&applied, "continuity.sqlite3", "mobkit-continuity");
2064 assert_eq!(entry.action, LedgerBaselineAction::Stamped);
2065 assert_eq!(entry.before, Some(1));
2066 assert_eq!(
2067 entry.after,
2068 Some(crate::identity_first::HEAD_CANONICAL_CONTINUITY_SCHEMA_VERSION)
2069 );
2070
2071 let second = migrate_state_dir(state, MigrateMode::Apply, None);
2073 assert!(!second.has_errors(), "{:?}", second.errors);
2074 assert_eq!(
2075 ledger_entry(&second, "continuity.sqlite3", "mobkit-continuity").action,
2076 LedgerBaselineAction::AlreadyCurrent
2077 );
2078 }
2079
2080 #[test]
2081 fn dry_run_is_a_read_only_version_matrix() {
2082 let temp = tempfile::tempdir().expect("tempdir");
2083 let state = temp.path();
2084 drop(create_legacy_continuity(&state.join("continuity.db")));
2085 {
2086 let conn = Connection::open(state.join("sessions.db")).expect("sessions");
2087 conn.execute_batch("CREATE TABLE sessions (session_id TEXT PRIMARY KEY)")
2088 .expect("sessions ddl");
2089 }
2090 drop(Connection::open(state.join(WORKGRAPH_ADMISSION_SIDECAR_FILE)).expect("sidecar"));
2091 let jobs = MobKitStorageLayout::with_injected_roots(state.to_path_buf(), None).jobs_db();
2092 drop(meerkat::SqliteDetachedJobStore::open(&jobs).expect("jobs"));
2093 let before = file_digest_hex(&state.join("continuity.db"));
2094 let jobs_before = file_digest_hex(&jobs);
2095
2096 let report = migrate_state_dir(state, MigrateMode::DryRun, None);
2097 assert!(!report.has_errors(), "{:?}", report.errors);
2098 assert_eq!(
2102 ledger_entry(&report, "continuity.db", "mobkit-continuity").action,
2103 LedgerBaselineAction::Recorded
2104 );
2105 assert_eq!(
2106 ledger_entry(&report, "sessions.db", "session-store").action,
2107 LedgerBaselineAction::ReportOnly
2108 );
2109 assert_eq!(
2110 ledger_entry(&report, "jobs.sqlite3", "jobs").action,
2111 LedgerBaselineAction::Recorded
2112 );
2113 assert_eq!(
2114 ledger_entry(
2115 &report,
2116 WORKGRAPH_ADMISSION_SIDECAR_FILE,
2117 "mobkit-workgraph-admission"
2118 )
2119 .action,
2120 LedgerBaselineAction::Exempt
2121 );
2122 assert!(
2123 report
2124 .notes
2125 .iter()
2126 .any(|note| note.contains("ledger-exempt")),
2127 "{:?}",
2128 report.notes
2129 );
2130 assert!(
2131 report
2132 .renames
2133 .iter()
2134 .any(|entry| entry.action == RenameAction::WouldRename
2135 && entry.from.ends_with("continuity.db")),
2136 "dry-run must report the pending rename: {:?}",
2137 report.renames
2138 );
2139 assert_eq!(
2140 file_digest_hex(&state.join("continuity.db")),
2141 before,
2142 "dry-run must leave the database byte-identical"
2143 );
2144 assert_eq!(
2145 file_digest_hex(&jobs),
2146 jobs_before,
2147 "dry-run must leave the inherited jobs database byte-identical"
2148 );
2149 assert!(
2150 state.join("continuity.db").is_file(),
2151 "dry-run must not rename"
2152 );
2153 }
2154
2155 #[test]
2156 fn apply_renames_with_wal_stamps_ledgers_adopts_and_is_idempotent() {
2157 let temp = tempfile::tempdir().expect("tempdir");
2158 let state = temp.path();
2159
2160 let legacy_db = state.join("continuity.db");
2164 let (sid, legacy_bytes) = legacy_session_bytes();
2165 {
2166 let conn = Connection::open(&legacy_db).expect("create fixture");
2167 conn.pragma_update(None, "journal_mode", "wal")
2168 .expect("wal mode");
2169 conn.execute_batch(LEGACY_CONTINUITY_DDL).expect("ddl");
2170 conn.execute_batch(
2173 "CREATE TABLE IF NOT EXISTS meerkat_schema (
2174 domain TEXT PRIMARY KEY,
2175 version INTEGER NOT NULL
2176 );
2177 INSERT INTO meerkat_schema (domain, version) VALUES ('mobkit-continuity', 1);",
2178 )
2179 .expect("stamp fixture ledger at the v1 floor");
2180 insert_record(&conn, "test:alice", &sid);
2181 insert_snapshot(&conn, &sid, "test:alice", &legacy_bytes);
2182 std::mem::forget(conn);
2184 }
2185 let wal = path_with_suffix(&legacy_db, "-wal");
2186 assert!(
2187 fs::metadata(&wal).map(|meta| meta.len()).unwrap_or(0) > 0,
2188 "fixture must leave a non-empty WAL"
2189 );
2190
2191 let legacy_metadata = state.join("mobkit_metadata.sqlite");
2194 drop(SqliteMetadataStore::open(&legacy_metadata).expect("metadata fixture"));
2195
2196 let legacy_memory = state.join("agent-memory-sqlite");
2198 {
2199 let store = SqliteAgentMemoryStore::open(&legacy_memory).expect("memory fixture");
2200 store.open_realm_ledgered("alpha").expect("realm fixture");
2201 }
2202
2203 let report = migrate_state_dir(state, MigrateMode::Apply, None);
2204 assert!(!report.has_errors(), "{:?}", report.errors);
2205
2206 let continuity_rename = report
2208 .renames
2209 .iter()
2210 .find(|entry| entry.slot == "continuity")
2211 .expect("continuity rename entry");
2212 assert_eq!(continuity_rename.action, RenameAction::Renamed);
2213 assert!(continuity_rename.wal_checkpointed);
2214 assert!(
2215 continuity_rename
2216 .siblings
2217 .iter()
2218 .any(|sibling| sibling.to.ends_with("continuity.sqlite3-wal")),
2219 "{:?}",
2220 continuity_rename.siblings
2221 );
2222 let marker = continuity_rename.marker.as_ref().expect("rename marker");
2223 assert!(marker.is_file());
2224 assert!(is_registered_backup_artifact_name(
2225 marker.file_name().and_then(|name| name.to_str()).unwrap()
2226 ));
2227 assert!(!legacy_db.exists());
2228 let canonical = state.join("continuity.sqlite3");
2229 assert!(canonical.is_file());
2230 let memory_rename = report
2231 .renames
2232 .iter()
2233 .find(|entry| entry.slot == "agent-memory")
2234 .expect("memory rename entry");
2235 assert_eq!(memory_rename.action, RenameAction::Renamed);
2236 assert!(state.join("agent-memory/alpha.sqlite3").is_file());
2237 assert!(!legacy_memory.exists());
2238
2239 let continuity_entry = ledger_entry(&report, "continuity.sqlite3", "mobkit-continuity");
2242 assert_eq!(continuity_entry.action, LedgerBaselineAction::Stamped);
2243 assert!(continuity_entry.after.is_some());
2244 for (file, domain) in [
2245 ("mobkit_metadata.sqlite3", "mobkit-metadata"),
2246 ("alpha.sqlite3", MEMORY_LEDGER_DOMAIN),
2247 ] {
2248 let entry = ledger_entry(&report, file, domain);
2249 assert_eq!(entry.action, LedgerBaselineAction::AlreadyCurrent, "{file}");
2250 }
2251 {
2254 let conn = Connection::open(&canonical).expect("reopen canonical");
2255 let data: Vec<u8> = conn
2256 .query_row(
2257 "SELECT data FROM session_snapshots WHERE session_id = ?1",
2258 [&sid],
2259 |row| row.get(0),
2260 )
2261 .expect("snapshot row");
2262 assert_eq!(data, legacy_bytes, "payload bytes untouched");
2263 }
2264
2265 let layout = MobKitStorageLayout::with_injected_roots(state.to_path_buf(), None);
2267 assert_eq!(
2268 layout.continuity_db().expect("resolve").provenance,
2269 DatabaseProvenance::Canonical
2270 );
2271
2272 let second = migrate_state_dir(state, MigrateMode::Apply, None);
2275 assert!(!second.has_errors(), "{:?}", second.errors);
2276 assert!(
2277 second
2278 .renames
2279 .iter()
2280 .all(|entry| entry.action != RenameAction::Renamed),
2281 "{:?}",
2282 second.renames
2283 );
2284 assert_eq!(
2285 ledger_entry(&second, "continuity.sqlite3", "mobkit-continuity").action,
2286 LedgerBaselineAction::AlreadyCurrent
2287 );
2288 }
2289
2290 #[test]
2291 fn twin_dry_run_reports_row_divergence_and_refuses() {
2292 let temp = tempfile::tempdir().expect("tempdir");
2293 let state = temp.path();
2294 let db_a = state.join("continuity.db");
2295 let db_b = state.join("continuity.sqlite3");
2296 {
2297 let conn = create_legacy_continuity(&db_a);
2298 insert_snapshot(&conn, "s-shared", "test:alice", b"same");
2299 insert_snapshot(&conn, "s-divergent", "test:alice", b"version-a");
2300 insert_snapshot(&conn, "s-only-a", "test:alice", b"solo");
2301 }
2302 {
2303 let conn = create_legacy_continuity(&db_b);
2304 insert_snapshot(&conn, "s-shared", "test:alice", b"same");
2305 insert_snapshot(&conn, "s-divergent", "test:alice", b"version-b");
2306 }
2307
2308 let report = migrate_state_dir(state, MigrateMode::DryRun, None);
2309 assert!(report.has_errors(), "twins must fail closed");
2310 assert!(
2311 report.errors[0].contains("continuity"),
2312 "{:?}",
2313 report.errors
2314 );
2315 assert!(report.ledger.is_empty());
2317 assert!(report.renames.is_empty());
2318
2319 let twin = &report.twins[0];
2320 assert_eq!(twin.slot, "continuity");
2321 assert!(!twin.byte_identical);
2322 assert_eq!(twin.rows_equal, 1, "{:?}", twin.rows);
2323 let status_of = |key: &str| {
2324 twin.rows
2325 .iter()
2326 .find(|row| row.key == format!("session_snapshots/{key}"))
2327 .map(|row| row.status.clone())
2328 };
2329 assert_eq!(status_of("s-divergent"), Some(DivergenceStatus::Divergent));
2330 assert_eq!(
2331 status_of("s-only-a"),
2332 Some(DivergenceStatus::OnlyIn { location: db_a })
2333 );
2334 assert!(matches!(
2335 twin.resolution,
2336 TwinResolution::Refused { ref reason } if reason.contains("--adopt")
2337 ));
2338 assert!(
2339 twin.notes
2340 .iter()
2341 .any(|note| note.contains("fencing tokens")),
2342 "{:?}",
2343 twin.notes
2344 );
2345 assert!(db_b.is_file());
2347 }
2348
2349 #[test]
2350 fn twin_apply_with_adopt_archives_read_only_and_canonicalizes() {
2351 let temp = tempfile::tempdir().expect("tempdir");
2352 let state = temp.path();
2353 let legacy = state.join("continuity.db");
2354 let canonical = state.join("continuity.sqlite3");
2355 {
2356 let conn = create_legacy_continuity(&legacy);
2357 insert_snapshot(&conn, "s-keep", "test:alice", b"keep-me");
2358 }
2359 {
2360 let conn = create_legacy_continuity(&canonical);
2361 insert_snapshot(&conn, "s-lose", "test:alice", b"archive-me");
2362 }
2363 let other_digest = file_digest_hex(&canonical);
2364
2365 let refused = migrate_state_dir(state, MigrateMode::Apply, None);
2367 assert!(refused.has_errors());
2368
2369 let report = migrate_state_dir(state, MigrateMode::Apply, Some(&legacy));
2370 assert!(!report.has_errors(), "{:?}", report.errors);
2371 let twin = &report.twins[0];
2372 let TwinResolution::Adopted { adopted, archived } = &twin.resolution else {
2373 panic!("expected adoption, got {:?}", twin.resolution);
2374 };
2375 assert_eq!(
2376 adopted, &canonical,
2377 "adopted copy lands at the canonical name"
2378 );
2379 assert_eq!(archived.len(), 1);
2380 let archive = &archived[0];
2381 assert!(archive.is_file());
2382 let archive_name = archive.file_name().and_then(|name| name.to_str()).unwrap();
2383 assert!(
2384 is_registered_backup_artifact_name(archive_name),
2385 "{archive_name}"
2386 );
2387 assert!(archive_name.ends_with(".twin"), "{archive_name}");
2388 assert!(
2389 fs::metadata(archive)
2390 .expect("archive meta")
2391 .permissions()
2392 .readonly(),
2393 "archive must be read-only"
2394 );
2395 assert_eq!(
2396 file_digest_hex(archive),
2397 other_digest,
2398 "archived content preserved exactly"
2399 );
2400
2401 let conn = Connection::open(&canonical).expect("reopen canonical");
2404 let data: Vec<u8> = conn
2405 .query_row(
2406 "SELECT data FROM session_snapshots WHERE session_id = 's-keep'",
2407 [],
2408 |row| row.get(0),
2409 )
2410 .expect("adopted row");
2411 assert_eq!(data, b"keep-me");
2412 drop(conn);
2413 assert!(!legacy.exists());
2414 let layout = MobKitStorageLayout::with_injected_roots(state.to_path_buf(), None);
2415 assert!(layout.continuity_db().is_ok());
2416 let again = migrate_state_dir(state, MigrateMode::DryRun, None);
2417 assert!(again.twins.is_empty(), "{:?}", again.twins);
2418 }
2419
2420 #[test]
2421 fn byte_identical_twins_dedup_under_plain_apply() {
2422 let temp = tempfile::tempdir().expect("tempdir");
2423 let state = temp.path();
2424 let canonical = state.join("mobkit_metadata.sqlite3");
2425 drop(SqliteMetadataStore::open(&canonical).expect("metadata fixture"));
2426 let legacy = state.join("mobkit_metadata.sqlite");
2427 fs::copy(&canonical, &legacy).expect("copy twin");
2428
2429 let report = migrate_state_dir(state, MigrateMode::Apply, None);
2430 assert!(!report.has_errors(), "{:?}", report.errors);
2431 let twin = report
2432 .twins
2433 .iter()
2434 .find(|twin| twin.slot == "metadata")
2435 .expect("metadata twin");
2436 assert!(twin.byte_identical);
2437 let TwinResolution::Deduped { kept, archived } = &twin.resolution else {
2438 panic!("expected dedup, got {:?}", twin.resolution);
2439 };
2440 assert_eq!(kept, &canonical);
2441 assert!(
2446 archived[0]
2447 .file_name()
2448 .and_then(|name| name.to_str())
2449 .unwrap()
2450 .ends_with(".twin-dedup")
2451 );
2452 assert!(
2453 archived.iter().skip(1).all(|path| {
2454 let name = path.file_name().and_then(|name| name.to_str()).unwrap();
2455 name.ends_with("-wal") || name.ends_with("-shm")
2456 }),
2457 "{archived:?}"
2458 );
2459 assert!(canonical.is_file());
2460 assert!(!legacy.exists());
2461 }
2462
2463 #[test]
2464 fn adopt_path_outside_the_twin_refuses() {
2465 let temp = tempfile::tempdir().expect("tempdir");
2466 let state = temp.path();
2467 {
2468 let conn = create_legacy_continuity(&state.join("continuity.db"));
2469 insert_snapshot(&conn, "s-a", "test:alice", b"a");
2470 }
2471 {
2472 let conn = create_legacy_continuity(&state.join("continuity.sqlite3"));
2473 insert_snapshot(&conn, "s-b", "test:alice", b"b");
2474 }
2475 let elsewhere = state.join("unrelated.db");
2476 fs::write(&elsewhere, b"x").expect("unrelated");
2477
2478 let report = migrate_state_dir(state, MigrateMode::Apply, Some(&elsewhere));
2479 assert!(report.has_errors());
2480 assert!(matches!(
2481 report.twins[0].resolution,
2482 TwinResolution::Refused { ref reason } if reason.contains("candidates")
2483 ));
2484 assert!(state.join("continuity.db").is_file(), "nothing moved");
2485 assert!(state.join("continuity.sqlite3").is_file());
2486 }
2487
2488 #[test]
2489 fn leftovers_census_reports_blobs_sidecar_artifacts_and_dead_registry_entries() {
2490 let temp = tempfile::tempdir().expect("tempdir");
2491 let state = temp.path();
2492 let shard = state.join("blobs").join("aa");
2494 fs::create_dir_all(&shard).expect("shard");
2495 fs::write(shard.join(format!("{}.json", "a".repeat(64))), b"{}").expect("legacy blob");
2496 drop(Connection::open(state.join(WORKGRAPH_ADMISSION_SIDECAR_FILE)).expect("sidecar"));
2497 fs::write(state.join("continuity.db.pre-0.0.1-1700000000"), b"backup")
2498 .expect("backup artifact");
2499 let mut child = std::process::Command::new("true").spawn().expect("spawn");
2501 let dead_pid = child.id();
2502 child.wait().expect("reap");
2503 fs::write(
2504 state.join(RUNTIME_REGISTRY_FILE_NAME),
2505 serde_json::to_vec(&serde_json::json!({
2506 "entries": [
2507 { "key": "k1", "runtime_id": "tux-dead", "http_base_url": "http://127.0.0.1:1",
2508 "pid": dead_pid, "updated_at_ms": 0 },
2509 { "key": "k2", "runtime_id": "tux-live", "http_base_url": "http://127.0.0.1:2",
2510 "pid": std::process::id(), "updated_at_ms": 0 },
2511 ]
2512 }))
2513 .expect("registry json"),
2514 )
2515 .expect("write registry");
2516
2517 let report = migrate_state_dir(state, MigrateMode::DryRun, None);
2518 assert!(!report.has_errors(), "{:?}", report.errors);
2519 let codes: Vec<&str> = report
2520 .findings
2521 .iter()
2522 .map(|finding| finding.code.as_str())
2523 .collect();
2524 for expected in [
2525 storage_doctor::FINDING_LEGACY_FS_BLOBS,
2526 storage_doctor::FINDING_WORKGRAPH_ADMISSION_SIDECAR,
2527 storage_doctor::FINDING_BACKUP_ARTIFACT,
2528 FINDING_DEAD_RUNTIME_REGISTRY_ENTRY,
2529 ] {
2530 assert!(codes.contains(&expected), "missing {expected}: {codes:?}");
2531 }
2532 let dead: Vec<_> = report
2533 .findings
2534 .iter()
2535 .filter(|finding| finding.code == FINDING_DEAD_RUNTIME_REGISTRY_ENTRY)
2536 .collect();
2537 assert_eq!(dead.len(), 1, "only the dead entry is flagged: {dead:?}");
2538 assert!(dead[0].message.contains("tux-dead"));
2539 }
2540
2541 #[test]
2542 fn strict_artifact_name_validation_rejects_lookalikes() {
2543 for valid in [
2546 backup_artifact_name("sessions.sqlite3", "").as_str(),
2547 backup_artifact_name("continuity.db", "renamed").as_str(),
2548 backup_artifact_name("team", "split-brain").as_str(),
2549 "sessions.sqlite3.pre-0.8.3-1700000000",
2550 "continuity.db.pre-0.8.2-1700000000.renamed",
2551 "mobkit_metadata.sqlite.pre-0.8.3-123.twin-dedup-wal",
2552 "agent-memory-sqlite.pre-0.0.1-1700000000.twin",
2553 ] {
2554 assert!(is_registered_backup_artifact_name(valid), "{valid}");
2555 }
2556 for invalid in [
2559 "notes.pre-release",
2560 "data.pre-view.txt",
2561 ".pre-0.8.3-1700000000",
2562 "foo.pre-1-1700000000",
2563 "foo.pre-0..3-1700000000",
2564 "foo.pre-0.8.3-17000x",
2565 "foo.pre-0.8.3-",
2566 "foo.pre-0.8.3-1700000000.",
2567 "sessions.sqlite3",
2568 ] {
2569 assert!(!is_registered_backup_artifact_name(invalid), "{invalid}");
2570 }
2571 for valid in [
2572 "alpha.sqlite3.corrupt-42",
2573 "session_index.sqlite3.corrupt-1700000000",
2574 ] {
2575 assert!(is_registered_quarantine_artifact_name(valid), "{valid}");
2576 }
2577 for invalid in [
2578 "report.corrupt-12a",
2579 "x.corrupt-",
2580 ".corrupt-42",
2581 "a.corrupt-1.bak",
2582 "notes.corrupted-1",
2583 ] {
2584 assert!(
2585 !is_registered_quarantine_artifact_name(invalid),
2586 "{invalid}"
2587 );
2588 }
2589 }
2590
2591 #[test]
2592 fn prune_respects_threshold_and_registered_patterns_only() {
2593 let temp = tempfile::tempdir().expect("tempdir");
2594 let state = temp.path();
2595 let old_backup = state.join("continuity.db.pre-0.0.1-1700000000.renamed");
2596 fs::write(&old_backup, b"old").expect("old backup");
2597 let old_time = std::time::SystemTime::now() - Duration::from_hours(40 * 24);
2598 fs::File::options()
2599 .write(true)
2600 .open(&old_backup)
2601 .expect("open for mtime")
2602 .set_modified(old_time)
2603 .expect("set mtime");
2604 let old_dir = state.join("agent-memory-sqlite.pre-0.0.1-1700000000.twin");
2606 fs::create_dir_all(&old_dir).expect("old dir");
2607 fs::write(old_dir.join("alpha.sqlite3"), b"realm").expect("realm bytes");
2608 fs::File::options()
2609 .write(true)
2610 .open(old_dir.join("alpha.sqlite3"))
2611 .expect("open for mtime")
2612 .set_modified(old_time)
2613 .expect("set inner mtime");
2614 fs::File::open(&old_dir)
2617 .expect("open dir")
2618 .set_modified(old_time)
2619 .expect("set dir mtime");
2620 let young_quarantine = state.join("alpha.sqlite3.corrupt-42");
2621 fs::write(&young_quarantine, b"young").expect("young quarantine");
2622 let distractor = state.join("notes.txt");
2623 fs::write(&distractor, b"keep me").expect("distractor");
2624 let lookalike_backup = state.join("notes.pre-release");
2627 let lookalike_quarantine = state.join("report.corrupt-12a");
2628 for lookalike in [&lookalike_backup, &lookalike_quarantine] {
2629 fs::write(lookalike, b"user file").expect("lookalike");
2630 fs::File::options()
2631 .write(true)
2632 .open(lookalike)
2633 .expect("open for mtime")
2634 .set_modified(old_time)
2635 .expect("set lookalike mtime");
2636 }
2637
2638 let dry = prune_state_dir(state, 30, MigrateMode::DryRun);
2639 assert!(!dry.has_errors());
2640 assert_eq!(dry.artifacts.len(), 3, "{:?}", dry.artifacts);
2641 let action_of = |report: &MobKitPruneReport, path: &Path| {
2642 report
2643 .artifacts
2644 .iter()
2645 .find(|artifact| artifact.path == path)
2646 .map(|artifact| artifact.action)
2647 .unwrap_or_else(|| panic!("no artifact entry for {}", path.display()))
2648 };
2649 assert_eq!(action_of(&dry, &old_backup), PruneAction::WouldDelete);
2650 assert_eq!(action_of(&dry, &old_dir), PruneAction::WouldDelete);
2651 assert_eq!(action_of(&dry, &young_quarantine), PruneAction::Kept);
2652 assert!(old_backup.is_file(), "dry-run deletes nothing");
2653
2654 let applied = prune_state_dir(state, 30, MigrateMode::Apply);
2655 assert!(!applied.has_errors(), "{:?}", applied.errors);
2656 assert_eq!(action_of(&applied, &old_backup), PruneAction::Deleted);
2657 assert_eq!(action_of(&applied, &old_dir), PruneAction::Deleted);
2658 assert!(!old_backup.exists());
2659 assert!(!old_dir.exists());
2660 assert!(young_quarantine.is_file(), "young artifacts are kept");
2661 assert!(distractor.is_file(), "unregistered names are never touched");
2662
2663 let sweep = prune_state_dir(state, 0, MigrateMode::Apply);
2665 assert!(!sweep.has_errors());
2666 assert!(!young_quarantine.exists());
2667 assert!(
2668 lookalike_backup.is_file() && lookalike_quarantine.is_file(),
2669 "names outside the registered full shape are never enumerated or deleted"
2670 );
2671 }
2672
2673 #[test]
2674 fn report_shapes_round_trip_through_json() {
2675 let mut report = MobKitMigrateReport::new(MigrateMode::Apply, Path::new("/state"));
2676 report.ledger.push(LedgerBaselineEntry {
2677 database: PathBuf::from("/state/continuity.sqlite3"),
2678 domain: "mobkit-continuity".to_string(),
2679 before: None,
2680 after: Some(1),
2681 action: LedgerBaselineAction::Stamped,
2682 });
2683 report.renames.push(FileRenameEntry {
2684 slot: "continuity".to_string(),
2685 from: PathBuf::from("/state/continuity.db"),
2686 to: PathBuf::from("/state/continuity.sqlite3"),
2687 siblings: vec![SiblingRename {
2688 from: PathBuf::from("/state/continuity.db-wal"),
2689 to: PathBuf::from("/state/continuity.sqlite3-wal"),
2690 }],
2691 wal_checkpointed: true,
2692 marker: Some(PathBuf::from("/state/continuity.db.pre-0.8.2-1.renamed")),
2693 action: RenameAction::Renamed,
2694 });
2695 report.twins.push(TwinReport {
2696 slot: "console".to_string(),
2697 paths: vec![PathBuf::from("/state/mobkit_console.sqlite")],
2698 byte_identical: false,
2699 rows_equal: 2,
2700 rows: vec![RowDivergenceEntry {
2701 key: "console_frames/7".to_string(),
2702 status: DivergenceStatus::Divergent,
2703 }],
2704 resolution: TwinResolution::Adopted {
2705 adopted: PathBuf::from("/state/mobkit_console.sqlite3"),
2706 archived: vec![PathBuf::from(
2707 "/state/mobkit_console.sqlite.pre-0.8.2-1.twin",
2708 )],
2709 },
2710 notes: vec![],
2711 errors: vec![],
2712 });
2713 let json = serde_json::to_string(&report).expect("serialize");
2714 let parsed: MobKitMigrateReport = serde_json::from_str(&json).expect("deserialize");
2715 assert!(matches!(parsed.mode, MigrateMode::Apply));
2716 assert_eq!(parsed.ledger[0].action, LedgerBaselineAction::Stamped);
2717 assert_eq!(parsed.renames[0].action, RenameAction::Renamed);
2718 assert!(matches!(
2719 parsed.twins[0].resolution,
2720 TwinResolution::Adopted { .. }
2721 ));
2722 assert!(!parsed.has_errors());
2723
2724 let sparse: MobKitMigrateReport =
2726 serde_json::from_str(r#"{"future_field":1}"#).expect("sparse migrate");
2727 assert!(matches!(sparse.mode, MigrateMode::DryRun));
2728 let sparse: MobKitPruneReport =
2729 serde_json::from_str(r#"{"future_field":1}"#).expect("sparse prune");
2730 assert!(sparse.artifacts.is_empty());
2731 }
2732
2733 #[test]
2734 fn missing_state_dir_errors_typed() {
2735 let temp = tempfile::tempdir().expect("tempdir");
2736 let missing = temp.path().join("nope");
2737 let report = migrate_state_dir(&missing, MigrateMode::DryRun, None);
2738 assert!(report.has_errors());
2739 assert!(report.errors[0].contains("does not exist"));
2740 }
2741}