diff --git a/crates/tracedecay-sessions/src/runtime/hosts/pi.rs b/crates/tracedecay-sessions/src/runtime/hosts/pi.rs index bc1f54997f..ce9f47c36a 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/pi.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/pi.rs @@ -28,7 +28,9 @@ use tracedecay_store::{ParseOffset, observation::ObservationCoverageReason}; use crate::admission::{HostAdmission, HostDiscoveryQueueEntry}; use crate::observation::ObservationCancellation; use crate::runtime::host_scan::{HOST_SCAN_WINDOW, HostScanBudget, isolatable_source_error}; -use crate::runtime::ingest::quarantine::{SourceClaim, SourceLedger, SourceSweep, Witness}; +use crate::runtime::ingest::quarantine::{ + FileSourceObservation, SourceClaim, SourceLedger, SourceSweep, Witness, +}; use crate::runtime::jsonl_observation_admission::{ JsonlFrameAdmission, JsonlObservationAdmissionProgress, JsonlObservationAdmissionRequest, admit_jsonl_observations, @@ -140,6 +142,7 @@ struct InvalidHeader { path: PathBuf, witness: Witness, error_kind: io::ErrorKind, + pending: SourceClaim, } impl PiDiscoveryReport { @@ -276,9 +279,16 @@ impl PiSource { { continue; } - let pending = match SourceClaim::observe_file_blocking(ledger, &candidate) { - Ok(Some(pending)) => pending, - Ok(None) => continue, + let pending = match SourceClaim::observe_file_discovery_blocking( + ledger, &candidate, + ) { + Ok(FileSourceObservation::Read(pending)) => pending, + Ok(FileSourceObservation::Skipped) => continue, + Ok(FileSourceObservation::Quarantined) => { + discovery.quarantined_headers = + discovery.quarantined_headers.saturating_add(1); + continue; + } Err(TranscriptIngestError::ScanIo { source, .. }) => { discovery.record_failure( PiDiscoveryFailureKind::SessionHeaderUnavailable, @@ -306,6 +316,7 @@ impl PiSource { path: candidate, witness, error_kind: error.kind(), + pending, }); continue; } @@ -888,6 +899,7 @@ async fn reconcile_invalid_headers( discovery.quarantined_headers = discovery.quarantined_headers.saturating_add(1); let memo = PathMemo::read(facade, scope, &invalid.path).await?; if memo.disposition(Some(invalid.witness)) == Some(Disposition::InvalidHeader) { + invalid.pending.quarantined()?; continue; } if !budget.try_charge_input(invalid.witness.len()) { @@ -902,9 +914,11 @@ async fn reconcile_invalid_headers( break; } if matches!(current, Ok(HeaderRead::Invalid { witness, .. }) if witness == invalid.witness) + && memo + .record(facade, scope, invalid.witness, Disposition::InvalidHeader) + .await? { - memo.record(facade, scope, invalid.witness, Disposition::InvalidHeader) - .await?; + invalid.pending.quarantined()?; } discovery.failure_count = discovery.failure_count.saturating_add(1); if discovery.failures.len() < MAX_DISCOVERY_FAILURE_EVIDENCE { diff --git a/crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs index 6b119e7a3a..80b705befb 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs @@ -10,12 +10,17 @@ use tracedecay_domain::{ use super::{ MAX_DISCOVERY_CANDIDATES, MAX_DISCOVERY_FAILURE_EVIDENCE, MAX_DISCOVERY_INPUT_BYTES, MAX_DISCOVERY_UNITS, PI_AGENT_RELATIVE, PiDiscoveryReport, PiSource, capture_pi_observations, - capture_pi_session, pi_agent_dir_for, session_dir_name, warn_discovery_failures, + capture_pi_session, capture_pi_sweep, pi_agent_dir_for, session_dir_name, + warn_discovery_failures, }; use crate::admission::{HostAdmission, test_support::MemoryHostAdmission}; use crate::observation::ObservationCancellation; use crate::runtime::host_scan::{HOST_SCAN_WINDOW, HostScanBudget}; -use crate::runtime::source::{FileDiscoveryReport, TranscriptDiscoveryBounds}; +use crate::runtime::hosts::codex::CodexDiscoveryHub; +use crate::runtime::ingest::quarantine::SourceSweep; +use crate::runtime::source::{ + FileDiscoveryReport, TranscriptDiscoveryBounds, spin_until_jsonl_change_settled, +}; const FIXTURE_NAME: &str = "2026-09-25T16-00-00-000Z_5f0c2a8e-3b1d-4c7e-9a2f-6d8e1b4c7a90.jsonl"; const FIXTURE: &str = include_str!( @@ -280,6 +285,85 @@ async fn repeated_discovery_summary_keeps_malformed_sources_and_coverage_refused ); } +#[tokio::test] +async fn a_quarantined_pi_header_skips_unchanged_discovery_without_masking_coverage() { + let layout = Layout::new(); + let path = layout.session_dir().join(FIXTURE_NAME); + std::fs::write(&path, b"{\"not\":\"a Pi session header\"}\n").unwrap(); + assert!( + spin_until_jsonl_change_settled(&path), + "the initial file must have a settled witness" + ); + + let hub = CodexDiscoveryHub::default(); + hub.register("pi-headers", Some(&layout.agent_dir)); + let admission = MemoryHostAdmission::default(); + let scope = ObservationScopeV1::Profile; + let run = || async { + let mut sweep = SourceSweep::new("pi", Some((&hub, "pi-headers"))); + capture_pi_sweep( + &admission, + &layout.source(), + &layout.project, + scope.clone(), + None, + &ObservationCancellation::default(), + &mut sweep, + ) + .await + .unwrap() + }; + + let first = run().await; + assert_eq!(first.discovery_failures, 1); + assert!(first.deferred); + assert!(admission.observations().is_empty()); + + // A second discovery with the same source ledger must not open/parse the + // quarantined header again, but its evidence still denies complete coverage. + let sweep = SourceSweep::new("pi", Some((&hub, "pi-headers"))); + let (discovery, _) = layout + .source() + .discover( + &layout.project, + TranscriptDiscoveryBounds::from_discovered_units(MAX_DISCOVERY_CANDIDATES), + None, + HostScanBudget::new( + MAX_DISCOVERY_INPUT_BYTES, + MAX_DISCOVERY_UNITS, + std::time::Instant::now() + HOST_SCAN_WINDOW, + ObservationCancellation::default(), + ), + sweep.ledger().as_ref(), + ) + .unwrap(); + assert!(discovery.invalid_headers.is_empty()); + assert_eq!(discovery.quarantined_headers, 1); + assert_eq!(discovery.failure_count, 0); + + let second = run().await; + assert_eq!(second.discovery_failures, 0); + assert!(second.deferred); + let coverage = admission + .get_parse_offset(&scope, "host-coverage://pi/v1") + .await + .unwrap() + .unwrap(); + assert_eq!(coverage.file_id, 2, "quarantine must remain partial"); + + // Changed source evidence releases the in-memory quarantine; the durable + // path memo then re-evaluates the repaired header and admits the session. + layout.write_fixture(); + assert!( + spin_until_jsonl_change_settled(&path), + "the repaired file must have a settled witness" + ); + let recovered = run().await; + assert!(!recovered.deferred, "{recovered:?}"); + assert_eq!(recovered.discovery_failures, 0); + assert_eq!(admission.observations().len(), 9); +} + #[tokio::test] async fn empty_host_is_observed_once_without_revising_unchanged_frontier() { let layout = Layout::new(); diff --git a/crates/tracedecay-sessions/src/runtime/ingest/quarantine.rs b/crates/tracedecay-sessions/src/runtime/ingest/quarantine.rs index 82f8cb8c12..a1d60c5bfc 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/quarantine.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/quarantine.rs @@ -56,11 +56,18 @@ impl SourceKey { impl SourceLedger { pub(crate) fn holds(&self, key: &SourceKey, witness: Witness) -> bool { + self.observed_quarantine(key, witness).is_some() + } + + /// An exact in-memory file witness can be settled or quarantined. Callers + /// that skip opening a quarantined header must retain partial coverage. + pub(crate) fn observed_quarantine(&self, key: &SourceKey, witness: Witness) -> Option { self.inner .lock() .unwrap_or_else(PoisonError::into_inner) .get(key) - .is_some_and(|recorded| recorded.witness == witness) + .filter(|recorded| recorded.witness == witness) + .map(|recorded| recorded.quarantined) } fn recorded_converged(&self, key: &SourceKey) -> bool { @@ -175,6 +182,14 @@ pub(crate) struct SourceClaim { settled: Option<(SourceLedger, Witness)>, } +/// Discovery may avoid opening a known-bad file, but must never report that +/// quarantined input as complete coverage. +pub(crate) enum FileSourceObservation { + Read(SourceClaim), + Skipped, + Quarantined, +} + impl SourceClaim { fn unless_recorded( ledger: Option<&SourceLedger>, @@ -212,17 +227,43 @@ impl SourceClaim { ledger: Option<&SourceLedger>, path: &Path, ) -> TranscriptIngestResult> { - let key = SourceKey::file(path); - if ledger.is_none() { - return Ok(Some(Self { key, settled: None })); + match Self::observe_file_discovery_blocking(ledger, path)? { + FileSourceObservation::Read(claim) => Ok(Some(claim)), + FileSourceObservation::Skipped | FileSourceObservation::Quarantined => Ok(None), } + } + + /// Preserve the reason a file was skipped for discovery sources that must + /// report quarantined evidence as partial, rather than complete, coverage. + pub(crate) fn observe_file_discovery_blocking( + ledger: Option<&SourceLedger>, + path: &Path, + ) -> TranscriptIngestResult { + let key = SourceKey::file(path); + let Some(ledger) = ledger else { + return Ok(FileSourceObservation::Read(Self { key, settled: None })); + }; // Codex archives a session by moving its rollout away; failing on the // vanished path would replay the unacknowledged discovery pass forever. let Some(metadata) = stat_if_present(path, "stat transcript for convergence")? else { - return Ok(None); + return Ok(FileSourceObservation::Skipped); }; let witness = Witness::settled_file(path, &metadata)?; - Ok(Self::unless_recorded(ledger, key, witness)) + if let Some(witness) = witness { + if let Some(quarantined) = ledger.observed_quarantine(&key, witness) { + return Ok(if quarantined { + FileSourceObservation::Quarantined + } else { + FileSourceObservation::Skipped + }); + } + } else if ledger.recorded_converged(&key) { + return Ok(FileSourceObservation::Skipped); + } + Ok(FileSourceObservation::Read(Self { + key, + settled: witness.map(|witness| (ledger.clone(), witness)), + })) } pub(crate) fn admitted( @@ -244,6 +285,15 @@ impl SourceClaim { } } + /// Cache a durable invalid-source disposition only after the canonical + /// source memo has confirmed it; this ledger is never its authority. + pub(crate) fn quarantined(self) -> TranscriptIngestResult<()> { + match self.settled { + Some((ledger, witness)) => ledger.record(&self.key, witness, true), + None => Ok(()), + } + } + pub(crate) fn admission_settles(&self, source_deferred: bool, covered_through: u64) -> bool { self.settled .as_ref()