test(sync): exercise refresh through adapter seam

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<i64> = 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<i64> = 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#"{

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<Self> {
-        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<dyn adapters::SourceAdapter>],
+    ) -> Result<Self> {
+        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<dyn adapters::SourceAdapter>]) -> 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<Vec<RawSession>> {
+            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<ResumeCommand> {
+            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<Box<dyn SourceAdapter>> =
+            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<Box<dyn SourceAdapter>> =
+            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::<Vec<_>>(),
+            ["first", "second"]
+        );
+    }
+
     #[test]
     fn delete_excluded_sessions_for_source_uses_persisted_source_file_path() {
         schema::register_sqlite_vec();