From 7ee5d212ac142296cc0f3c8e51d9c97aee39a042 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 3 Nov 2026 10:16:36 -0800 Subject: [PATCH] perf(sessions): skip re-reading quarantined Pi headers Preserve the exact file-witness quarantine disposition in the shared source ledger. Once durable Pi memo admission confirms an invalid header, avoid reopening it on later discovery passes without treating skipped evidence as complete coverage. A changed file witness re-enables header validation. Add a real Pi discovery and capture regression for skip, partial coverage, and recovery on a repaired file. Refs #4481 --- .../src/runtime/hosts/pi.rs | 26 ++++-- .../src/runtime/hosts/pi_tests.rs | 88 ++++++++++++++++++- .../src/runtime/ingest/quarantine.rs | 62 +++++++++++-- 3 files changed, 162 insertions(+), 14 deletions(-) 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()