diff --git a/src/extraction/mod.rs b/src/extraction/mod.rs index 5445cb2b..076ce8b9 100644 --- a/src/extraction/mod.rs +++ b/src/extraction/mod.rs @@ -64,6 +64,10 @@ pub struct SessionExtractionBatch { pub ingested_session_ids: BTreeSet, 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, } @@ -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(), diff --git a/src/extraction/tests.rs b/src/extraction/tests.rs index c5568ac8..b4491b97 100644 --- a/src/extraction/tests.rs +++ b/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() { diff --git a/src/main.rs b/src/main.rs index 61799440..fa47aa46 100644 --- a/src/main.rs +++ b/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() + ); + } } }