Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 20 additions & 6 deletions crates/tracedecay-sessions/src/runtime/hosts/pi.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -140,6 +142,7 @@ struct InvalidHeader {
path: PathBuf,
witness: Witness,
error_kind: io::ErrorKind,
pending: SourceClaim,
}

impl PiDiscoveryReport {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -306,6 +316,7 @@ impl PiSource {
path: candidate,
witness,
error_kind: error.kind(),
pending,
});
continue;
}
Expand Down Expand Up @@ -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()) {
Expand All @@ -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 {
Expand Down
88 changes: 86 additions & 2 deletions crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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!(
Expand Down Expand Up @@ -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();
Expand Down
62 changes: 56 additions & 6 deletions crates/tracedecay-sessions/src/runtime/ingest/quarantine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<bool> {
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 {
Expand Down Expand Up @@ -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>,
Expand Down Expand Up @@ -212,17 +227,43 @@ impl SourceClaim {
ledger: Option<&SourceLedger>,
path: &Path,
) -> TranscriptIngestResult<Option<Self>> {
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<FileSourceObservation> {
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(
Expand All @@ -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()
Expand Down
Loading