diff --git a/src/integration/regression.rs b/src/integration/regression.rs index 4ad61d0..511329c 100644 --- a/src/integration/regression.rs +++ b/src/integration/regression.rs @@ -793,70 +793,6 @@ fn f32_slice_to_bytes_roundtrip() { assert_eq!(original, roundtrip); } -#[test] -fn sync_skips_unchanged_session() { - let store = setup(); - let session = Session { - id: "s1".to_string(), - source: "test".to_string(), - source_id: "raw1".to_string(), - title: "Original".to_string(), - directory: None, - repo_remote: None, - repo_slug: None, - repo_name: None, - started_at: 1000, - updated_at: Some(2000), - message_count: 2, - entrypoint: None, - custom_title: None, - summary: None, - duration_minutes: None, - source_file_path: None, - is_import: false, - }; - store.insert_session(&session).unwrap(); - - let meta = store.session_meta("test", "raw1").unwrap(); - assert_eq!(meta, Some((Some(2000), 2))); -} - -#[test] -fn sync_detects_new_messages() { - let store = setup(); - let session = Session { - id: "s1".to_string(), - source: "test".to_string(), - source_id: "raw1".to_string(), - title: "Original".to_string(), - directory: None, - repo_remote: None, - repo_slug: None, - repo_name: None, - started_at: 1000, - updated_at: Some(2000), - message_count: 2, - entrypoint: None, - custom_title: None, - summary: None, - duration_minutes: None, - source_file_path: None, - is_import: false, - }; - store.insert_session(&session).unwrap(); - store.insert_messages(&[make_message("s1", Role::User, "hello", 0)]).unwrap(); - - let meta = store.session_meta("test", "raw1").unwrap().unwrap(); - let (old_updated_at, old_msg_count) = meta; - - let new_msg_count: u32 = 5; - let new_updated_at: Option = Some(3000); - - let changed = old_msg_count != new_msg_count - || (new_updated_at.is_some() && new_updated_at != old_updated_at); - assert!(changed, "sync must detect message count change"); -} - #[test] fn replace_session_rolls_back_delete_when_reinsert_fails() { let store = setup(); @@ -1039,40 +975,6 @@ fn replace_session_clears_import_marker_on_success() { ); } -#[test] -fn sync_detects_updated_timestamp() { - let store = setup(); - let session = Session { - id: "s1".to_string(), - source: "test".to_string(), - source_id: "raw1".to_string(), - title: "Original".to_string(), - directory: None, - repo_remote: None, - repo_slug: None, - repo_name: None, - started_at: 1000, - updated_at: Some(2000), - message_count: 3, - entrypoint: None, - custom_title: None, - summary: None, - duration_minutes: None, - source_file_path: None, - is_import: false, - }; - store.insert_session(&session).unwrap(); - - let (old_updated_at, old_msg_count) = store.session_meta("test", "raw1").unwrap().unwrap(); - - let new_msg_count: u32 = 3; - let new_updated_at: Option = Some(5000); - - let changed = old_msg_count != new_msg_count - || (new_updated_at.is_some() && new_updated_at != old_updated_at); - assert!(changed, "sync must detect updated_at change even when message count is same"); -} - #[test] fn gemini_parser_plain_conversation() { let json = r#"{ diff --git a/src/sync.rs b/src/sync.rs index 69ba64f..ed846b2 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -151,7 +151,9 @@ impl ExistingState { } pub(crate) fn run_sync_job_inner(options: SyncRunOptions) -> Result<()> { - SyncJob::new(options)?.run() + let available_adapters = adapters::all_adapters(); + SyncJob::new(options, Store::open()?, AppConfig::load_or_default(), &available_adapters)? + .run(&available_adapters) } struct SyncJob { @@ -166,10 +168,16 @@ struct SyncJob { } impl SyncJob { - fn new(options: SyncRunOptions) -> Result { - let store = Store::open()?; - let labels = adapters::source_labels(); - let mut config = AppConfig::load_or_default(); + fn new( + options: SyncRunOptions, + store: Store, + mut config: AppConfig, + available_adapters: &[Box], + ) -> Result { + let labels: Vec<_> = available_adapters + .iter() + .map(|adapter| (adapter.id().to_string(), adapter.label().to_string())) + .collect(); config.normalize_sources(&labels); let since_ts = if options.usage_only { None } else { config.sync_window.to_since_cutoff() }; let path_excluder = config.build_path_excluder()?; @@ -185,9 +193,8 @@ impl SyncJob { }) } - fn run(&mut self) -> Result<()> { - let all = adapters::all_adapters(); - for adapter in &all { + fn run(&mut self, available_adapters: &[Box]) -> Result<()> { + for adapter in available_adapters { self.sync_adapter(adapter.as_ref())?; } self.report_progress() @@ -735,18 +742,60 @@ fn path_or_ancestor_matches(path: &str, matcher: &globset::GlobSet) -> bool { mod tests { use std::collections::HashSet; - use crate::adapters::RawSession; + use crate::adapters::{RawMessage, RawSession, ResumeCommand, SourceAdapter}; + use crate::config::AppConfig; use crate::db::{ schema, store::{SessionPath, Store}, }; - use crate::types::Session; + use crate::types::{Role, Session}; use super::{ - BackfillPlan, ExistingSessionAction, decide_existing_session_action, - delete_excluded_sessions_for_source, raw_session_metadata_changed, + BackfillPlan, ExistingSessionAction, SyncJob, SyncRunOptions, + decide_existing_session_action, delete_excluded_sessions_for_source, + raw_session_metadata_changed, }; + struct StaticAdapter { + updated_at: i64, + messages: &'static [&'static str], + } + + impl SourceAdapter for StaticAdapter { + fn id(&self) -> &str { + "test" + } + + fn label(&self) -> &str { + "Test" + } + + fn scan(&self) -> anyhow::Result> { + let messages = self + .messages + .iter() + .enumerate() + .map(|(seq, content)| RawMessage { + role: Role::User, + content: (*content).to_string(), + timestamp: Some(self.updated_at + seq as i64), + }) + .collect(); + Ok(vec![RawSession::search_only( + "raw1", + None, + 1_000, + Some(self.updated_at), + None, + messages, + )]) + } + + fn resume_command(&self, _source_id: &str) -> Option { + None + } + } + fn matcher(pattern: &str) -> globset::GlobSet { let mut builder = globset::GlobSetBuilder::new(); builder.add(globset::Glob::new(pattern).unwrap()); @@ -853,6 +902,42 @@ mod tests { assert!(raw_session_metadata_changed(&raw_with_path, None, &missing)); } + #[test] + fn sync_job_refreshes_changed_session_through_adapter_seam() { + schema::register_sqlite_vec(); + let initial: Vec> = + vec![Box::new(StaticAdapter { updated_at: 2_000, messages: &["first"] })]; + let mut job = SyncJob::new( + SyncRunOptions { + force: false, + verbose: false, + emit: false, + usage_only: false, + backfill_events: false, + sources: None, + }, + Store::open_in_memory().unwrap(), + AppConfig::default(), + &initial, + ) + .unwrap(); + + job.run(&initial).unwrap(); + assert_eq!(job.store.session_meta("test", "raw1").unwrap(), Some((Some(2_000), 1))); + + let updated: Vec> = + vec![Box::new(StaticAdapter { updated_at: 3_000, messages: &["first", "second"] })]; + job.run(&updated).unwrap(); + + assert_eq!(job.store.session_meta("test", "raw1").unwrap(), Some((Some(3_000), 2))); + let session = job.store.list_recent_sessions(1).unwrap().pop().unwrap(); + let messages = job.store.get_messages(&session.id).unwrap(); + assert_eq!( + messages.iter().map(|message| message.content.as_str()).collect::>(), + ["first", "second"] + ); + } + #[test] fn delete_excluded_sessions_for_source_uses_persisted_source_file_path() { schema::register_sqlite_vec();