diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index cae50b9f87..6d2ae4281f 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -579,9 +579,14 @@ impl CodexDiscoveryHub { detail: "Codex replay index disappeared before delivery", }, )?; - let effective_position = (consumer_generation == index.generation) - .then_some(consumer_position) - .flatten(); + let effective_position = + if frontier.is_complete() && frontier == index.frontier { + index.paths.last().cloned() + } else { + (consumer_generation == index.generation) + .then_some(consumer_position) + .flatten() + }; let index_generation = index.generation; let lower = effective_position .as_ref() diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index b453fedc65..d4e5ba1c6d 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -1713,6 +1713,74 @@ mod recent_first_discovery_tests { assert_eq!(index.files_considered, 73); } + #[tokio::test] + async fn a_late_consumer_whose_durable_frontier_covers_the_corpus_replays_nothing() { + crate::runtime::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority(); + let temp = TempDir::new().unwrap(); + let home = temp.path(); + let rollouts = (0..21) + .map(|index| { + write_dated_rollout(home, ("2026", "08", "23"), &format!("covered-{index:03}")) + }) + .collect::>(); + let settled = rollouts + .iter() + .all(|path| spin_until_jsonl_change_settled(path)); + let source = CodexSource::with_home(home); + let bounds = TranscriptDiscoveryBounds::from_discovered_units(8); + + let previous_process = CodexDiscoveryHub::default(); + previous_process.register("project", Some(home)); + previous_process.register("profile", Some(home)); + let (_, project_frontier) = drain_hub_consumer( + &previous_process, + "project", + &source, + bounds, + CodexDiscoveryFrontier::initial(), + ) + .await; + let (_, profile_frontier) = drain_hub_consumer( + &previous_process, + "profile", + &source, + bounds, + CodexDiscoveryFrontier::initial(), + ) + .await; + assert_eq!(project_frontier, profile_frontier); + + let restarted = CodexDiscoveryHub::default(); + restarted.register("project", Some(home)); + let (project_paths, _) = + drain_hub_consumer(&restarted, "project", &source, bounds, project_frontier).await; + restarted.register("profile", Some(home)); + let (profile_paths, profile_after_restart) = + drain_hub_consumer(&restarted, "profile", &source, bounds, profile_frontier).await; + + if settled { + assert!(project_paths.is_empty()); + assert!( + profile_paths.is_empty(), + "a consumer whose complete frontier matches the corpus epoch was re-offered {} \ + rollouts", + profile_paths.len() + ); + assert_eq!(profile_after_restart, profile_frontier); + } + + let added = write_dated_rollout(home, ("2026", "08", "24"), "after-restart"); + let (grown, _) = drain_hub_consumer( + &restarted, + "profile", + &source, + bounds, + profile_after_restart, + ) + .await; + assert!(grown.contains(&added)); + } + #[tokio::test] async fn completed_secondary_source_rebuilds_after_add_and_delete() { crate::runtime::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority(); diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs index 9330bde261..9b492ece73 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs @@ -1,12 +1,14 @@ //! Strict advisory runtime acceptance over authentic provider response captures. use std::collections::{BTreeSet, VecDeque}; +use std::io::SeekFrom; use std::process::{Command, Stdio}; use std::sync::{Arc, Mutex}; use std::time::Duration; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; use tracedecay_application::advisory::ci_runtime::GitHubCiOfficialResponseDecoderV1; #[cfg(feature = "test-transport")] use tracedecay_application::advisory::ci_runtime::{ @@ -1216,6 +1218,26 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() { // client protocol; only a non-retryable failure is a test failure. let advisory_deadline = std::time::Instant::now() + Duration::from_secs(60); let advisory = loop { + let daemon_log_before_invocation = + tokio::time::timeout_at(tokio::time::Instant::from_std(advisory_deadline), async { + const MAX_LOG_BYTES: u64 = 16 * 1024; + let mut file = tokio::fs::File::open(&daemon_log).await?; + let end = file.seek(SeekFrom::End(0)).await?; + file.seek(SeekFrom::Start(end.saturating_sub(MAX_LOG_BYTES))) + .await?; + let mut bytes = Vec::new(); + file.take(MAX_LOG_BYTES).read_to_end(&mut bytes).await?; + Ok::<_, std::io::Error>(String::from_utf8_lossy(&bytes).into_owned()) + }) + .await + .map_or_else( + |_| "daemon log snapshot exceeded the registration deadline".to_owned(), + |result| result.unwrap_or_else(|error| format!("daemon log unavailable: {error}")), + ); + assert!( + std::time::Instant::now() < advisory_deadline, + "advisory registration deadline elapsed before invocation\ndaemon log captured before invocation (up to 16 KiB):\n{daemon_log_before_invocation}", + ); let advisory = common::tracedecay_command_with_home(environment.home()) .args([ "tool", @@ -1243,13 +1265,13 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() { }); assert!( retryable_unavailable, - "registered advisory cycle failed\nstdout:\n{stdout}\nstderr:\n{}", + "registered advisory cycle failed\nstdout:\n{stdout}\nstderr:\n{}\ndaemon log captured before this invocation (up to 16 KiB):\n{daemon_log_before_invocation}", String::from_utf8_lossy(&advisory.stderr), ); assert!( std::time::Instant::now() < advisory_deadline, - "advisory cycle stayed unavailable past the deferred registration deadline\nstdout:\n{stdout}\ndaemon log:\n{}", - std::fs::read_to_string(&daemon_log).expect("read isolated advisory daemon log"), + "advisory cycle stayed unavailable past the deferred registration deadline\nstdout:\n{stdout}\nstderr:\n{}\ndaemon log captured before this invocation (up to 16 KiB):\n{daemon_log_before_invocation}", + String::from_utf8_lossy(&advisory.stderr), ); tokio::time::sleep(Duration::from_millis(250)).await; };