fix(sessions): skip Codex replay a complete frontier covers

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()

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::<Vec<_>>();
+        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();

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;
     };