From a9b4f16c5eea41c2998622790f301c0209b9a2f2 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 10 Jul 2026 11:25:43 +0000 Subject: [PATCH] fix(storage): preserve divergent session variants --- src/migrate/consolidate/finalize.rs | 2 +- src/migrate/consolidate/mod.rs | 63 +- src/migrate/consolidate/sqlite.rs | 361 ++++++++-- src/migrate/consolidate/sqlite/inspect.rs | 94 ++- src/migrate/consolidate/sqlite/verify.rs | 133 +++- src/migrate/consolidate/tests.rs | 792 ++++++++++++++++++++++ 6 files changed, 1360 insertions(+), 85 deletions(-) diff --git a/src/migrate/consolidate/finalize.rs b/src/migrate/consolidate/finalize.rs index fcc75b9a..e7ce230f 100644 --- a/src/migrate/consolidate/finalize.rs +++ b/src/migrate/consolidate/finalize.rs @@ -80,9 +80,9 @@ pub(super) async fn verify_destination( &source_input, &target_input, &destination_snapshots, - &sessions, destination, session_offsets, + &resolved.report.source.project_id, ) .await?; Ok(()) diff --git a/src/migrate/consolidate/mod.rs b/src/migrate/consolidate/mod.rs index 757a527b..51ab2d11 100644 --- a/src/migrate/consolidate/mod.rs +++ b/src/migrate/consolidate/mod.rs @@ -41,7 +41,7 @@ use crate::storage::{ self, EnrollmentMarker, PrivateStoreIo, StorageMode, StoreKind, StoreLayout, StoreManifest, }; -const LEDGER_SCHEMA_VERSION: u32 = 1; +const LEDGER_SCHEMA_VERSION: u32 = 2; const BACKUP_DIR: &str = "migration-backups"; const LEDGER_DIR: &str = "migration-inventory"; const PRESERVED_DIR: &str = "consolidation-preserved"; @@ -88,6 +88,16 @@ pub struct CollisionSummary { pub session_overlaps: u64, pub message_overlaps: u64, pub lcm_message_overlaps: u64, + #[serde(default)] + pub divergent_lcm_messages: u64, + #[serde(default)] + pub divergent_lcm_session_ids: u64, + #[serde(default)] + pub divergent_lcm_content_hashes: u64, + #[serde(default)] + pub divergent_lcm_storage_kinds: u64, + #[serde(default)] + pub divergent_lcm_payload_refs: u64, pub artifact_path_overlaps: usize, pub differing_artifact_paths: Vec, pub semantics: Vec, @@ -697,12 +707,19 @@ async fn collision_summary( session_overlaps: db.sessions, message_overlaps: db.messages, lcm_message_overlaps: db.lcm_messages, + divergent_lcm_messages: db.divergent_lcm_messages, + divergent_lcm_session_ids: db.divergent_lcm_session_ids, + divergent_lcm_content_hashes: db.divergent_lcm_content_hashes, + divergent_lcm_storage_kinds: db.divergent_lcm_storage_kinds, + divergent_lcm_payload_refs: db.divergent_lcm_payload_refs, artifact_path_overlaps: overlaps.len(), differing_artifact_paths: differing, semantics: vec![ "facts: union by content; tags/entities/metadata are merged, counters take maxima, newest trust/category wins, feedback events are deduplicated".to_string(), "sessions: union by provider/session id; time bounds widen and non-null target fields win".to_string(), - "messages and LCM payload identities: identical rows deduplicate; divergent content is a hard error".to_string(), + "session-message projections: the selected target row remains canonical; divergent source rows and their overlapping parent-linked session family are preserved as active consolidated// variants".to_string(), + "LCM raw messages: the selected target row remains canonical; source rows receive active variants only for content-hash divergence, while equal-content representation drift deduplicates to the target raw family".to_string(), + "LCM external payloads and summaries: identical identities deduplicate; divergent content is a hard error".to_string(), "branch graphs: target branches retain their names; every source branch is preserved under consolidated//...".to_string(), "artifact paths: identical files deduplicate; divergent non-reference files are preserved under consolidation-preserved; divergent payload/handle files are a hard error".to_string(), ], @@ -773,7 +790,23 @@ fn fingerprint_inputs( } fn load_or_create_ledger(resolved: &ResolvedPlan, path: &Path) -> Result { - if let Some(ledger) = load_ledger(path)? { + if let Some(mut ledger) = load_ledger(path)? { + validate_ledger_inventory(&ledger, resolved)?; + if ledger.schema_version == 1 { + if !matches!( + ledger.state, + ConsolidationState::Planned + | ConsolidationState::BackupsReady + | ConsolidationState::DestinationReady + ) { + return Err(config_error(format!( + "consolidation ledger v1 cannot be migrated safely after database merge state {:?}", + ledger.state + ))); + } + ledger.schema_version = LEDGER_SCHEMA_VERSION; + save_ledger(path, &ledger)?; + } return Ok(ledger); } if resolved.report.destination_data_root.exists() { @@ -802,13 +835,24 @@ fn load_or_create_ledger(resolved: &ResolvedPlan, path: &Path) -> Result Result<()> { - if ledger.schema_version != LEDGER_SCHEMA_VERSION - || ledger.migration_id != resolved.report.migration_id + validate_ledger_inventory(ledger, resolved)?; + if ledger.schema_version != LEDGER_SCHEMA_VERSION { + return Err(config_error(format!( + "unsupported consolidation ledger schema version {}; expected {}", + ledger.schema_version, LEDGER_SCHEMA_VERSION + ))); + } + Ok(()) +} + +fn validate_ledger_inventory(ledger: &ConsolidationLedger, resolved: &ResolvedPlan) -> Result<()> { + if ledger.migration_id != resolved.report.migration_id || ledger.confirmation_token != resolved.report.confirmation_token || ledger.input_fingerprint != resolved.input_fingerprint || ledger.source_project_id != resolved.report.source.project_id || ledger.target_project_id != resolved.report.target.project_id || ledger.destination_project_id != resolved.report.destination_project_id + || !same_path(&ledger.project_root, &resolved.report.project_root) || !same_path(&ledger.git_common_dir, &resolved.report.git_common_dir) { return Err(config_error( @@ -881,7 +925,14 @@ async fn merge_databases(resolved: &ResolvedPlan, ledger: &mut ConsolidationLedg .session_offsets .as_ref() .ok_or_else(|| config_error("session merge offsets are missing from the ledger"))?; - sqlite::merge_sessions(&target_sessions, &source_sessions, offsets).await?; + sqlite::merge_sessions( + &target_sessions, + &source_sessions, + &target_input, + &resolved.report.source.project_id, + offsets, + ) + .await?; Ok(()) } diff --git a/src/migrate/consolidate/sqlite.rs b/src/migrate/consolidate/sqlite.rs index 87b77f40..3262f261 100644 --- a/src/migrate/consolidate/sqlite.rs +++ b/src/migrate/consolidate/sqlite.rs @@ -19,6 +19,18 @@ pub(super) use inspect::{ }; pub(super) use verify::verify_session_union_sql; +pub(super) const LCM_RAW_MESSAGE_DIVERGENCE_PREDICATE: &str = + "t.session_id IS NOT s.session_id OR t.content_hash IS NOT s.content_hash + OR t.storage_kind IS NOT s.storage_kind OR t.payload_ref IS NOT s.payload_ref"; +pub(super) const LCM_CONTENT_HASH_DIVERGENCE_PREDICATE: &str = + "t.content_hash IS NOT s.content_hash"; +const SESSION_MESSAGE_DIVERGENCE_PREDICATE: &str = + "t.session_id IS NOT s.session_id OR t.role IS NOT s.role + OR t.timestamp IS NOT s.timestamp OR t.ordinal IS NOT s.ordinal + OR t.text IS NOT s.text OR t.kind IS NOT s.kind OR t.model IS NOT s.model + OR t.tool_names IS NOT s.tool_names OR t.source_path IS NOT s.source_path + OR t.source_offset IS NOT s.source_offset OR t.metadata_json IS NOT s.metadata_json"; + #[derive(Debug, Clone, Serialize, Deserialize)] pub(super) struct GraphMergeOffsets { pub source_path: PathBuf, @@ -311,6 +323,8 @@ pub(super) async fn plan_session_offsets( pub(super) async fn merge_sessions( target_path: &Path, source_path: &Path, + target_input_path: &Path, + source_project_id: &str, offsets: &SessionMergeOffsets, ) -> Result<()> { normalize_sessions(target_path).await?; @@ -319,7 +333,8 @@ pub(super) async fn merge_sessions( .await .ok_or_else(|| db_message("merge_sessions", "could not open target sessions DB"))?; attach_as(target.conn(), source_path, "source").await?; - reject_session_content_collisions(target.conn()).await?; + attach_as(target.conn(), target_input_path, "target_input").await?; + reject_session_content_collisions(target.conn(), "source", "target_input").await?; target .conn() .execute("PRAGMA foreign_keys = OFF", ()) @@ -330,7 +345,17 @@ pub(super) async fn merge_sessions( .execute("BEGIN IMMEDIATE", ()) .await .map_err(|error| db_error("merge_sessions", error))?; - let result = merge_sessions_tx(target.conn(), offsets).await; + let result = match build_consolidation_message_map( + target.conn(), + "source", + "target_input", + source_project_id, + ) + .await + { + Ok(()) => merge_sessions_tx(target.conn(), offsets).await, + Err(error) => Err(error), + }; match result { Ok(()) => target .conn() @@ -340,6 +365,10 @@ pub(super) async fn merge_sessions( Err(error) => { let _ = target.conn().execute("ROLLBACK", ()).await; let _ = target.conn().execute("DETACH DATABASE source", ()).await; + let _ = target + .conn() + .execute("DETACH DATABASE target_input", ()) + .await; return Err(error); } }; @@ -348,6 +377,11 @@ pub(super) async fn merge_sessions( .execute("DETACH DATABASE source", ()) .await .map_err(|error| db_error("merge_sessions", error))?; + target + .conn() + .execute("DETACH DATABASE target_input", ()) + .await + .map_err(|error| db_error("merge_sessions", error))?; crate::sessions::lcm::schema::rebuild_raw_fts(target.conn()) .await .ok_or_else(|| db_message("merge_sessions", "could not rebuild raw-message FTS"))?; @@ -418,35 +452,26 @@ async fn reject_session_registry_rows(path: &Path) -> Result<()> { Ok(()) } -async fn reject_session_content_collisions(conn: &Connection) -> Result<()> { +async fn reject_session_content_collisions( + conn: &Connection, + source_schema: &str, + target_schema: &str, +) -> Result<()> { + let source = quote_identifier(source_schema); + let target = quote_identifier(target_schema); + let external_payloads = format!( + "SELECT COUNT(*) FROM {source}.lcm_external_payloads s + JOIN {target}.lcm_external_payloads t ON t.payload_ref=s.payload_ref + WHERE t.content_hash IS NOT s.content_hash OR t.byte_count IS NOT s.byte_count" + ); + let summary_nodes = format!( + "SELECT COUNT(*) FROM {source}.lcm_summary_nodes s + JOIN {target}.lcm_summary_nodes t ON t.node_id=s.node_id + WHERE t.summary_hash IS NOT s.summary_hash OR t.summary_text IS NOT s.summary_text" + ); for (label, sql) in [ - ( - "session message", - "SELECT COUNT(*) FROM source.session_messages s - JOIN session_messages t ON t.provider=s.provider AND t.message_id=s.message_id - WHERE t.session_id IS NOT s.session_id OR t.role IS NOT s.role - OR t.ordinal IS NOT s.ordinal OR t.text IS NOT s.text - OR t.kind IS NOT s.kind OR t.model IS NOT s.model", - ), - ( - "LCM raw message", - "SELECT COUNT(*) FROM source.lcm_raw_messages s - JOIN lcm_raw_messages t ON t.provider=s.provider AND t.message_id=s.message_id - WHERE t.session_id IS NOT s.session_id OR t.content_hash IS NOT s.content_hash - OR t.storage_kind IS NOT s.storage_kind OR t.payload_ref IS NOT s.payload_ref", - ), - ( - "LCM external payload", - "SELECT COUNT(*) FROM source.lcm_external_payloads s - JOIN lcm_external_payloads t ON t.payload_ref=s.payload_ref - WHERE t.content_hash IS NOT s.content_hash OR t.byte_count IS NOT s.byte_count", - ), - ( - "LCM summary node", - "SELECT COUNT(*) FROM source.lcm_summary_nodes s - JOIN lcm_summary_nodes t ON t.node_id=s.node_id - WHERE t.summary_hash IS NOT s.summary_hash OR t.summary_text IS NOT s.summary_text", - ), + ("LCM external payload", external_payloads.as_str()), + ("LCM summary node", summary_nodes.as_str()), ] { let count = query_i64(conn, sql).await?; if count > 0 { @@ -461,7 +486,207 @@ async fn reject_session_content_collisions(conn: &Connection) -> Result<()> { Ok(()) } +pub(super) async fn build_consolidation_message_map( + conn: &Connection, + source_schema: &str, + target_schema: &str, + source_project_id: &str, +) -> Result<()> { + conn.execute_batch( + "PRAGMA query_only = OFF; + CREATE TEMP TABLE IF NOT EXISTS consolidation_message_map( + provider TEXT NOT NULL, + original_id TEXT NOT NULL, + mapped_id TEXT NOT NULL, + session_divergent INTEGER NOT NULL, + raw_content_divergent INTEGER NOT NULL, + PRIMARY KEY(provider, original_id), + UNIQUE(provider, mapped_id) + );", + ) + .await + .map_err(|error| db_error("message_variant_map_init", error))?; + conn.execute("DELETE FROM temp.consolidation_message_map", ()) + .await + .map_err(|error| db_error("message_variant_map_reset", error))?; + let source = quote_identifier(source_schema); + let target = quote_identifier(target_schema); + let sql = format!( + "INSERT INTO consolidation_message_map( + provider, original_id, mapped_id, session_divergent, raw_content_divergent + ) + SELECT provider, message_id, 'consolidated/' || ?1 || '/' || message_id, + MAX(session_divergent), MAX(raw_content_divergent) + FROM ( + SELECT s.provider, s.message_id, 1 AS session_divergent, + 0 AS raw_content_divergent + FROM {source}.session_messages s + JOIN {target}.session_messages t + ON t.provider=s.provider AND t.message_id=s.message_id + WHERE {SESSION_MESSAGE_DIVERGENCE_PREDICATE} + UNION + SELECT s.provider, s.message_id, 0 AS session_divergent, + 1 AS raw_content_divergent + FROM {source}.lcm_raw_messages s + JOIN {target}.lcm_raw_messages t + ON t.provider=s.provider AND t.message_id=s.message_id + WHERE {LCM_CONTENT_HASH_DIVERGENCE_PREDICATE} + ) GROUP BY provider, message_id" + ); + conn.execute(&sql, params![source_project_id]) + .await + .map_err(|error| db_error("message_variant_map_fill", error))?; + let session_family_sql = format!( + "WITH RECURSIVE variant_family(provider, message_id) AS ( + SELECT provider, original_id FROM consolidation_message_map + UNION + SELECT child.provider, child.message_id + FROM {source}.session_messages child + JOIN {target}.session_messages target_child + ON target_child.provider=child.provider + AND target_child.message_id=child.message_id + JOIN variant_family parent + ON parent.provider=child.provider + AND CASE WHEN json_valid(child.metadata_json) + THEN json_extract(child.metadata_json, '$.parent_message_id') END + =parent.message_id + ) + INSERT OR IGNORE INTO consolidation_message_map( + provider, original_id, mapped_id, session_divergent, raw_content_divergent + ) + SELECT provider, message_id, 'consolidated/' || ?1 || '/' || message_id, 1, 0 + FROM variant_family WHERE 1 + ON CONFLICT(provider, original_id) DO UPDATE SET + session_divergent=MAX(session_divergent, excluded.session_divergent), + raw_content_divergent=MAX(raw_content_divergent, excluded.raw_content_divergent)" + ); + conn.execute(&session_family_sql, params![source_project_id]) + .await + .map_err(|error| db_error("message_session_variant_family", error))?; + let collision_sql = format!( + "SELECT COUNT(*) FROM consolidation_message_map m + WHERE EXISTS (SELECT 1 FROM {source}.session_messages x + WHERE x.provider=m.provider AND x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {source}.lcm_raw_messages x + WHERE x.provider=m.provider AND x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {target}.session_messages x + WHERE x.provider=m.provider AND x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {target}.lcm_raw_messages x + WHERE x.provider=m.provider AND x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {source}.lcm_external_payloads x + WHERE x.provider=m.provider AND x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {target}.lcm_external_payloads x + WHERE x.provider=m.provider AND x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {source}.dashboard_token_counts x + WHERE x.provider=m.provider AND x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {target}.dashboard_token_counts x + WHERE x.provider=m.provider AND x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {source}.turns x WHERE x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {target}.turns x WHERE x.message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {source}.commit_sessions x + WHERE x.provider=m.provider AND x.evidence_message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {target}.commit_sessions x + WHERE x.provider=m.provider AND x.evidence_message_id=m.mapped_id) + OR EXISTS (SELECT 1 FROM {source}.session_messages x + WHERE x.provider=m.provider + AND CASE WHEN json_valid(x.metadata_json) + THEN json_extract(x.metadata_json, '$.parent_message_id') END + =m.mapped_id) + OR EXISTS (SELECT 1 FROM {target}.session_messages x + WHERE x.provider=m.provider + AND CASE WHEN json_valid(x.metadata_json) + THEN json_extract(x.metadata_json, '$.parent_message_id') END + =m.mapped_id) + OR EXISTS (SELECT 1 FROM {source}.lcm_raw_messages x + WHERE x.provider=m.provider + AND CASE WHEN json_valid(x.metadata_json) + THEN json_extract(x.metadata_json, '$.parent_message_id') END + =m.mapped_id) + OR EXISTS (SELECT 1 FROM {target}.lcm_raw_messages x + WHERE x.provider=m.provider + AND CASE WHEN json_valid(x.metadata_json) + THEN json_extract(x.metadata_json, '$.parent_message_id') END + =m.mapped_id)" + ); + let collisions = query_i64(conn, &collision_sql).await?; + if collisions != 0 { + return Err(db_message( + "message_variant_map", + format!( + "{collisions} synthetic consolidation message key collision(s); inputs and backups were preserved" + ), + )); + } + let ambiguous_turns = query_i64( + conn, + &format!( + "SELECT COUNT(*) FROM {source}.turns tr + WHERE EXISTS ( + SELECT 1 FROM {source}.session_messages sm + JOIN consolidation_message_map m + ON m.provider=sm.provider AND m.original_id=sm.message_id + WHERE sm.message_id=tr.message_id AND sm.session_id=tr.session_id + ) AND ( + SELECT COUNT(*) FROM {source}.session_messages sm + WHERE sm.message_id=tr.message_id AND sm.session_id=tr.session_id + ) != 1" + ), + ) + .await?; + if ambiguous_turns != 0 { + return Err(db_message( + "message_variant_map", + format!( + "{ambiguous_turns} source turn message mapping ambiguity collision(s); inputs and backups were preserved" + ), + )); + } + Ok(()) +} + +pub(super) fn mapped_parent_metadata(alias: &str, raw_family_only: bool) -> String { + let family_filter = if raw_family_only { + " AND parent_map.raw_content_divergent=1" + } else { + "" + }; + format!( + "CASE + WHEN NOT json_valid({alias}.metadata_json) + THEN {alias}.metadata_json + ELSE COALESCE(( + SELECT json_set( + {alias}.metadata_json, + '$.consolidation_original_parent_message_id', + json_extract({alias}.metadata_json, '$.parent_message_id'), + '$.parent_message_id', parent_map.mapped_id + ) + FROM consolidation_message_map parent_map + WHERE parent_map.provider={alias}.provider + AND parent_map.original_id=json_extract( + {alias}.metadata_json, '$.parent_message_id' + ){family_filter} + ), {alias}.metadata_json) + END" + ) +} + +pub(super) fn mapped_turn_message_id(alias: &str, source_schema: &str) -> String { + format!( + "COALESCE(( + SELECT m.mapped_id + FROM {source_schema}.session_messages sm + JOIN consolidation_message_map m + ON m.provider=sm.provider AND m.original_id=sm.message_id + WHERE sm.message_id={alias}.message_id AND sm.session_id={alias}.session_id + ), {alias}.message_id)" + ) +} + async fn merge_sessions_tx(conn: &Connection, offsets: &SessionMergeOffsets) -> Result<()> { + let session_metadata = mapped_parent_metadata("s", false); + let raw_metadata = mapped_parent_metadata("s", true); + let turn_message_id = mapped_turn_message_id("s", "source"); conn.execute_batch(&format!( "CREATE TEMP TABLE IF NOT EXISTS consolidation_raw_map( source_id INTEGER PRIMARY KEY, target_id INTEGER NOT NULL @@ -478,9 +703,9 @@ async fn merge_sessions_tx(conn: &Connection, offsets: &SessionMergeOffsets) -> message_id, project_hash, session_id, model, timestamp, input_tokens, output_tokens, cache_write_tokens, cache_read_tokens, cost_usd, category, tool_names - ) SELECT message_id, project_hash, session_id, model, timestamp, input_tokens, - output_tokens, cache_write_tokens, cache_read_tokens, cost_usd, - category, tool_names FROM source.turns; + ) SELECT {turn_message_id}, s.project_hash, s.session_id, s.model, s.timestamp, + s.input_tokens, s.output_tokens, s.cache_write_tokens, s.cache_read_tokens, + s.cost_usd, s.category, s.tool_names FROM source.turns s; INSERT OR IGNORE INTO parse_offsets(file_path, byte_offset, mtime, file_id) SELECT file_path, byte_offset, mtime, file_id FROM source.parse_offsets; @@ -529,9 +754,12 @@ async fn merge_sessions_tx(conn: &Connection, offsets: &SessionMergeOffsets) -> INSERT OR IGNORE INTO session_messages( provider, message_id, session_id, role, timestamp, ordinal, text, kind, model, tool_names, source_path, source_offset, metadata_json - ) SELECT provider, message_id, session_id, role, timestamp, ordinal, text, kind, - model, tool_names, source_path, source_offset, metadata_json - FROM source.session_messages; + ) SELECT s.provider, COALESCE(m.mapped_id, s.message_id), s.session_id, s.role, + s.timestamp, s.ordinal, s.text, s.kind, s.model, s.tool_names, + s.source_path, s.source_offset, {session_metadata} + FROM source.session_messages s + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id; INSERT OR IGNORE INTO session_schema_migrations(name, version, applied_at) SELECT name, version, applied_at FROM source.session_schema_migrations; @@ -543,19 +771,38 @@ async fn merge_sessions_tx(conn: &Connection, offsets: &SessionMergeOffsets) -> provider, message_id, session_id, store_id, role, ordinal, timestamp, content, content_hash, storage_kind, payload_ref, snippet_text, index_text, legacy_source, legacy_truncated, metadata_json - ) SELECT provider, message_id, session_id, store_id + {raw}, role, ordinal, - timestamp, content, content_hash, storage_kind, payload_ref, snippet_text, - index_text, legacy_source, legacy_truncated, metadata_json - FROM source.lcm_raw_messages; + ) SELECT s.provider, + CASE WHEN m.raw_content_divergent=1 THEN m.mapped_id ELSE s.message_id END, + s.session_id, + s.store_id + {raw}, s.role, s.ordinal, s.timestamp, s.content, + s.content_hash, s.storage_kind, s.payload_ref, s.snippet_text, + s.index_text, s.legacy_source, s.legacy_truncated, {raw_metadata} + FROM source.lcm_raw_messages s + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id; INSERT INTO consolidation_raw_map(source_id, target_id) SELECT s.store_id, t.store_id FROM source.lcm_raw_messages s - JOIN lcm_raw_messages t ON t.provider=s.provider AND t.message_id=s.message_id; + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id + JOIN lcm_raw_messages t + ON t.provider=s.provider + AND t.message_id=CASE WHEN m.raw_content_divergent=1 + THEN m.mapped_id ELSE s.message_id END; INSERT OR IGNORE INTO lcm_external_payloads( payload_ref, provider, session_id, message_id, kind, content_hash, byte_count, char_count, created_at, metadata_json - ) SELECT payload_ref, provider, session_id, message_id, kind, content_hash, - byte_count, char_count, created_at, metadata_json FROM source.lcm_external_payloads; + ) SELECT s.payload_ref, s.provider, s.session_id, + CASE WHEN m.raw_content_divergent=1 THEN m.mapped_id ELSE s.message_id END, + s.kind, s.content_hash, + s.byte_count, s.char_count, s.created_at, s.metadata_json + FROM source.lcm_external_payloads s + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id + WHERE NOT EXISTS ( + SELECT 1 FROM target_input.lcm_external_payloads t + WHERE t.payload_ref=s.payload_ref + ); INSERT OR IGNORE INTO lcm_gc_marks(payload_ref, state, first_seen_at, updated_at) SELECT payload_ref, state, first_seen_at, updated_at FROM source.lcm_gc_marks; INSERT OR IGNORE INTO lcm_gc_meta(key, value) SELECT key, value FROM source.lcm_gc_meta; @@ -644,10 +891,16 @@ async fn merge_sessions_tx(conn: &Connection, offsets: &SessionMergeOffsets) -> commit_sha, provider, session_id, branch, worktree, committed_at, span_overlap_kind, span_id, relation, evidence, confidence, evidence_message_id, created_at - ) SELECT commit_sha, provider, session_id, branch, worktree, committed_at, - span_overlap_kind, CASE WHEN span_id IS NULL THEN NULL ELSE span_id + {span} END, - relation, evidence, confidence, evidence_message_id, created_at - FROM source.commit_sessions; + ) SELECT cs.commit_sha, cs.provider, cs.session_id, cs.branch, cs.worktree, + cs.committed_at, cs.span_overlap_kind, + CASE WHEN cs.span_id IS NULL THEN NULL ELSE cs.span_id + {span} END, + cs.relation, cs.evidence, cs.confidence, + COALESCE((SELECT mapped_id FROM consolidation_message_map m + WHERE m.provider=cs.provider + AND m.original_id=cs.evidence_message_id), + cs.evidence_message_id), + cs.created_at + FROM source.commit_sessions cs; UPDATE commit_sessions AS t SET branch = (SELECT s.branch FROM source.commit_sessions s WHERE s.commit_sha=t.commit_sha AND s.provider=t.provider AND s.session_id=t.session_id), worktree = (SELECT s.worktree FROM source.commit_sessions s WHERE s.commit_sha=t.commit_sha AND s.provider=t.provider AND s.session_id=t.session_id), @@ -657,7 +910,12 @@ async fn merge_sessions_tx(conn: &Connection, offsets: &SessionMergeOffsets) -> relation = (SELECT s.relation FROM source.commit_sessions s WHERE s.commit_sha=t.commit_sha AND s.provider=t.provider AND s.session_id=t.session_id), evidence = (SELECT s.evidence FROM source.commit_sessions s WHERE s.commit_sha=t.commit_sha AND s.provider=t.provider AND s.session_id=t.session_id), confidence = (SELECT s.confidence FROM source.commit_sessions s WHERE s.commit_sha=t.commit_sha AND s.provider=t.provider AND s.session_id=t.session_id), - evidence_message_id = (SELECT s.evidence_message_id FROM source.commit_sessions s WHERE s.commit_sha=t.commit_sha AND s.provider=t.provider AND s.session_id=t.session_id) + evidence_message_id = (SELECT COALESCE( + (SELECT mapped_id FROM consolidation_message_map m + WHERE m.provider=s.provider AND m.original_id=s.evidence_message_id), + s.evidence_message_id) + FROM source.commit_sessions s + WHERE s.commit_sha=t.commit_sha AND s.provider=t.provider AND s.session_id=t.session_id) WHERE EXISTS ( SELECT 1 FROM source.commit_sessions s WHERE s.commit_sha=t.commit_sha AND s.provider=t.provider AND s.session_id=t.session_id @@ -677,8 +935,11 @@ async fn merge_sessions_tx(conn: &Connection, offsets: &SessionMergeOffsets) -> INSERT OR IGNORE INTO dashboard_token_counts( store, provider, message_id, text_len, encoder, token_count, computed_at - ) SELECT store, provider, message_id, text_len, encoder, token_count, computed_at - FROM source.dashboard_token_counts;", + ) SELECT s.store, s.provider, COALESCE(m.mapped_id, s.message_id), + s.text_len, s.encoder, s.token_count, s.computed_at + FROM source.dashboard_token_counts s + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id;", raw = offsets.raw, span = offsets.span, savings = offsets.savings, diff --git a/src/migrate/consolidate/sqlite/inspect.rs b/src/migrate/consolidate/sqlite/inspect.rs index 4f1243bb..6eeb9c3b 100644 --- a/src/migrate/consolidate/sqlite/inspect.rs +++ b/src/migrate/consolidate/sqlite/inspect.rs @@ -3,7 +3,10 @@ use std::path::{Path, PathBuf}; use libsql::{Builder, Connection, Database as LibsqlDatabase}; -use super::{attach, db_error, db_message, query_i64, quote_identifier, table_exists}; +use super::{ + LCM_RAW_MESSAGE_DIVERGENCE_PREDICATE, attach, db_error, db_message, query_i64, + quote_identifier, table_exists, +}; use crate::errors::Result; #[derive(Debug, Clone, Copy)] @@ -11,6 +14,21 @@ pub(in crate::migrate::consolidate) struct DatabaseCollisionCounts { pub sessions: u64, pub messages: u64, pub lcm_messages: u64, + pub divergent_lcm_messages: u64, + pub divergent_lcm_session_ids: u64, + pub divergent_lcm_content_hashes: u64, + pub divergent_lcm_storage_kinds: u64, + pub divergent_lcm_payload_refs: u64, +} + +#[derive(Debug, Default)] +struct LcmMessageCollisionCounts { + overlaps: u64, + divergent: u64, + session_ids: u64, + content_hashes: u64, + storage_kinds: u64, + payload_refs: u64, } pub(in crate::migrate::consolidate) struct OfflineDatabaseGuards { @@ -278,18 +296,16 @@ pub(in crate::migrate::consolidate) async fn inspect_collisions( "s.provider = t.provider AND s.message_id = t.message_id", ) .await?; - let lcm_messages = overlap_count( - snapshots, - source_sessions, - target_sessions, - "lcm_raw_messages", - "s.provider = t.provider AND s.message_id = t.message_id", - ) - .await?; + let lcm = lcm_message_collision_counts(snapshots, source_sessions, target_sessions).await?; Ok(DatabaseCollisionCounts { sessions, messages, - lcm_messages, + lcm_messages: lcm.overlaps, + divergent_lcm_messages: lcm.divergent, + divergent_lcm_session_ids: lcm.session_ids, + divergent_lcm_content_hashes: lcm.content_hashes, + divergent_lcm_storage_kinds: lcm.storage_kinds, + divergent_lcm_payload_refs: lcm.payload_refs, }) } @@ -327,6 +343,64 @@ async fn overlap_count( u64::try_from(count).map_err(|error| db_error("inspect_collisions", error)) } +async fn lcm_message_collision_counts( + snapshots: &crate::sqlite_read_snapshot::SnapshotSet, + source: &Path, + target: &Path, +) -> Result { + if !source.is_file() || !target.is_file() { + return Ok(LcmMessageCollisionCounts::default()); + } + let source = read_snapshot(snapshots, source)?; + let target = read_snapshot(snapshots, target)?; + let conn = source.connection(); + attach(conn, target.path()).await?; + let counts = if table_exists(conn, "main", "lcm_raw_messages").await? + && table_exists(conn, "other", "lcm_raw_messages").await? + { + let sql = format!( + "SELECT COUNT(*), + COALESCE(SUM(CASE WHEN {LCM_RAW_MESSAGE_DIVERGENCE_PREDICATE} THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN t.session_id IS NOT s.session_id THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN t.content_hash IS NOT s.content_hash THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN t.storage_kind IS NOT s.storage_kind THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN t.payload_ref IS NOT s.payload_ref THEN 1 ELSE 0 END), 0) + FROM main.lcm_raw_messages s + JOIN other.lcm_raw_messages t + ON s.provider=t.provider AND s.message_id=t.message_id" + ); + let mut rows = conn + .query(&sql, ()) + .await + .map_err(|error| db_error("inspect_collisions", error))?; + let row = rows + .next() + .await + .map_err(|error| db_error("inspect_collisions", error))? + .ok_or_else(|| db_message("inspect_collisions", "collision query returned no row"))?; + let count = |index| -> Result { + let value = row + .get::(index) + .map_err(|error| db_error("inspect_collisions", error))?; + u64::try_from(value).map_err(|error| db_error("inspect_collisions", error)) + }; + LcmMessageCollisionCounts { + overlaps: count(0)?, + divergent: count(1)?, + session_ids: count(2)?, + content_hashes: count(3)?, + storage_kinds: count(4)?, + payload_refs: count(5)?, + } + } else { + LcmMessageCollisionCounts::default() + }; + conn.execute("DETACH DATABASE other", ()) + .await + .map_err(|error| db_error("inspect_collisions", error))?; + Ok(counts) +} + fn read_snapshot<'a>( snapshots: &'a crate::sqlite_read_snapshot::SnapshotSet, path: &Path, diff --git a/src/migrate/consolidate/sqlite/verify.rs b/src/migrate/consolidate/sqlite/verify.rs index 33dbb941..726ca613 100644 --- a/src/migrate/consolidate/sqlite/verify.rs +++ b/src/migrate/consolidate/sqlite/verify.rs @@ -2,7 +2,10 @@ use std::path::Path; use libsql::Connection; -use super::{SessionMergeOffsets, attach_as, db_error, db_message, query_i64}; +use super::{ + SessionMergeOffsets, attach_as, build_consolidation_message_map, db_error, db_message, + mapped_parent_metadata, mapped_turn_message_id, query_i64, +}; use crate::errors::Result; struct TableVerification { @@ -17,11 +20,12 @@ pub(in crate::migrate::consolidate) async fn verify_session_union_sql( source: &Path, target: &Path, destination_snapshots: &crate::sqlite_read_snapshot::SnapshotSet, - destination: &Path, destination_root: &Path, offsets: &SessionMergeOffsets, + source_project_id: &str, ) -> Result<()> { - let conn = destination_snapshots.get(destination).map_err(|error| { + let destination = destination_root.join(crate::storage::SESSIONS_DB_FILENAME); + let conn = destination_snapshots.get(&destination).map_err(|error| { db_error( "verify_consolidation", format!("could not read destination snapshot: {error}"), @@ -49,6 +53,11 @@ pub(in crate::migrate::consolidate) async fn verify_session_union_sql( "target_input", ) .await?; + build_consolidation_message_map(conn, "source_input", "target_input", source_project_id) + .await?; + conn.execute_batch("PRAGMA query_only = ON;") + .await + .map_err(|error| db_error("verify_consolidation", error))?; let result = match verify_attached_tables(conn, offsets).await { Ok(()) => verify_payload_files(conn, destination_root).await, @@ -126,6 +135,9 @@ async fn verify_table(conn: &Connection, spec: &TableVerification) -> Result<()> } fn verification_specs(offsets: &SessionMergeOffsets) -> Vec { + let turn_message_id = mapped_turn_message_id("s", "source_input"); + let session_metadata = mapped_parent_metadata("s", false); + let raw_metadata = mapped_parent_metadata("s", true); let mut specs = vec![ custom( "project accounting", @@ -136,11 +148,26 @@ fn verification_specs(offsets: &SessionMergeOffsets) -> Vec { UNION ALL SELECT path, tokens_saved FROM source_input.projects ) GROUP BY path", ), - target_wins( + custom_owned( "turn", "turns", "message_id, project_hash, session_id, model, timestamp, input_tokens, output_tokens, cache_write_tokens, cache_read_tokens, cost_usd, category, tool_names", - "t.message_id=s.message_id", + format!( + "SELECT message_id, project_hash, session_id, model, timestamp, + input_tokens, output_tokens, cache_write_tokens, + cache_read_tokens, cost_usd, category, tool_names + FROM target_input.turns + UNION ALL + SELECT {turn_message_id}, s.project_hash, s.session_id, s.model, + s.timestamp, s.input_tokens, s.output_tokens, + s.cache_write_tokens, s.cache_read_tokens, s.cost_usd, + s.category, s.tool_names + FROM source_input.turns s + WHERE {turn_message_id} != s.message_id OR NOT EXISTS ( + SELECT 1 FROM target_input.turns t + WHERE t.message_id=s.message_id + )" + ), ), custom( "parse offset", @@ -212,11 +239,24 @@ fn verification_specs(offsets: &SessionMergeOffsets) -> Vec { FROM source_input.sessions s WHERE NOT EXISTS (SELECT 1 FROM target_input.sessions t WHERE t.provider=s.provider AND t.session_id=s.session_id)", ), - target_wins( + custom_owned( "session message", "session_messages", "provider, message_id, session_id, role, timestamp, ordinal, text, kind, model, tool_names, source_path, source_offset, metadata_json", - "t.provider=s.provider AND t.message_id=s.message_id", + format!("SELECT provider, message_id, session_id, role, timestamp, ordinal, text, + kind, model, tool_names, source_path, source_offset, metadata_json + FROM target_input.session_messages + UNION ALL + SELECT s.provider, COALESCE(m.mapped_id, s.message_id), s.session_id, + s.role, s.timestamp, s.ordinal, s.text, s.kind, s.model, + s.tool_names, s.source_path, s.source_offset, {session_metadata} + FROM source_input.session_messages s + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id + WHERE m.mapped_id IS NOT NULL OR NOT EXISTS ( + SELECT 1 FROM target_input.session_messages t + WHERE t.provider=s.provider AND t.message_id=s.message_id + )"), ), custom( "session schema migration", @@ -227,21 +267,53 @@ fn verification_specs(offsets: &SessionMergeOffsets) -> Vec { UNION ALL SELECT name, version, applied_at FROM source_input.session_schema_migrations ) GROUP BY name", ), - projected_target_wins( + custom_owned( "LCM raw message", "lcm_raw_messages", "provider, message_id, session_id, store_id, role, ordinal, timestamp, content, content_hash, storage_kind, payload_ref, snippet_text, index_text, legacy_source, legacy_truncated, metadata_json", - &format!( - "provider, message_id, session_id, store_id + {}, role, ordinal, timestamp, content, content_hash, storage_kind, payload_ref, snippet_text, index_text, legacy_source, legacy_truncated, metadata_json", + format!( + "SELECT provider, message_id, session_id, store_id, role, ordinal, + timestamp, content, content_hash, storage_kind, payload_ref, + snippet_text, index_text, legacy_source, legacy_truncated, metadata_json + FROM target_input.lcm_raw_messages + UNION ALL + SELECT s.provider, + CASE WHEN m.raw_content_divergent=1 + THEN m.mapped_id ELSE s.message_id END, + s.session_id, + s.store_id + {}, s.role, s.ordinal, s.timestamp, s.content, + s.content_hash, s.storage_kind, s.payload_ref, s.snippet_text, + s.index_text, s.legacy_source, s.legacy_truncated, {raw_metadata} + FROM source_input.lcm_raw_messages s + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id + WHERE m.raw_content_divergent=1 OR NOT EXISTS ( + SELECT 1 FROM target_input.lcm_raw_messages t + WHERE t.provider=s.provider AND t.message_id=s.message_id + )", offsets.raw ), - "t.provider=s.provider AND t.message_id=s.message_id", ), - target_wins( + custom( "LCM external payload", "lcm_external_payloads", "payload_ref, provider, session_id, message_id, kind, content_hash, byte_count, char_count, created_at, metadata_json", - "t.payload_ref=s.payload_ref", + "SELECT payload_ref, provider, session_id, message_id, kind, content_hash, + byte_count, char_count, created_at, metadata_json + FROM target_input.lcm_external_payloads + UNION ALL + SELECT s.payload_ref, s.provider, s.session_id, + CASE WHEN m.raw_content_divergent=1 + THEN m.mapped_id ELSE s.message_id END, + s.kind, s.content_hash, + s.byte_count, s.char_count, s.created_at, s.metadata_json + FROM source_input.lcm_external_payloads s + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id + WHERE NOT EXISTS ( + SELECT 1 FROM target_input.lcm_external_payloads t + WHERE t.payload_ref=s.payload_ref + )", ), target_wins( "LCM GC mark", @@ -350,11 +422,23 @@ fn verification_specs(offsets: &SessionMergeOffsets) -> Vec { commit_sessions(offsets.span), max_meta("git correlation metadata", "git_correlation_meta"), max_meta("session backfill metadata", "session_backfill_meta"), - target_wins( + custom( "dashboard token count", "dashboard_token_counts", "store, provider, message_id, text_len, encoder, token_count, computed_at", - "t.store=s.store AND t.provider=s.provider AND t.message_id=s.message_id", + "SELECT store, provider, message_id, text_len, encoder, token_count, computed_at + FROM target_input.dashboard_token_counts + UNION ALL + SELECT s.store, s.provider, COALESCE(m.mapped_id, s.message_id), + s.text_len, s.encoder, s.token_count, s.computed_at + FROM source_input.dashboard_token_counts s + LEFT JOIN consolidation_message_map m + ON m.provider=s.provider AND m.original_id=s.message_id + WHERE m.mapped_id IS NOT NULL OR NOT EXISTS ( + SELECT 1 FROM target_input.dashboard_token_counts t + WHERE t.store=s.store AND t.provider=s.provider + AND t.message_id=s.message_id + )", ), ]); specs @@ -465,7 +549,12 @@ fn commit_sessions(span_offset: i64) -> TableVerification { CASE WHEN s.confidence > t.confidence THEN s.relation ELSE t.relation END, CASE WHEN s.confidence > t.confidence THEN s.evidence ELSE t.evidence END, MAX(t.confidence, s.confidence), - CASE WHEN s.confidence > t.confidence THEN s.evidence_message_id ELSE t.evidence_message_id END, + CASE WHEN s.confidence > t.confidence THEN COALESCE( + (SELECT mapped_id FROM consolidation_message_map m + WHERE m.provider=s.provider + AND m.original_id=s.evidence_message_id), + s.evidence_message_id) + ELSE t.evidence_message_id END, t.created_at FROM target_input.commit_sessions t JOIN source_input.commit_sessions s @@ -483,7 +572,11 @@ fn commit_sessions(span_offset: i64) -> TableVerification { SELECT s.commit_sha, s.provider, s.session_id, s.branch, s.worktree, s.committed_at, s.span_overlap_kind, CASE WHEN s.span_id IS NULL THEN NULL ELSE s.span_id + {span_offset} END, - s.relation, s.evidence, s.confidence, s.evidence_message_id, s.created_at + s.relation, s.evidence, s.confidence, COALESCE( + (SELECT mapped_id FROM consolidation_message_map m + WHERE m.provider=s.provider + AND m.original_id=s.evidence_message_id), + s.evidence_message_id), s.created_at FROM source_input.commit_sessions s WHERE NOT EXISTS ( SELECT 1 FROM target_input.commit_sessions t @@ -497,8 +590,12 @@ fn remapped_raw_id(expression: &str, offset: i64) -> String { format!( "(SELECT COALESCE(t.store_id, r.store_id + {offset}) FROM source_input.lcm_raw_messages r + LEFT JOIN consolidation_message_map m + ON m.provider=r.provider AND m.original_id=r.message_id LEFT JOIN target_input.lcm_raw_messages t - ON t.provider=r.provider AND t.message_id=r.message_id + ON t.provider=r.provider + AND t.message_id=CASE WHEN m.raw_content_divergent=1 + THEN m.mapped_id ELSE r.message_id END WHERE r.store_id={expression})" ) } diff --git a/src/migrate/consolidate/tests.rs b/src/migrate/consolidate/tests.rs index d6a2f98d..5f645833 100644 --- a/src/migrate/consolidate/tests.rs +++ b/src/migrate/consolidate/tests.rs @@ -494,6 +494,66 @@ async fn consolidation_restarts_after_every_durable_state() { } } +#[tokio::test] +async fn version_one_premerge_ledger_migrates_before_resume() { + let fixture = fixture().await; + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let error = apply_with_stop( + &options, + &report.confirmation_token, + Some(ConsolidationState::DestinationReady), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("synthetic interruption")); + + let mut ledger = load_ledger(&report.ledger_path).unwrap().unwrap(); + ledger.schema_version = 1; + save_ledger(&report.ledger_path, &ledger).unwrap(); + + let applied = apply(&options, &report.confirmation_token).await.unwrap(); + assert_eq!(applied.state, ConsolidationState::Applied); + assert_eq!( + load_ledger(&report.ledger_path) + .unwrap() + .unwrap() + .schema_version, + LEDGER_SCHEMA_VERSION + ); +} + +#[tokio::test] +async fn version_one_postmerge_ledger_fails_closed() { + let fixture = fixture().await; + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let error = apply_with_stop( + &options, + &report.confirmation_token, + Some(ConsolidationState::DatabasesMerged), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("synthetic interruption")); + + let mut ledger = load_ledger(&report.ledger_path).unwrap().unwrap(); + ledger.schema_version = 1; + save_ledger(&report.ledger_path, &ledger).unwrap(); + + let error = apply(&options, &report.confirmation_token) + .await + .unwrap_err(); + assert!(error.to_string().contains("cannot be migrated safely")); + assert_eq!( + storage::read_repository_identity_marker(&fixture.project) + .unwrap() + .unwrap() + .project_id, + fixture.target_id + ); +} + #[tokio::test] async fn verification_rejects_a_missing_unique_row_when_target_is_larger() { let fixture = fixture().await; @@ -625,6 +685,738 @@ async fn verification_checks_session_bounds_and_immutable_message_payloads() { ); } +#[tokio::test] +async fn divergent_session_message_projections_preserve_a_source_variant() { + let fixture = fixture().await; + let source_sessions = fixture + .profile + .join("projects") + .join(&fixture.source_id) + .join(storage::SESSIONS_DB_FILENAME); + execute_sql( + &source_sessions, + "INSERT INTO session_messages( + provider, message_id, session_id, role, timestamp, ordinal, text, kind, model + ) VALUES + ('codex', 'message-current-session', 'legacy-session', 'assistant', + 1800000000, 1, 'source divergent projection', 'message', 'source-model'), + ('codex', 'source-only-message', 'legacy-session', 'user', + 1800000001, 2, 'source-only projection', 'message', NULL);", + ) + .await; + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + assert_eq!(report.collisions.message_overlaps, 1); + assert_eq!(report.collisions.divergent_lcm_messages, 0); + + let applied = apply(&options, &report.confirmation_token).await.unwrap(); + let sessions = GlobalDb::open_read_only_at( + &applied + .destination_data_root + .join(storage::SESSIONS_DB_FILENAME), + ) + .await + .unwrap(); + let selected = sessions + .get_session_message("codex", "message-current-session") + .await + .unwrap(); + assert_eq!(selected.session_id, "current-session"); + assert_eq!(selected.text, "message from current-session"); + assert_eq!(selected.role, "user"); + let variant_id = format!("consolidated/{}/message-current-session", fixture.source_id); + let variant = sessions + .get_session_message("codex", &variant_id) + .await + .unwrap(); + assert_eq!(variant.session_id, "legacy-session"); + assert_eq!(variant.text, "source divergent projection"); + assert_eq!(variant.role, "assistant"); + let source_only = sessions + .get_session_message("codex", "source-only-message") + .await + .unwrap(); + assert_eq!(source_only.session_id, "legacy-session"); + assert_eq!(source_only.text, "source-only projection"); + sessions.close(); +} + +#[tokio::test] +async fn lcm_representation_drift_uses_the_selected_target_row() { + let fixture = fixture().await; + let source_sessions = fixture + .profile + .join("projects") + .join(&fixture.source_id) + .join(storage::SESSIONS_DB_FILENAME); + let target_sessions = fixture + .profile + .join("projects") + .join(&fixture.target_id) + .join(storage::SESSIONS_DB_FILENAME); + let target = GlobalDb::open_read_only_at(&target_sessions).await.unwrap(); + let target_raw = target + .lcm_load_raw_message("codex", "message-current-session") + .await + .unwrap(); + target.close(); + let source = GlobalDb::open_at(&source_sessions).await.unwrap(); + source + .conn() + .execute( + "UPDATE lcm_raw_messages + SET message_id=?1, content=?2, content_hash=?3, + storage_kind='external', payload_ref=NULL + WHERE provider='codex' AND message_id='message-legacy-session'", + libsql::params![ + target_raw.message_id.clone(), + target_raw.content.clone(), + target_raw.content_hash.clone() + ], + ) + .await + .unwrap(); + source.checkpoint().await; + source.close(); + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + assert_eq!(report.collisions.lcm_message_overlaps, 1); + assert_eq!(report.collisions.divergent_lcm_messages, 1); + assert_eq!(report.collisions.divergent_lcm_session_ids, 1); + assert_eq!(report.collisions.divergent_lcm_content_hashes, 0); + assert_eq!(report.collisions.divergent_lcm_storage_kinds, 1); + assert_eq!(report.collisions.divergent_lcm_payload_refs, 0); + + let applied = apply(&options, &report.confirmation_token).await.unwrap(); + let destination = GlobalDb::open_read_only_at( + &applied + .destination_data_root + .join(storage::SESSIONS_DB_FILENAME), + ) + .await + .unwrap(); + assert_eq!( + destination + .lcm_load_raw_message("codex", "message-current-session") + .await + .unwrap(), + target_raw + ); + destination.close(); +} + +#[tokio::test] +async fn session_only_divergence_does_not_duplicate_identical_external_raw_family() { + let fixture = fixture().await; + let source = layout_for_id(&fixture.project, &fixture.profile, &fixture.source_id).unwrap(); + let target = layout_for_id(&fixture.project, &fixture.profile, &fixture.target_id).unwrap(); + let payload = b"shared payload"; + let payload_ref = "shared.payload"; + let content_hash = crate::sessions::lcm::raw::sha256_hex("shared payload"); + for layout in [&source, &target] { + fs::create_dir_all(layout.data_root.join("lcm-payloads")).unwrap(); + fs::write( + layout.data_root.join("lcm-payloads").join(payload_ref), + payload, + ) + .unwrap(); + } + execute_sql( + &source.sessions_db_path, + &format!( + "UPDATE session_messages + SET message_id='message-current-session', text='source projection' + WHERE message_id='message-legacy-session'; + UPDATE lcm_raw_messages + SET message_id='message-current-session', content=NULL, + content_hash='{content_hash}', storage_kind='external', + payload_ref='{payload_ref}' + WHERE message_id='message-legacy-session'; + INSERT INTO lcm_external_payloads( + payload_ref, provider, session_id, message_id, kind, content_hash, + byte_count, char_count + ) VALUES('{payload_ref}', 'codex', 'legacy-session', + 'message-current-session', 'message', '{content_hash}', 14, 14);" + ), + ) + .await; + execute_sql( + &target.sessions_db_path, + &format!( + "UPDATE lcm_raw_messages + SET content=NULL, content_hash='{content_hash}', storage_kind='external', + payload_ref='{payload_ref}' + WHERE message_id='message-current-session'; + INSERT INTO lcm_external_payloads( + payload_ref, provider, session_id, message_id, kind, content_hash, + byte_count, char_count + ) VALUES('{payload_ref}', 'codex', 'current-session', + 'message-current-session', 'message', '{content_hash}', 14, 14);" + ), + ) + .await; + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let applied = apply(&options, &report.confirmation_token).await.unwrap(); + let variant_id = format!("consolidated/{}/message-current-session", fixture.source_id); + let sessions = GlobalDb::open_read_only_at( + &applied + .destination_data_root + .join(storage::SESSIONS_DB_FILENAME), + ) + .await + .unwrap(); + assert!( + sessions + .get_session_message("codex", &variant_id) + .await + .is_some() + ); + assert!( + sessions + .lcm_load_raw_message("codex", &variant_id) + .await + .is_none() + ); + let raw = sessions + .lcm_load_raw_message("codex", "message-current-session") + .await + .unwrap(); + assert_eq!(raw.content_hash, content_hash); + assert_eq!(raw.payload_ref.as_deref(), Some(payload_ref)); + let mut rows = sessions + .conn() + .query( + "SELECT message_id FROM lcm_external_payloads WHERE payload_ref=?1", + [payload_ref], + ) + .await + .unwrap(); + assert_eq!( + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), + "message-current-session" + ); + sessions.close(); +} + +#[tokio::test] +async fn distinct_external_content_variant_preserves_owner_expansion_and_retry() { + let fixture = fixture().await; + let source = layout_for_id(&fixture.project, &fixture.profile, &fixture.source_id).unwrap(); + let target = layout_for_id(&fixture.project, &fixture.profile, &fixture.target_id).unwrap(); + let mut source_ref = None; + for (layout, session_id, old_message_id, content) in [ + ( + &source, + "legacy-session", + "message-legacy-session", + "source external body", + ), + ( + &target, + "current-session", + "message-current-session", + "target external body", + ), + ] { + let payload = crate::sessions::lcm::payload::write_external_payload( + &layout.data_root, + "codex", + session_id, + "message-current-session", + "message", + content, + None, + ) + .unwrap(); + if session_id == "legacy-session" { + source_ref = Some(payload.payload_ref.clone()); + } + let db = GlobalDb::open_at(&layout.sessions_db_path).await.unwrap(); + crate::sessions::lcm::payload::upsert_payload_metadata(db.conn(), &payload) + .await + .unwrap(); + db.conn() + .execute( + "UPDATE session_messages + SET message_id='message-current-session', text=?1 + WHERE provider='codex' AND message_id=?2", + libsql::params![content, old_message_id], + ) + .await + .unwrap(); + db.conn() + .execute( + "UPDATE lcm_raw_messages + SET message_id='message-current-session', content=NULL, + content_hash=?1, storage_kind='external', payload_ref=?2 + WHERE provider='codex' AND message_id=?3", + libsql::params![payload.content_hash, payload.payload_ref, old_message_id], + ) + .await + .unwrap(); + db.checkpoint().await; + db.close(); + } + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let applied = apply(&options, &report.confirmation_token).await.unwrap(); + let variant_id = format!("consolidated/{}/message-current-session", fixture.source_id); + let source_ref = source_ref.unwrap(); + let sessions = GlobalDb::open_read_only_at( + &applied + .destination_data_root + .join(storage::SESSIONS_DB_FILENAME), + ) + .await + .unwrap(); + let raw = sessions + .lcm_load_raw_message("codex", &variant_id) + .await + .unwrap(); + assert_eq!(raw.payload_ref.as_deref(), Some(source_ref.as_str())); + let mut owners = sessions + .conn() + .query( + "SELECT message_id FROM lcm_external_payloads WHERE payload_ref=?1", + [source_ref.as_str()], + ) + .await + .unwrap(); + assert_eq!( + owners + .next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), + variant_id + ); + drop(owners); + let expanded = crate::sessions::lcm::payload::LcmStore::new( + sessions.conn(), + applied.destination_data_root.clone(), + ) + .lcm_expand_payload("codex", "legacy-session", &source_ref, 0, 100) + .await + .unwrap(); + assert_eq!(expanded.content, "source external body"); + sessions.close(); + + let retried = apply(&options, &report.confirmation_token).await.unwrap(); + assert_eq!(retried.state, ConsolidationState::Applied); +} + +#[tokio::test] +async fn divergent_projection_and_raw_content_preserve_a_linked_source_variant() { + let fixture = fixture().await; + let source_sessions = fixture + .profile + .join("projects") + .join(&fixture.source_id) + .join(storage::SESSIONS_DB_FILENAME); + let target_sessions = fixture + .profile + .join("projects") + .join(&fixture.target_id) + .join(storage::SESSIONS_DB_FILENAME); + execute_sql( + &source_sessions, + "UPDATE session_messages + SET message_id='message-current-session', text='source divergent projection', + metadata_json='{\"parent_message_id\":\"message-current-session\"}' + WHERE provider='codex' AND message_id='message-legacy-session'; + UPDATE lcm_raw_messages + SET message_id='message-current-session', + metadata_json='{\"parent_message_id\":\"message-current-session\"}' + WHERE provider='codex' AND message_id='message-legacy-session'; + INSERT INTO turns( + message_id, project_hash, session_id, model, timestamp, input_tokens, + output_tokens, cost_usd, category + ) VALUES( + 'message-current-session', 'project', 'legacy-session', 'model', + 1800000000, 1, 1, 0.0, 'task' + ); + INSERT INTO session_messages( + provider, message_id, session_id, role, timestamp, ordinal, text, + kind, metadata_json + ) VALUES( + 'codex', 'message-current-session:thinking', 'legacy-session', + 'assistant', 1800000001, 2, 'source thinking', 'reasoning', + '{\"parent_message_id\":\"message-current-session\"}' + ); + INSERT INTO lcm_raw_messages( + provider, message_id, session_id, role, ordinal, timestamp, content, + content_hash, storage_kind, snippet_text, index_text, metadata_json + ) VALUES( + 'codex', 'message-current-session:thinking', 'legacy-session', + 'assistant', 2, 1800000001, 'source thinking', 'thinking-hash', + 'inline', 'source thinking', 'source thinking', + '{\"parent_message_id\":\"message-current-session\"}' + ); + INSERT INTO lcm_summary_nodes( + node_id, provider, conversation_id, session_id, depth, summary_text, + summary_hash, summary_token_count, source_token_count, created_at + ) VALUES( + 'variant-summary', 'codex', 'source-conversation', 'legacy-session', 1, + 'variant summary', 'variant-summary-hash', 1, 1, 1800000002 + ); + INSERT INTO lcm_summary_sources(node_id, source_kind, source_id, ordinal) + SELECT 'variant-summary', 'raw_message', CAST(store_id AS TEXT), 0 + FROM lcm_raw_messages WHERE message_id='message-current-session';", + ) + .await; + execute_sql( + &target_sessions, + "INSERT INTO session_messages( + provider, message_id, session_id, role, timestamp, ordinal, text, + kind, metadata_json + ) VALUES( + 'codex', 'message-current-session:thinking', 'current-session', + 'assistant', 1800000001, 2, 'source thinking', 'reasoning', + '{\"parent_message_id\":\"message-current-session\"}' + ); + INSERT INTO lcm_raw_messages( + provider, message_id, session_id, role, ordinal, timestamp, content, + content_hash, storage_kind, snippet_text, index_text, metadata_json + ) VALUES( + 'codex', 'message-current-session:thinking', 'current-session', + 'assistant', 2, 1800000001, 'source thinking', 'thinking-hash', + 'inline', 'source thinking', 'source thinking', + '{\"parent_message_id\":\"message-current-session\"}' + );", + ) + .await; + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + assert_eq!(report.collisions.divergent_lcm_messages, 2); + assert_eq!(report.collisions.divergent_lcm_session_ids, 2); + assert_eq!(report.collisions.divergent_lcm_content_hashes, 1); + assert_eq!(report.collisions.divergent_lcm_storage_kinds, 0); + assert_eq!(report.collisions.divergent_lcm_payload_refs, 0); + + let applied = apply(&options, &report.confirmation_token).await.unwrap(); + let variant_id = format!("consolidated/{}/message-current-session", fixture.source_id); + let sessions = GlobalDb::open_read_only_at( + &applied + .destination_data_root + .join(storage::SESSIONS_DB_FILENAME), + ) + .await + .unwrap(); + assert_eq!( + sessions + .get_session_message("codex", "message-current-session") + .await + .unwrap() + .text, + "message from current-session" + ); + let variant = sessions + .get_session_message("codex", &variant_id) + .await + .unwrap(); + assert_eq!(variant.text, "source divergent projection"); + let metadata: serde_json::Value = + serde_json::from_str(variant.metadata_json.as_deref().unwrap()).unwrap(); + assert_eq!(metadata["parent_message_id"], variant_id); + assert_eq!( + metadata["consolidation_original_parent_message_id"], + "message-current-session" + ); + let raw_variant = sessions + .lcm_load_raw_message("codex", &variant_id) + .await + .unwrap(); + assert_eq!(raw_variant.content, "message from legacy-session"); + let raw_metadata: serde_json::Value = + serde_json::from_str(raw_variant.metadata_json.as_deref().unwrap()).unwrap(); + assert_eq!(raw_metadata["parent_message_id"], variant_id); + assert_eq!( + raw_metadata["consolidation_original_parent_message_id"], + "message-current-session" + ); + let thinking_variant_id = format!("{variant_id}:thinking"); + let thinking = sessions + .get_session_message("codex", &thinking_variant_id) + .await + .unwrap(); + let thinking_metadata: serde_json::Value = + serde_json::from_str(thinking.metadata_json.as_deref().unwrap()).unwrap(); + assert_eq!(thinking_metadata["parent_message_id"], variant_id); + assert_eq!( + thinking_metadata["consolidation_original_parent_message_id"], + "message-current-session" + ); + assert!( + sessions + .lcm_load_raw_message("codex", &thinking_variant_id) + .await + .is_none() + ); + let thinking_raw = sessions + .lcm_load_raw_message("codex", "message-current-session:thinking") + .await + .unwrap(); + let thinking_raw_metadata: serde_json::Value = + serde_json::from_str(thinking_raw.metadata_json.as_deref().unwrap()).unwrap(); + assert_eq!( + thinking_raw_metadata["parent_message_id"], + "message-current-session" + ); + let mut turn_rows = sessions + .conn() + .query( + "SELECT message_id FROM turns WHERE session_id='legacy-session'", + (), + ) + .await + .unwrap(); + assert_eq!( + turn_rows + .next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), + variant_id + ); + drop(turn_rows); + let mut rows = sessions + .conn() + .query( + "SELECT r.message_id + FROM lcm_summary_sources s + JOIN lcm_raw_messages r ON r.store_id=CAST(s.source_id AS INTEGER) + WHERE s.node_id='variant-summary'", + (), + ) + .await + .unwrap(); + assert_eq!( + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), + variant_id + ); + assert!(rows.next().await.unwrap().is_none()); + drop(rows); + sessions.close(); + + let retried = apply(&options, &report.confirmation_token).await.unwrap(); + assert_eq!(retried.state, ConsolidationState::Applied); + let sessions = GlobalDb::open_read_only_at( + &retried + .destination_data_root + .join(storage::SESSIONS_DB_FILENAME), + ) + .await + .unwrap(); + assert_eq!(sessions.session_message_count().await.unwrap(), 4); + sessions.close(); +} + +#[tokio::test] +async fn synthetic_message_key_collision_fails_before_merge() { + let fixture = fixture().await; + let source_sessions = fixture + .profile + .join("projects") + .join(&fixture.source_id) + .join(storage::SESSIONS_DB_FILENAME); + let synthetic = format!("consolidated/{}/message-current-session", fixture.source_id); + let source = GlobalDb::open_at(&source_sessions).await.unwrap(); + source + .conn() + .execute( + "UPDATE session_messages + SET message_id='message-current-session', text='source divergent projection' + WHERE provider='codex' AND message_id='message-legacy-session'", + (), + ) + .await + .unwrap(); + source + .conn() + .execute( + "INSERT INTO session_messages( + provider, message_id, session_id, role, ordinal, text, kind + ) VALUES('codex', ?1, 'legacy-session', 'user', 2, 'native collision', 'message')", + [synthetic], + ) + .await + .unwrap(); + source.checkpoint().await; + source.close(); + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let error = apply(&options, &report.confirmation_token) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("synthetic consolidation message key collision"), + "{error}" + ); + assert_eq!( + storage::read_repository_identity_marker(&fixture.project) + .unwrap() + .unwrap() + .project_id, + fixture.target_id + ); +} + +#[tokio::test] +async fn ambiguous_cross_provider_turn_mapping_fails_closed() { + let fixture = fixture().await; + let source = layout_for_id(&fixture.project, &fixture.profile, &fixture.source_id).unwrap(); + let target = layout_for_id(&fixture.project, &fixture.profile, &fixture.target_id).unwrap(); + execute_sql( + &source.sessions_db_path, + "UPDATE session_messages + SET message_id='shared-message', text='source codex' + WHERE provider='codex' AND message_id='message-legacy-session'; + INSERT INTO sessions(provider, session_id, project_key, project_path) + VALUES('claude', 'legacy-session', 'project', '/repo'); + INSERT INTO session_messages( + provider, message_id, session_id, role, ordinal, text, kind + ) VALUES('claude', 'shared-message', 'legacy-session', 'assistant', 1, + 'source claude', 'message'); + INSERT INTO turns( + message_id, project_hash, session_id, model, timestamp, input_tokens, + output_tokens, cost_usd, category + ) VALUES('shared-message', 'project', 'legacy-session', 'model', + 1800000000, 1, 1, 0.0, 'task');", + ) + .await; + execute_sql( + &target.sessions_db_path, + "UPDATE session_messages SET message_id='shared-message' + WHERE provider='codex' AND message_id='message-current-session'; + INSERT INTO sessions(provider, session_id, project_key, project_path) + VALUES('claude', 'current-session', 'project', '/repo'); + INSERT INTO session_messages( + provider, message_id, session_id, role, ordinal, text, kind + ) VALUES('claude', 'shared-message', 'current-session', 'user', 1, + 'target claude', 'message');", + ) + .await; + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let error = apply(&options, &report.confirmation_token) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("source turn message mapping ambiguity collision"), + "{error}" + ); +} + +#[tokio::test] +async fn divergent_external_payload_identity_remains_a_hard_error() { + let fixture = fixture().await; + let source = layout_for_id(&fixture.project, &fixture.profile, &fixture.source_id).unwrap(); + let target = layout_for_id(&fixture.project, &fixture.profile, &fixture.target_id).unwrap(); + execute_sql( + &source.sessions_db_path, + "INSERT INTO lcm_external_payloads( + payload_ref, provider, session_id, message_id, kind, content_hash, + byte_count, char_count, metadata_json + ) VALUES('shared-ref', 'codex', 'legacy-session', 'message-legacy-session', + 'tool', 'source-hash', 10, 10, NULL);", + ) + .await; + execute_sql( + &target.sessions_db_path, + "INSERT INTO lcm_external_payloads( + payload_ref, provider, session_id, message_id, kind, content_hash, + byte_count, char_count, metadata_json + ) VALUES('shared-ref', 'codex', 'current-session', 'message-current-session', + 'tool', 'target-hash', 11, 11, NULL);", + ) + .await; + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let error = apply(&options, &report.confirmation_token) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("divergent LCM external payload collision"), + "{error}" + ); +} + +#[tokio::test] +async fn divergent_summary_node_identity_remains_a_hard_error() { + let fixture = fixture().await; + let source = layout_for_id(&fixture.project, &fixture.profile, &fixture.source_id).unwrap(); + let target = layout_for_id(&fixture.project, &fixture.profile, &fixture.target_id).unwrap(); + for (path, session, text, hash) in [ + ( + &source.sessions_db_path, + "legacy-session", + "source summary", + "source-hash", + ), + ( + &target.sessions_db_path, + "current-session", + "target summary", + "target-hash", + ), + ] { + let (db, _) = Database::open(path).await.unwrap(); + db.conn() + .execute( + "INSERT INTO lcm_summary_nodes( + node_id, provider, conversation_id, session_id, depth, summary_text, + summary_hash, summary_token_count, source_token_count, created_at + ) VALUES('shared-summary', 'codex', 'conversation', ?1, 1, ?2, ?3, 1, 1, 1800000002)", + libsql::params![session, text, hash], + ) + .await + .unwrap(); + db.checkpoint().await.unwrap(); + db.close(); + } + + let options = fixture.options(); + let report = plan(&options).await.unwrap(); + let error = apply(&options, &report.confirmation_token) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("divergent LCM summary node collision"), + "{error}" + ); +} + #[tokio::test] async fn identity_survives_symlink_and_repository_move() { let fixture = fixture().await;