[claude/hotfix] fix(store): drop duplicate physical sources for the same logical session

src/extraction/mod.rs

@@ -64,6 +64,10 @@ pub struct SessionExtractionBatch {
     pub ingested_session_ids: BTreeSet<String>,
     pub selected_sessions: usize,
     pub ingested_sessions: usize,
+    /// Extra physical sources whose logical session was already ingested in
+    /// this batch (e.g. an archived copy of the same rollout file). Dropped,
+    /// not skipped: they must not hold the watermark or trigger retries.
+    pub duplicate_sources: usize,
     pub skipped: Vec<SessionSkip>,
 }
 
@@ -76,11 +80,27 @@ impl SessionExtractionBatch {
             canonical_cards: Vec::new(),
             ingested_session_ids: BTreeSet::new(),
             selected_sessions: 0,
+            duplicate_sources: 0,
             skipped: Vec::new(),
         }
     }
 }
 
+/// Freshest copy first: one logical session can be discoverable through more
+/// than one physical path (e.g. a live rollout file plus an archived copy).
+/// Parsing the most recently modified source first lets the per-batch
+/// session-id guard drop the older duplicates, so the canonical projection
+/// never sees two card sets for one session (`duplicate canonical card id`).
+#[cfg(feature = "app")]
+fn order_sources_freshest_first(sources: &mut [crate::session_catalog::CatalogSource]) {
+    sources.sort_by(|left, right| {
+        right
+            .fingerprint
+            .modified_unix_nanos
+            .cmp(&left.fingerprint.modified_unix_nanos)
+    });
+}
+
 /// Discover identities with the catalog and parse every selected session once.
 ///
 /// App-only: session discovery (`session_catalog`), parser dispatch, and
@@ -105,11 +125,13 @@ pub fn extract_agent_sessions(
     let scan = crate::session_catalog::SessionCatalog::new(agent, &root)?.scan_with_stats();
     let parser_agent = parser_agent(agent);
     let mut batch = SessionExtractionBatch::default();
-    for source in scan
+    let mut sources: Vec<_> = scan
         .result?
         .into_iter()
         .filter(|source| source_is_selected(source.fingerprint.modified_unix_nanos, config))
-    {
+        .collect();
+    order_sources_freshest_first(&mut sources);
+    for source in sources {
         batch.selected_sessions += 1;
         let parsed = crate::parser_dispatch::parse_file(
             parser_agent,
@@ -119,6 +141,13 @@ pub fn extract_agent_sessions(
         );
         match parsed {
             Ok(session) => {
+                if batch
+                    .ingested_session_ids
+                    .contains(&session.model().session_id)
+                {
+                    batch.duplicate_sources += 1;
+                    continue;
+                }
                 let projection = match aicx_parser::projections::project_validated_session(
                     &session,
                     &canonical_projection_config(),

src/extraction/tests.rs

@@ -32,6 +32,44 @@ fn stale_fatal_completeness_is_not_mislabeled_in_flight() {
     ));
 }
 
+#[test]
+#[cfg(feature = "app")]
+fn duplicate_session_sources_are_ordered_freshest_first() {
+    fn source(source_id: &str, modified_unix_nanos: u128) -> crate::session_catalog::CatalogSource {
+        crate::session_catalog::CatalogSource {
+            agent: crate::session_catalog::AgentKind::Codex,
+            source_id: source_id.to_owned(),
+            logical_session_id: Some("session-dup".to_owned()),
+            aliases: Vec::new(),
+            filename_aliases: Vec::new(),
+            scoped_children: Vec::new(),
+            path: std::path::PathBuf::from(format!("/tmp/{source_id}.jsonl")),
+            identity_inferred: false,
+            fingerprint: crate::session_catalog::SourceFingerprint {
+                len: 1,
+                modified_unix_nanos,
+            },
+            header_truncated: false,
+        }
+    }
+    let mut sources = vec![
+        source("stale-archived-copy", 100),
+        source("fresh-live-rollout", 300),
+        source("middle-copy", 200),
+    ];
+    order_sources_freshest_first(&mut sources);
+    let order: Vec<&str> = sources
+        .iter()
+        .map(|source| source.source_id.as_str())
+        .collect();
+    assert_eq!(
+        order,
+        vec!["fresh-live-rollout", "middle-copy", "stale-archived-copy"],
+        "the freshest physical copy must be parsed first so the per-batch \
+         session-id guard drops the older duplicates"
+    );
+}
+
 #[test]
 #[cfg(feature = "app")]
 fn session_selection_uses_watermark_as_retry_window_floor() {

src/main.rs

@@ -6353,12 +6353,22 @@ fn emit_session_batch_summary(agent: &str, batch: &sources::SessionExtractionBat
         eprintln!("    recover: {}", skip.recover);
     }
     if batch.selected_sessions > 0 || !batch.skipped.is_empty() {
-        eprintln!(
-            "  [{}] session ingest summary: ingested={} skipped={}",
-            agent,
-            batch.ingested_sessions,
-            batch.skipped.len()
-        );
+        if batch.duplicate_sources > 0 {
+            eprintln!(
+                "  [{}] session ingest summary: ingested={} skipped={} duplicate_sources={}",
+                agent,
+                batch.ingested_sessions,
+                batch.skipped.len(),
+                batch.duplicate_sources
+            );
+        } else {
+            eprintln!(
+                "  [{}] session ingest summary: ingested={} skipped={}",
+                agent,
+                batch.ingested_sessions,
+                batch.skipped.len()
+            );
+        }
     }
 }