From 6049e6641a7e8ad7f80db27c792c079a385860e8 Mon Sep 17 00:00:00 2001 From: Szowesgad Date: Wed, 15 Jul 2026 15:19:22 +0200 Subject: [PATCH] =?UTF-8?q?[claude/hotfix]=20fix(store):=20drop=20duplicat?= =?UTF-8?q?e=20physical=20sources=20of=20one=20logical=20session=20?= =?UTF-8?q?=E2=80=94=20freshest=20copy=20wins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - aicx all -H 30 na żywym korpusie padoł na końcu biegu: 'duplicate canonical card id: card3:1980d570c88bce990c7e30e8' — ta samo logiczno sesjo odkryto z dwóch ścieżek fizycznych (np. live rollout + zarchiwizowano kopia) - root cause: ingested_session_ids (BTreeSet) dubla nie wstawiał, ale canonical_cards.extend() szedł per-źródło -> dwa komplety kart o identycznych content-derived id w projekcji - fix: order_sources_freshest_first (mtime desc, sort stabilny) + strażnik per-batch: druga kopia już-ingestowanej sesji -> drop liczony w duplicate_sources (NIE skip: nie trzymie watermarku, nie triggeruje retry) - summary raportuje duplicate_sources=N tylko gdy N>0 - test: duplicate_session_sources_are_ordered_freshest_first Authored-By: claude session_id: 49a84e4c-f24d-4167-9f55-ae0f8123a66f timestamp: 2026_0715_1540_CEST runtime: claude-code-interactive --- src/extraction/mod.rs | 33 +++++++++++++++++++++++++++++++-- src/extraction/tests.rs | 38 ++++++++++++++++++++++++++++++++++++++ src/main.rs | 22 ++++++++++++++++------ 3 files changed, 85 insertions(+), 8 deletions(-) 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() + ); + } } }