diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3e82ac6c34..dc96d97cb4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -596,7 +596,9 @@ jobs: shell: bash # Compile and link every target, but defer unused test runfiles trees. # Bazel still materializes runfiles required by build-time tools. - run: bazel build --config=ci --nobuild_runfile_links //... + # Four concurrent Rust actions exhausted the hosted runner's commit + # limit while mapping asupersync metadata (Windows error 1455). + run: bazel build --config=ci --jobs=2 --nobuild_runfile_links //... - name: Build and test native targets if: ${{ github.event_name == 'workflow_dispatch' && inputs.run_os }} @@ -619,7 +621,7 @@ jobs: *) echo "Expected a Bazel target pattern: $target" >&2; exit 1 ;; esac done - bazel test --config=ci --test_output=all "${targets[@]}" + bazel test --config=ci --jobs=2 --test_output=all "${targets[@]}" - name: Upload failed test diagnostics if: failure() diff --git a/crates/tracedecay-sessions/src/runtime/hosts/pi.rs b/crates/tracedecay-sessions/src/runtime/hosts/pi.rs index 1d96a58c3b..c8d77c3571 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/pi.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/pi.rs @@ -66,6 +66,8 @@ static PI_DISCOVERY_FAILURE_GATE: StateChangeLogGate< String, (PiDiscoveryFailureKind, io::ErrorKind), > = StateChangeLogGate::new(); +static PI_DISCOVERY_SUMMARY_GATE: StateChangeLogGate = + StateChangeLogGate::new(); /// The agent directory Pi loads for `home`. `PI_CODING_AGENT_DIR` relocates /// it only to an absolute directory inside `home`: a sandbox or tempdir home @@ -560,7 +562,7 @@ pub(crate) async fn capture_pi_sweep( ) .await .map_err(|_| TranscriptIngestError::BlockingScanTaskFailed { provider: PROVIDER })??; - warn_discovery_failures(&discovery); + warn_discovery_failures(&source.sessions_root(), &discovery); let discovery_truncated = discovery.files.is_truncated(); let discovery_skipped = discovery.files.skipped_oversized_entries; let last_discovered_entry = if cancellation.is_cancelled() { @@ -915,7 +917,7 @@ async fn persist_coverage( .await } -fn warn_discovery_failures(discovery: &PiDiscoveryReport) { +fn warn_discovery_failures(sessions_root: &Path, discovery: &PiDiscoveryReport) { for failure in &discovery.failures { if !PI_DISCOVERY_FAILURE_GATE.admit( failure.source_digest.clone(), @@ -931,7 +933,18 @@ fn warn_discovery_failures(discovery: &PiDiscoveryReport) { "Pi session discovery is incomplete" ); } - if discovery.failure_count > discovery.failures.len() as u64 { + let summary_key = sessions_root.to_path_buf(); + let fully_reported = discovery.failure_count <= discovery.failures.len() as u64; + if fully_reported { + // An interrupted empty sweep has not established recovery; keep the + // last admitted summary so the next complete pass does not re-log it. + if discovery.scan_complete || discovery.failure_count > 0 { + PI_DISCOVERY_SUMMARY_GATE.clear(&summary_key); + } + } else if PI_DISCOVERY_SUMMARY_GATE.admit( + summary_key, + (discovery.failure_count, discovery.failures.len()), + ) { tracing::warn!( provider = PROVIDER, failure_count = discovery.failure_count, diff --git a/crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs index 3bb14de497..2ac8464cc2 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/pi_tests.rs @@ -8,11 +8,14 @@ use tracedecay_domain::{ }; use super::{ - PI_AGENT_RELATIVE, PiSource, capture_pi_observations, capture_pi_session, pi_agent_dir_for, - session_dir_name, + 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, }; 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}; const FIXTURE_NAME: &str = "2026-09-25T16-00-00-000Z_5f0c2a8e-3b1d-4c7e-9a2f-6d8e1b4c7a90.jsonl"; const FIXTURE: &str = include_str!( @@ -56,6 +59,17 @@ impl Layout { path } + fn write_malformed_files(&self, count: usize) { + for index in 0..count { + std::fs::write( + self.session_dir() + .join(format!("2026-09-25T16-00-00-000Z_invalid-{index}.jsonl")), + "{\"not\":\"a Pi session header\"}\n", + ) + .unwrap(); + } + } + fn source(&self) -> PiSource { PiSource::with_agent_dir(&self.agent_dir) } @@ -93,6 +107,148 @@ fn record<'a>( .unwrap_or_else(|| panic!("no record {stable}")) } +#[test] +fn discovery_summary_reports_changes_and_recovery_without_repeating_unchanged_failures() { + let first = Layout::new(); + let second = Layout::new(); + first.write_malformed_files(24); + second.write_malformed_files(25); + let report = |layout: &Layout| { + 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(), + ), + None, + ) + .unwrap() + .0 + }; + let first_report = report(&first); + assert_eq!(first_report.failure_count, 24); + assert_eq!(first_report.failures.len(), MAX_DISCOVERY_FAILURE_EVIDENCE); + let summary_count = |layout: &Layout, discovery: &PiDiscoveryReport| { + let (_, output) = tracedecay_runtime_core::logging::capture_formatted_tracing(|| { + warn_discovery_failures(&layout.source().sessions_root(), discovery); + }); + output + .matches("additional Pi discovery failures were bounded") + .count() + }; + assert_eq!(summary_count(&first, &first_report), 1); + assert_eq!( + summary_count(&first, &first_report), + 0, + "an unchanged refusal summary is not repeated" + ); + let interrupted = PiDiscoveryReport { + files: FileDiscoveryReport { + paths: Vec::new(), + truncated: None, + skipped_oversized_entries: 0, + bytes_charged: 0, + files_considered: 0, + }, + failures: Vec::new(), + failure_count: 0, + scan_complete: false, + reached_end: false, + }; + assert_eq!(summary_count(&first, &interrupted), 0); + assert_eq!( + summary_count(&first, &first_report), + 0, + "an interrupted empty scan must not clear the admitted summary" + ); + first.write_malformed_files(25); + let changed = report(&first); + assert_eq!(changed.failure_count, 25); + assert_eq!( + summary_count(&first, &changed), + 1, + "changed totals are logged" + ); + assert_eq!( + summary_count(&second, &report(&second)), + 1, + "another sessions root is logged independently" + ); + for index in MAX_DISCOVERY_FAILURE_EVIDENCE..25 { + std::fs::remove_file( + first + .session_dir() + .join(format!("2026-09-25T16-00-00-000Z_invalid-{index}.jsonl")), + ) + .unwrap(); + } + let fully_reported = report(&first); + assert_eq!( + fully_reported.failure_count, + MAX_DISCOVERY_FAILURE_EVIDENCE as u64 + ); + assert_eq!( + fully_reported.failures.len(), + MAX_DISCOVERY_FAILURE_EVIDENCE + ); + assert_eq!(summary_count(&first, &fully_reported), 0); + first.write_malformed_files(25); + assert_eq!( + summary_count(&first, &report(&first)), + 1, + "full failure evidence clears the previous summary" + ); + for index in 0..25 { + std::fs::remove_file( + first + .session_dir() + .join(format!("2026-09-25T16-00-00-000Z_invalid-{index}.jsonl")), + ) + .unwrap(); + } + first.write_fixture(); + let recovered = report(&first); + assert_eq!(recovered.failure_count, 0); + assert_eq!(summary_count(&first, &recovered), 0); + first.write_malformed_files(25); + let recurred = report(&first); + assert_eq!( + summary_count(&first, &recurred), + 1, + "recurrence after recovery is logged" + ); + assert_eq!(summary_count(&first, &recurred), 0); +} + +#[tokio::test] +async fn repeated_discovery_summary_keeps_malformed_sources_and_coverage_refused() { + let layout = Layout::new(); + let malformed = MAX_DISCOVERY_FAILURE_EVIDENCE + 1; + layout.write_malformed_files(malformed); + let admission = MemoryHostAdmission::default(); + for _ in 0..2 { + let outcome = capture(&layout, &admission).await; + assert_eq!(outcome.discovery_failures, malformed as u64); + assert!(outcome.deferred); + assert!(admission.observations().is_empty()); + let coverage = admission + .get_parse_offset(&ObservationScopeV1::Profile, "host-coverage://pi/v1") + .await + .unwrap() + .unwrap(); + assert_eq!( + coverage.file_id, 2, + "malformed sources retain partial coverage" + ); + } +} + #[tokio::test] async fn empty_host_is_observed_once_without_revising_unchanged_frontier() { let layout = Layout::new();