diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs
index 0935aee090..c82155ff9b 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs
+++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs
@@ -1288,39 +1288,80 @@ async fn quiesced_background_reconcile_admission(
     admission
 }
 
-/// Hold background admission only once no pass and no wake are outstanding.
+/// Settle the owner *and* burn the coalesced wake permit a settled owner can
+/// still be holding, so the global admission is idle and stays idle.
 ///
-/// [`quiesced_background_reconcile_admission`] can win the permit while text
-/// seating still owns `reconcile_in_progress`: that seating is a real refresh,
-/// and a freshness read then reports `Verifying` even though the scheduler
-/// mutex is free. A caller that must attribute `Verifying` to something other
-/// than an unrelated mutex holder has to sample after both that guard and any
-/// coalesced wake are clear, and keep the permit so a successor cannot start
-/// under the sample.
-async fn hold_settled_background_admission(
+/// [`wait_for_settled_owner`] proves the pending-arrival slot is empty now, but
+/// emptiness is not the whole queue: `note_worker_continuation` replenishes the
+/// `Notify` permit whenever it cannot claim the slot, and `note_wake` posts a
+/// permit of its own for an arrival a running pass then claims. Either leaves a
+/// banked permit behind a settled owner, and the no-op pass it starts owns the
+/// single background admission while it runs. A test that reads
+/// `available_permits`, or one that reads the freshness ladder (whose
+/// `refresh_in_flight` is the pass counter *or* the pending slot), samples that
+/// pass and not the quiet worktree it set up.
+///
+/// Holding the permit parks such a pass at its dequeue point, before it claims
+/// an arrival or enters its guard. Releasing it hands it straight over, so the
+/// drain is done only once a release leaves the permit free.
+///
+/// The registry must be single-permit
+/// ([`CodeIndexSchedulerRegistryV1::with_background_reconcile_permits`]): with
+/// the host's default bound, one held permit parks nothing.
+async fn settled_owner_with_idle_admission(
     registry: &CodeIndexSchedulerRegistryV1,
     project_root: &Path,
-) -> tokio::sync::OwnedSemaphorePermit {
+) {
+    let admission = registry.background_reconcile_admission();
     let deadline = Instant::now() + SERVING_SEAT_FAILURE_CEILING;
     loop {
-        let admission = registry
-            .background_reconcile_admission()
-            .acquire_owned()
-            .await
-            .expect("hold background worker at its dequeue point");
-        wait_for_quiescent_owner_pass(registry, project_root).await;
-        let pending_wake = registry.pending_wake_micros_for_root(project_root).await;
-        let in_progress = registry.reconcile_in_progress_for_test(project_root).await;
-        if pending_wake == Some(0) && !in_progress {
-            return admission;
+        drop(quiesced_background_reconcile_admission(registry, project_root).await);
+        // A banked permit is claimed by the worker's very next `notified()`,
+        // whose first act is to take this admission. Give that claim its turn,
+        // then settle: a pass that did start moves the guard or the slot this
+        // wait joins, and the free permit afterwards is the proof none is left.
+        tokio::time::sleep(Duration::from_millis(5)).await;
+        wait_for_settled_owner(registry, project_root).await;
+        if admission.available_permits() == 1 {
+            return;
         }
-        drop(admission);
         assert!(
             Instant::now() <= deadline,
-            "the owner for {} never settled before a freshness sample",
+            "the admission for {} never went idle",
             project_root.display()
         );
-        wait_for_settled_owner(registry, project_root).await;
+        tokio::time::sleep(Duration::from_millis(2)).await;
+    }
+}
+
+/// Empty the coalesced pending-wake slot and prove the owner's pass tail is
+/// done disturbing it.
+///
+/// The caller must already hold the single background admission, so no further
+/// pass can start. One pass can still be finishing: the worker releases that
+/// admission halfway through its body and drops its `reconcile_pass` guard
+/// before the branches that call `note_worker_continuation`, so both
+/// `reconcile_in_progress` and the slot read quiet while the tail is still
+/// about to stamp `BusyFollowUp` into it. [`wait_for_settled_owner`] samples
+/// exactly those two, so it cannot see that tail. With the admission held the
+/// tail is finite and unrepeatable, so clearing until the slot survives a quiet
+/// window is the proof the settle cannot give.
+async fn clear_pending_wake_until_quiet(
+    registry: &CodeIndexSchedulerRegistryV1,
+    scope: &tracedecay_contracts::ResolvedScope,
+) {
+    let deadline = Instant::now() + SERVING_SEAT_FAILURE_CEILING;
+    loop {
+        registry.clear_pending_wake_for_scope(scope).await;
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        if registry.pending_wake_micros_for_scope(scope).await == Some(0) {
+            return;
+        }
+        assert!(
+            Instant::now() <= deadline,
+            "the pending-wake slot for {:?} never stayed empty",
+            scope.worktree_id
+        );
     }
 }
 
diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs
index 84b7c4b0c6..138157469d 100644
--- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs
+++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs
@@ -24,18 +24,18 @@ use tracedecay_runtime_core::resident_memory::{
 
 use super::{
     ALPHA_LIB_V1, GitFixture, RETAINED_REVISION_0, SERVING_SEAT_FAILURE_CEILING,
-    advance_pointer_to_unseated_successor, application_context, committed_capture_corpus_files,
-    core_search_request, drain_clone_backfill, git, git_stdout, hold_settled_background_admission,
+    advance_pointer_to_unseated_successor, application_context, clear_pending_wake_until_quiet,
+    committed_capture_corpus_files, core_search_request, drain_clone_backfill, git, git_stdout,
     mounted_core_query_worktree, mounted_core_query_worktree_with_one_permit, published,
     query_authority, query_meta, quiesced_background_reconcile_admission,
     replace_scheduler_chunker_revision, replace_scheduler_policy_revision,
     rewrite_active_rust_extractor_revision, rewrite_preserving_stat, scheduler,
-    scheduler_with_policy, served_lexical_texts, test_project_id, wait_for_dashboard_ready,
-    wait_for_event_to_ready, wait_for_generation_change, wait_for_initial_generation,
-    wait_for_live_complete_generation, wait_for_live_complete_generation_by_polling,
-    wait_for_queryable_text_generation, wait_for_queryable_text_generation_change,
-    wait_for_queryable_text_generation_id, wait_for_quiescent_owner_pass, wait_for_settled_owner,
-    wait_until_serving_seat, write,
+    scheduler_with_policy, served_lexical_texts, settled_owner_with_idle_admission,
+    test_project_id, wait_for_dashboard_ready, wait_for_event_to_ready, wait_for_generation_change,
+    wait_for_initial_generation, wait_for_live_complete_generation,
+    wait_for_live_complete_generation_by_polling, wait_for_queryable_text_generation,
+    wait_for_queryable_text_generation_change, wait_for_queryable_text_generation_id,
+    wait_for_quiescent_owner_pass, wait_for_settled_owner, wait_until_serving_seat, write,
 };
 use crate::{
     code_index::{
@@ -2156,7 +2156,10 @@ async fn unchanged_git_watcher_probe_does_not_enqueue_authoritative_capture() {
         .acquire_owned()
         .await
         .expect("hold background reconcile admission");
-    registry.clear_pending_wake_for_scope(&scope).await;
+    // The settle above cannot see a pass tail that has dropped its guard and
+    // not yet stamped its `BusyFollowUp` follow-up, so prove the slot stays
+    // empty before asserting that nothing queued a capture pass.
+    clear_pending_wake_until_quiet(&registry, &scope).await;
     assert_eq!(
         scheduler
             .lock()
@@ -2869,7 +2872,11 @@ async fn ignored_dependency_waits_for_global_admission_before_publication_gate()
     let store = TempDir::new().expect("store root");
     let (registry, _) = mounted_core_query_worktree_with_one_permit(&fixture, &store).await;
     let latest = wait_for_live_complete_generation(&registry, fixture.path()).await;
+    // Draining leaves the busy follow-up wake armed, and every pass it starts
+    // owns the single global admission permit this test needs idle. Settle
+    // that chain and burn the banked permit behind it before sampling.
     drain_clone_backfill(&registry, fixture.path()).await;
+    settled_owner_with_idle_admission(&registry, fixture.path()).await;
     let generation = latest.generation();
     let verified_import = generation
         .imports()
@@ -2900,37 +2907,13 @@ async fn ignored_dependency_waits_for_global_admission_before_publication_gate()
         )
     };
     let global_admission = registry.background_reconcile_admission();
-    // `drain_clone_backfill` drops the permit it held, and the worker may
-    // already be inside the pass that follows. That pass owns the only
-    // permit, so the assertion below is a race unless setup waits for the
-    // worker to release it — and drops this gate if the worker is blocked on
-    // it — before the request under test is admitted.
     registry.clear_pending_wake_for_scope(&scope).await;
-    let idle_deadline = Instant::now() + SERVING_SEAT_FAILURE_CEILING;
-    let publication = loop {
-        while global_admission.available_permits() == 0
-            || registry
-                .reconcile_in_progress_for_test(fixture.path())
-                .await
-        {
-            assert!(
-                Instant::now() <= idle_deadline,
-                "background worker must release global admission before the publication gate is held"
-            );
-            tokio::time::sleep(Duration::from_millis(2)).await;
-        }
-        registry.clear_pending_wake_for_scope(&scope).await;
-        let publication = publication_gate.lock().await;
-        let available = global_admission.available_permits();
-        if available == 1 {
-            break publication;
-        }
-        drop(publication);
-        assert!(
-            Instant::now() <= idle_deadline,
-            "test setup requires an idle global admission"
-        );
-    };
+    let publication = publication_gate.lock().await;
+    assert_eq!(
+        global_admission.available_permits(),
+        1,
+        "test setup requires an idle global admission"
+    );
 
     let request_registry = registry.clone();
     let project_root = fixture.path().to_path_buf();
@@ -3695,7 +3678,10 @@ async fn dashboard_progress_does_not_wait_for_the_scheduler_mutex() {
         .collect::<Vec<_>>();
     let fixture = GitFixture::new(&borrowed);
     let store = TempDir::new().expect("store root");
-    let registry = CodeIndexSchedulerRegistryV1::new(1);
+    // One background permit, so holding it is what parks the owner: the
+    // default bound is the host core count and a single held permit would
+    // leave the other passes free to run under the sample below.
+    let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1);
     registry
         .mount_worktree(
             test_project_id(),
@@ -3706,22 +3692,40 @@ async fn dashboard_progress_does_not_wait_for_the_scheduler_mutex() {
         .expect("mount daemon-owned scheduler");
     wait_for_initial_generation(&registry, fixture.path()).await;
     wait_for_dashboard_ready(&registry, fixture.path()).await;
-    // Clone backfill posts its own wakes after the seat. Drain them before
-    // the progress sample so a successor pass cannot look like this test's
-    // unrelated scheduler-mutex holder.
+    // `refresh_in_flight` is the pass counter *or* the pending-wake slot, and
+    // `wait_for_dashboard_ready` only joins the running pass. The seat no
+    // longer waits for the clone successor, so the mount leaves backfill work
+    // behind, and the wakes that drain it leave a banked permit whose no-op
+    // pass projects Verifying instead of Fresh. Settle the whole mount-era
+    // chain, then hold the admission so no further pass can start under the
+    // sample below.
     drain_clone_backfill(&registry, fixture.path()).await;
+    settled_owner_with_idle_admission(&registry, fixture.path()).await;
+    let _quiet_owner = quiesced_background_reconcile_admission(&registry, fixture.path()).await;
     let canonical_root = fixture
         .path()
         .canonicalize()
         .expect("canonical fixture root");
-    let (scheduler, progress_slot) = {
+    let (scheduler, progress_slot, scope) = {
         let mounted = registry.mounted.lock().await;
         let worktree = mounted.get(&canonical_root).expect("mounted worktree");
         (
             Arc::clone(&worktree.scheduler),
             Arc::clone(&worktree.build_progress),
+            tracedecay_contracts::ResolvedScope::new(
+                test_project_id(),
+                worktree.repository_id.clone(),
+                worktree.worktree_id.clone(),
+                None,
+            )
+            .expect("resolved scope"),
         )
     };
+    // `refresh_in_flight` also reads the pending-wake slot, and the settled
+    // owner's pass tail can still stamp `BusyFollowUp` into it after every
+    // settle check above (CI run 35412193695). With the admission held that
+    // tail is finite: empty the slot until it stays empty.
+    clear_pending_wake_until_quiet(&registry, &scope).await;
     let expected = tokio::time::timeout(Duration::from_secs(5), async {
         loop {
             if let Some(progress) = progress_slot.read().expect("progress slot").snapshot() {
@@ -3732,10 +3736,6 @@ async fn dashboard_progress_does_not_wait_for_the_scheduler_mutex() {
     })
     .await
     .expect("background text build publishes progress");
-    // Text seating keeps `reconcile_in_progress` after it releases the
-    // scheduler mutex. Sampling in that window reports Verifying for a real
-    // pass, which is not this test's unrelated holder.
-    let _admission = hold_settled_background_admission(&registry, fixture.path()).await;
 
     let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
     let (release_tx, release_rx) = tokio::sync::oneshot::channel();
@@ -5236,11 +5236,13 @@ async fn concurrent_query_admissions_claim_one_pending_wake_before_worker_coales
     // about simultaneous query admissions, so finish that independent
     // production journey before establishing the empty-slot precondition.
     drain_clone_backfill(&registry, fixture.path()).await;
-    let admission = registry
-        .background_reconcile_admission()
-        .acquire_owned()
-        .await
-        .expect("background reconcile admission");
+    // Take the shared admission first, through the helper that also waits out
+    // an in-flight pass: from here no new pass can start, so the quiet window
+    // established below stays quiet. A raw `acquire_owned` returns the instant
+    // a running pass releases the admission mid-body, and that pass's tail then
+    // stamps `BusyFollowUp` over the empty slot this test set up, which makes
+    // every claim below decline.
+    let admission = quiesced_background_reconcile_admission(&registry, fixture.path()).await;
     let scheduler = {
         let mounted = registry.mounted.lock().await;
         Arc::clone(
@@ -5250,8 +5252,11 @@ async fn concurrent_query_admissions_claim_one_pending_wake_before_worker_coales
                 .scheduler,
         )
     };
+    // The tail of the pass the admission was taken from also publishes text
+    // owners, so empty the wake slot and prove it stays empty before clearing
+    // the generations this test needs absent.
+    clear_pending_wake_until_quiet(&registry, &scope).await;
     registry.clear_serving_generation_for_scope(&scope).await;
-    registry.clear_pending_wake_for_scope(&scope).await;
     let held = scheduler
         .lock()
         .expect("hold the scheduler as a rebuild would");
@@ -5476,7 +5481,10 @@ async fn foreign_wake_keeps_pending_arrival_when_query_claim_is_released() {
     let store = TempDir::new().expect("store root");
     let (registry, scope) = mounted_core_query_worktree_with_one_permit(&fixture, &store).await;
     let admission = quiesced_background_reconcile_admission(&registry, fixture.path()).await;
-    registry.clear_pending_wake_for_scope(&scope).await;
+    // A `BusyFollowUp` stamp from the finishing pass's tail would make the
+    // request below decline before it ever reaches the claim gate, and
+    // `wait_for_query_claim` would then hang instead of failing.
+    clear_pending_wake_until_quiet(&registry, &scope).await;
     registry.install_query_claim_gate(&scope);
 
     let request = {
@@ -5526,7 +5534,9 @@ async fn foreign_wake_arriving_during_query_claim_drop_is_retained() {
     // reaches the claim gate under test; settle the backfill first.
     drain_clone_backfill(&registry, fixture.path()).await;
     let admission = quiesced_background_reconcile_admission(&registry, fixture.path()).await;
-    registry.clear_pending_wake_for_scope(&scope).await;
+    // Same hang: a tail's `BusyFollowUp` stamp declines the request before the
+    // claim gate this test waits on.
+    clear_pending_wake_until_quiet(&registry, &scope).await;
     registry.install_query_claim_gate(&scope);
     registry.install_pending_wake_drop_gate(&scope).await;
 
@@ -6954,7 +6964,9 @@ async fn text_freshness_query_during_owner_work_is_current_when_source_is_unchan
         .acquire_owned()
         .await
         .expect("hold background reconcile admission");
-    registry.clear_pending_wake_for_scope(&scope).await;
+    // A pass tail that stamps `BusyFollowUp` after this clear would fail the
+    // "no wake" assertion below, so prove the empty slot holds.
+    clear_pending_wake_until_quiet(&registry, &scope).await;
     // Stand in for a worker pass re-observing an unchanged tree: in-progress,
     // scheduler mutex free, nothing moved on disk or in git.
     let owner_pass = registry
@@ -9142,7 +9154,8 @@ async fn graph_off_remount_preserves_an_unhinted_source_reconcile() {
         .acquire_owned()
         .await
         .expect("hold worker after remount dequeue");
-    registry.clear_pending_wake_for_scope(&scope).await;
+    // Same tail: its stamp would look like the unhinted edit's own wake below.
+    clear_pending_wake_until_quiet(&registry, &scope).await;
     fixture.edit("src/lib.rs", "pub fn beta() -> usize { 2 }\n");
     git(fixture.path(), &["commit", "-qam", "unhinted remount edit"]);
 
diff --git a/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs b/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs
index c07cdca96a..89c3b65fb3 100644
--- a/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs
+++ b/crates/tracedecay-code-index/tests/code_index_suite/production_orchestration.rs
@@ -4548,19 +4548,42 @@ fn rss_scaled_request(file_count: usize) -> CodeIndexBuildRequestV1 {
     }
 }
 
-/// Decode `manifest`, serving every segment read from `segments`, and return
-/// the peak RSS growth over a freshly reset high water mark.
-fn rss_measure_decode(label: &str, manifest: &[u8], segments: &BTreeMap<String, Vec<u8>>) -> u64 {
+/// What one decode of a generation observed: the shape of the segment reads
+/// the restore issued, and the peak RSS it grew over a freshly reset high
+/// water mark.
+struct RssDecodeProbeV1 {
+    hwm_delta_kib: u64,
+    evidence_reads: usize,
+    evidence_read_whole: bool,
+    largest_evidence_read: u64,
+    largest_evidence_buffer: usize,
+}
+
+/// Decode `manifest`, serving every segment read from `segments`, and report
+/// both the read shape and the peak RSS growth.
+fn rss_measure_decode(
+    label: &str,
+    manifest: &[u8],
+    segments: &BTreeMap<String, Vec<u8>>,
+    evidence_digest: &str,
+) -> RssDecodeProbeV1 {
     assert!(rss_reset_peak(), "reset VmHWM");
     let hwm_before = rss_proc_kib("VmHWM").expect("VmHWM");
     let mut whole_reads = 0_usize;
     let mut ranged_reads = 0_usize;
+    let mut probe = RssDecodeProbeV1 {
+        hwm_delta_kib: 0,
+        evidence_reads: 0,
+        evidence_read_whole: false,
+        largest_evidence_read: 0,
+        largest_evidence_buffer: 0,
+    };
     let restored =
         CodeIndexPublishedGenerationV1::decode_partitioned_sealed(manifest, |request, buffer| {
-            let (digest, offset, length) = match request {
+            let (digest, offset, length, whole) = match request {
                 SealedGenerationSegmentReadV1::Whole { digest, size_bytes } => {
                     whole_reads += 1;
-                    (digest, 0, size_bytes)
+                    (digest, 0, size_bytes, true)
                 }
                 SealedGenerationSegmentReadV1::Range {
                     digest,
@@ -4569,9 +4592,15 @@ fn rss_measure_decode(label: &str, manifest: &[u8], segments: &BTreeMap<String,
                     ..
                 } => {
                     ranged_reads += 1;
-                    (digest, offset, length)
+                    (digest, offset, length, false)
                 }
             };
+            let evidence = digest.as_str() == evidence_digest;
+            if evidence {
+                probe.evidence_reads += 1;
+                probe.evidence_read_whole |= whole;
+                probe.largest_evidence_read = probe.largest_evidence_read.max(length);
+            }
             let bytes = segments.get(digest.as_str()).ok_or_else(|| {
                 CodeIndexProductionErrorV1::Contract("measured segment is missing".to_owned())
             })?;
@@ -4579,6 +4608,12 @@ fn rss_measure_decode(label: &str, manifest: &[u8], segments: &BTreeMap<String,
             let end = start + usize::try_from(length).expect("segment length");
             buffer.clear();
             buffer.extend_from_slice(&bytes[start..end]);
+            if evidence {
+                // The restore owns this buffer and reuses it across reads, so
+                // its capacity is the segment bytes the restore holds at once.
+                probe.largest_evidence_buffer =
+                    probe.largest_evidence_buffer.max(buffer.capacity());
+            }
             Ok(())
         })
         .expect("measured manifest decodes")
@@ -4586,55 +4621,36 @@ fn rss_measure_decode(label: &str, manifest: &[u8], segments: &BTreeMap<String,
     let hwm_after = rss_proc_kib("VmHWM").expect("VmHWM");
     let file_count = restored.snapshot().files.len();
     drop(restored);
-    let hwm_delta = hwm_after.saturating_sub(hwm_before);
+    probe.hwm_delta_kib = hwm_after.saturating_sub(hwm_before);
     println!(
         "rss_probe form={label} files={file_count} whole_reads={whole_reads} \
-ranged_reads={ranged_reads} hwm_before_kib={hwm_before} hwm_after_kib={hwm_after} \
-hwm_delta_kib={hwm_delta}"
-    );
-    hwm_delta
+ranged_reads={ranged_reads} evidence_reads={} evidence_read_whole={} \
+largest_evidence_read={} largest_evidence_buffer={} hwm_before_kib={hwm_before} \
+hwm_after_kib={hwm_after} hwm_delta_kib={}",
+        probe.evidence_reads,
+        probe.evidence_read_whole,
+        probe.largest_evidence_read,
+        probe.largest_evidence_buffer,
+        probe.hwm_delta_kib,
+    );
+    probe
 }
 
-/// Restoring a pre-paging generation must not materialize its evidence
-/// segment.
-///
-/// The shipped restore read the whole segment, parsed a `serde_json::Value`
-/// from it, rewrote identities in that tree and deserialized the tree again:
-/// peak memory was 2.35x the on-disk generation and grew with the corpus. The
-/// paged restore of the same generation runs first here as the control: both
-/// forms pay the restored generation's own memory, so the legacy restore's
-/// peak growth beyond the control is the extra cost of the pre-paging path
-/// alone, and it must stay far below the evidence segment rather than
-/// scaling with it.
-#[test]
-fn legacy_generation_restore_does_not_materialize_its_evidence_segment() {
-    const RSS_CHILD: &str = "TD_LEGACY_RSS_CHILD";
-    const RSS_TEST: &str = concat!(
-        "production_orchestration::",
-        "legacy_generation_restore_does_not_materialize_its_evidence_segment"
-    );
-
-    // VmHWM and `clear_refs` are Linux-only; elsewhere there is nothing to read.
-    if rss_proc_kib("VmHWM").is_none() || !rss_reset_peak() {
-        return;
-    }
-    // VmHWM is process-wide, so the reading only means anything while nothing
-    // else is allocating: take it in a child that runs this test alone.
-    if std::env::var_os(RSS_CHILD).is_none() {
-        let status = std::process::Command::new(std::env::current_exe().expect("test binary"))
-            .args([RSS_TEST, "--exact", "--nocapture", "--test-threads=1"])
-            .env(RSS_CHILD, "1")
-            .status()
-            .expect("run the peak-RSS measurement alone");
-        assert!(status.success(), "isolated peak-RSS measurement failed");
-        return;
-    }
-
-    let file_count: usize = std::env::var("TD_LEGACY_RSS_FILES")
-        .ok()
-        .and_then(|value| value.parse().ok())
-        .unwrap_or(300);
+/// The same generation encoded both ways, with the segments both forms read.
+struct LegacyRssFixtureV1 {
+    paged_manifest: Vec<u8>,
+    legacy_manifest: Vec<u8>,
+    segments: BTreeMap<String, Vec<u8>>,
+    evidence_digest: String,
+    evidence_bytes: usize,
+    generation_bytes: usize,
+    file_count: usize,
+}
 
+/// Publish one generation, then rewrite its descriptor into the pre-paging
+/// shape a historical writer emitted: one whole authenticated evidence
+/// segment, no page table.
+fn legacy_rss_fixture(file_count: usize) -> LegacyRssFixtureV1 {
     let store = SharedPublicationStore::default();
     let mut owner = CodeIndexProductionOwnerV1::new(config(), store, ApplyingProjectionSink)
         .expect("rss fixture owner");
@@ -4669,8 +4685,6 @@ fn legacy_generation_restore_does_not_materialize_its_evidence_segment() {
         })
         .expect("rss fixture encodes");
 
-    // Rewrite the descriptor into the pre-paging shape a historical writer
-    // emitted: one whole authenticated evidence segment, no page table.
     let mut envelope: serde_json::Value =
         serde_json::from_slice(&paged_manifest).expect("manifest JSON");
     envelope["generation"]["generation_evidence"]
@@ -4696,13 +4710,160 @@ fn legacy_generation_restore_does_not_materialize_its_evidence_segment() {
     drop(generation);
     drop(owner);
 
+    LegacyRssFixtureV1 {
+        paged_manifest,
+        legacy_manifest,
+        segments,
+        evidence_digest,
+        evidence_bytes,
+        generation_bytes,
+        file_count,
+    }
+}
+
+/// Restoring a pre-paging generation must not materialize its evidence
+/// segment.
+///
+/// The shipped restore read the whole segment, parsed a `serde_json::Value`
+/// from it, rewrote identities in that tree and deserialized the tree again:
+/// peak memory was 2.35x the on-disk generation and grew with the corpus.
+///
+/// The proof is the read shape the restore asks the caller for, not its
+/// memory footprint. A restore that materializes the segment has to hold it,
+/// so it must ask for the whole segment in one read or for a range that grows
+/// with the segment - the restore states its own peak segment residency in the
+/// requests it issues. The paged form of the identical generation is the
+/// reference: the pre-paging form must ask for the same bounded ranges into
+/// the same bounded buffer. That is exact, needs no memory reading, and holds
+/// on every platform.
+///
+/// Peak RSS follows only as a loose ceiling on the whole restore, and it is
+/// deliberately not the proof. VmHWM cannot see an allocation that fits inside
+/// the heap the restored generation already made resident, so it does not
+/// detect retention on its own: injecting a copy of every evidence page into a
+/// buffer held for the decode moves it by a fifth of the segment or less, well
+/// inside the clean band. What it does see is additive noise - a trimmed
+/// allocator makes the next probe fault fresh pages, which on a loaded runner
+/// cost more than the segment under test - so each form is measured over
+/// several alternating rounds and the smallest growth is taken as its cost.
+/// A restore that materializes pays that cost on every round, so the minimum
+/// keeps whatever signal RSS carries and drops the noise that made a single
+/// pair of probes flaky.
+#[test]
+fn legacy_generation_restore_does_not_materialize_its_evidence_segment() {
+    const RSS_CHILD: &str = "TD_LEGACY_RSS_CHILD";
+    const RSS_TEST: &str = concat!(
+        "production_orchestration::",
+        "legacy_generation_restore_does_not_materialize_its_evidence_segment"
+    );
+    /// Alternating paged/legacy rounds behind the discarded warm-up.
+    const RSS_ROUNDS: usize = 4;
+
+    // VmHWM and `clear_refs` are Linux-only. The read-shape guard needs
+    // neither, so run it alone elsewhere rather than skipping the test.
+    let rss_readable = rss_proc_kib("VmHWM").is_some() && rss_reset_peak();
+
+    // VmHWM is process-wide, so the reading only means anything while nothing
+    // else is allocating: take it in a child that runs this test alone.
+    if rss_readable && std::env::var_os(RSS_CHILD).is_none() {
+        let status = std::process::Command::new(std::env::current_exe().expect("test binary"))
+            .args([RSS_TEST, "--exact", "--nocapture", "--test-threads=1"])
+            .env(RSS_CHILD, "1")
+            .status()
+            .expect("run the peak-RSS measurement alone");
+        assert!(
+            status.success(),
+            "the isolated restore measurement failed; its own failure is above"
+        );
+        return;
+    }
+
+    let file_count: usize = std::env::var("TD_LEGACY_RSS_FILES")
+        .ok()
+        .and_then(|value| value.parse().ok())
+        .unwrap_or(300);
+    let fixture = legacy_rss_fixture(file_count);
+
+    let paged = rss_measure_decode(
+        "paged",
+        &fixture.paged_manifest,
+        &fixture.segments,
+        &fixture.evidence_digest,
+    );
+    let legacy = rss_measure_decode(
+        "legacy",
+        &fixture.legacy_manifest,
+        &fixture.segments,
+        &fixture.evidence_digest,
+    );
+
+    // --- Guard 1: the read shape, exact. ------------------------------------
+    assert!(
+        !paged.evidence_read_whole && paged.evidence_reads > 1,
+        "the paged control must itself page the evidence segment: \
+         evidence_reads={} evidence_read_whole={}",
+        paged.evidence_reads,
+        paged.evidence_read_whole
+    );
+    assert!(
+        !legacy.evidence_read_whole,
+        "restoring a pre-paging generation asked for its whole \
+         {}-byte evidence segment in one read: the segment is being materialized",
+        fixture.evidence_bytes
+    );
+    assert!(
+        legacy.largest_evidence_read <= paged.largest_evidence_read,
+        "restoring a pre-paging generation read up to {} bytes of its \
+         {}-byte evidence segment at once, beyond the {}-byte bound the paged \
+         restore of the same generation holds to",
+        legacy.largest_evidence_read,
+        fixture.evidence_bytes,
+        paged.largest_evidence_read
+    );
+    assert!(
+        legacy.largest_evidence_buffer <= paged.largest_evidence_buffer,
+        "restoring a pre-paging generation held a {}-byte evidence buffer, \
+         beyond the {}-byte buffer the paged restore of the same generation \
+         holds to: the segment is being materialized",
+        legacy.largest_evidence_buffer,
+        paged.largest_evidence_buffer
+    );
+    assert_eq!(
+        legacy.evidence_reads, paged.evidence_reads,
+        "the pre-paging restore must read the evidence segment in the same \
+         bounded chunks the page table would have named"
+    );
+
+    // --- Guard 2: peak RSS, noise-tolerant. ---------------------------------
+    if !rss_readable {
+        return;
+    }
     // The first restore in a process pays a cold-start cost (the arena a
     // restored generation needs) that has nothing to do with the form being
-    // restored. Spend it on a discarded probe so the two measured probes
-    // start from the same allocator state and stay comparable.
-    rss_measure_decode("warmup", &paged_manifest, &segments);
-    let paged_hwm = rss_measure_decode("paged", &paged_manifest, &segments);
-    let legacy_hwm = rss_measure_decode("legacy", &legacy_manifest, &segments);
+    // restored; the two probes above spent it. Alternate from here so neither
+    // form is systematically the one that inherits a trimmed allocator.
+    let mut paged_hwm = u64::MAX;
+    let mut legacy_hwm = u64::MAX;
+    for _ in 0..RSS_ROUNDS {
+        paged_hwm = paged_hwm.min(
+            rss_measure_decode(
+                "paged",
+                &fixture.paged_manifest,
+                &fixture.segments,
+                &fixture.evidence_digest,
+            )
+            .hwm_delta_kib,
+        );
+        legacy_hwm = legacy_hwm.min(
+            rss_measure_decode(
+                "legacy",
+                &fixture.legacy_manifest,
+                &fixture.segments,
+                &fixture.evidence_digest,
+            )
+            .hwm_delta_kib,
+        );
+    }
     // The paged decode of the same generation is the control, not a warm-up:
     // both forms restore the identical generation, so only the difference is
     // the pre-paging path's own cost. An absolute peak is not a usable
@@ -4712,23 +4873,28 @@ fn legacy_generation_restore_does_not_materialize_its_evidence_segment() {
     let legacy_extra_bytes = legacy_hwm.saturating_sub(paged_hwm) * 1024;
 
     println!(
-        "rss_summary files={file_count} generation_on_disk_bytes={generation_bytes} \
-evidence_segment_bytes={evidence_bytes} paged_hwm_delta_kib={paged_hwm} \
+        "rss_summary files={} generation_on_disk_bytes={} \
+evidence_segment_bytes={} rounds={RSS_ROUNDS} paged_hwm_delta_kib={paged_hwm} \
 legacy_hwm_delta_kib={legacy_hwm} legacy_extra_bytes={legacy_extra_bytes} \
 legacy_extra_over_evidence={:.3}",
-        legacy_extra_bytes as f64 / evidence_bytes as f64,
-    );
-
-    // A restore that materializes the segment holds all of it at once and
-    // parses it on top, so it costs at least the segment; the streaming
-    // restore costs one bounded page buffer plus allocator slack, measured
-    // at a fifth to a third of the segment. The bound sits between them, and
-    // is tighter than the half-the-whole-generation bound it replaces (which
-    // permitted two and a half times this segment).
+        fixture.file_count,
+        fixture.generation_bytes,
+        fixture.evidence_bytes,
+        legacy_extra_bytes as f64 / fixture.evidence_bytes as f64,
+    );
+
+    // The read-shape guard above already refused a restore that holds the
+    // segment. This is the ceiling on the rest of the restore: the pre-paging
+    // path must not cost a segment's worth of anything over the paged path.
+    // The bound is the segment because that is the size the guard is about,
+    // not a threshold tuned to the noise - the minimum over the rounds is
+    // what removes the noise.
     assert!(
-        legacy_extra_bytes < evidence_bytes as u64,
+        legacy_extra_bytes < fixture.evidence_bytes as u64,
         "restoring a pre-paging generation cost {legacy_extra_bytes} bytes of peak RSS beyond the \
-         paged restore of the same {generation_bytes}-byte generation, which is not far below its \
-         {evidence_bytes}-byte evidence segment: the segment is being materialized"
+         paged restore of the same {}-byte generation, which is not far below its \
+         {}-byte evidence segment: the segment is being materialized",
+        fixture.generation_bytes,
+        fixture.evidence_bytes
     );
 }
diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs
index c0a7668ffb..846387e8ca 100644
--- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs
+++ b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest.rs
@@ -119,9 +119,12 @@ async fn admit_codex_project_rollouts(
     project_id: ProjectId,
     max_new_bytes: Option<u64>,
     cancellation: &ObservationCancellation,
-) -> Result<bool> {
+) -> Result<CodexRolloutAdmission> {
     let mut budget = max_new_bytes;
     let mut deferred = false;
+    let mut observations_committed = 0_u64;
+    let mut scanned = 0_u64;
+    let mut replayed = 0_u64;
     let mut paths = source.transcript_paths(project_root).into_iter().peekable();
     while let Some(path) = paths.next() {
         let progress =
@@ -136,6 +139,11 @@ async fn admit_codex_project_rollouts(
             .await
             .map_err(|error| map_transcript_ingest_error(&error))?;
         deferred |= progress.source_deferred;
+        observations_committed = observations_committed.saturating_add(progress.frames_persisted);
+        scanned = scanned.saturating_add(1);
+        if progress.resumed && progress.frames_persisted == 0 {
+            replayed = replayed.saturating_add(1);
+        }
         if let Some(remaining) = budget.as_mut() {
             *remaining = remaining.saturating_sub(progress.bytes_consumed);
             if *remaining == 0 {
@@ -144,7 +152,27 @@ async fn admit_codex_project_rollouts(
             }
         }
     }
-    Ok(deferred)
+    Ok(CodexRolloutAdmission {
+        deferred,
+        // Only when every scanned rollout was a pure replay. One rollout with
+        // new frames makes the pass a commit, not a duplicate, and a pass that
+        // scanned nothing has nothing to call durable.
+        exact_duplicate: scanned > 0 && scanned == replayed,
+        observations_committed,
+    })
+}
+
+/// What one Codex rollout admission pass committed, apart from what its own
+/// projection drain later catches. The projection queue is shared per scope
+/// with the project catch-up sweep, which can consume these rows first.
+pub(super) struct CodexRolloutAdmission {
+    pub(super) deferred: bool,
+    /// Every scanned rollout resumed at its stored cursor with nothing new to
+    /// persist, so the transcript was already durable when this pass ran.
+    /// `JsonlObservationAdmissionProgress::resumed` is what separates that from
+    /// an empty source, so a first-ever scan is never called a duplicate.
+    pub(super) exact_duplicate: bool,
+    pub(super) observations_committed: u64,
 }
 
 async fn drain_host_observation_projections(
@@ -642,18 +670,32 @@ pub async fn ingest_transcript_with_cancellation(
         source_deferred,
         lcm_receipt,
         route_admission,
+        observations_committed: route_observations_committed,
+        exact_duplicate: route_exact_duplicate,
     } = capture;
+    // Admission is the durable commit; projection is downstream materialization
+    // off a queue this scope shares with the project catch-up sweep. Counting
+    // only the projections this pass drained itself reports a pass whose rows a
+    // peer drainer took as though it had captured nothing.
     let authority_changed = messages_upserted > 0
+        || route_observations_committed > 0
         || snapshot_capture
             .as_ref()
             .is_some_and(|capture| capture.stats.messages_upserted > 0)
         || claude_observation_stats
             .as_ref()
             .is_some_and(|stats| stats.observations_committed > 0 || stats.cursor_advances > 0);
+    // A pass that changed nothing is only `accepted_for_replay` when it cannot
+    // prove the data is already there. Routes that can prove it say so: Claude
+    // through its duplicate counters, every other route through
+    // `exact_duplicate`. Without this a replay whose observations a peer
+    // drainer already projected reports a terminal, non-retryable status that
+    // neither proves a commit nor invites a retry.
     let exact_duplicate = !authority_changed
-        && claude_observation_stats
-            .as_ref()
-            .is_some_and(|stats| stats.observation_duplicates > 0 || stats.cursor_duplicates > 0);
+        && (route_exact_duplicate
+            || claude_observation_stats.as_ref().is_some_and(|stats| {
+                stats.observation_duplicates > 0 || stats.cursor_duplicates > 0
+            }));
     let deferred_by_byte_cap = source_deferred
         || snapshot_capture
             .as_ref()
@@ -710,6 +752,12 @@ pub async fn ingest_transcript_with_cancellation(
         .await;
         output["hint_outcomes"] = settlement.as_json();
     }
+    // Routes that admit observations directly report what they committed, so a
+    // `messages_upserted: 0` pass is readable without guessing which drainer
+    // won. The snapshot and Claude blocks below own the key for their routes.
+    if route_observations_committed > 0 {
+        output["observations_committed"] = json!(route_observations_committed);
+    }
     if let Some(capture) = snapshot_capture {
         output["observations_committed"] = json!(capture.stats.messages_upserted);
         output["bytes_consumed"] = json!(capture.bytes_consumed);
diff --git a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs
index dbfda00811..fbf97ef025 100644
--- a/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs
+++ b/crates/tracedecay-mcp/src/handlers/hook_runtime/ingest/kernels.rs
@@ -121,6 +121,14 @@ pub(super) struct TranscriptCaptureOutcome {
     pub(super) snapshot: Option<SnapshotCaptureOutcome>,
     pub(super) claude_observation: Option<ClaudeObservationIngestStats>,
     pub(super) source_deferred: bool,
+    /// Observations the route durably admitted, whoever later projects them.
+    /// `messages_upserted` counts only the projections this pass drained
+    /// itself, which a peer drainer can legitimately take first.
+    pub(super) observations_committed: u64,
+    /// The route committed nothing because its observations were already
+    /// durable. Kept apart from `messages_upserted == 0`, which cannot tell an
+    /// already-committed replay from a pass that captured nothing.
+    pub(super) exact_duplicate: bool,
     /// Set by routes that commit through the LCM authority instead of a source
     /// scan; rendered as `authority_outcome` and `committed_state`.
     pub(super) lcm_receipt: Option<LcmAuthorityResponse>,
@@ -405,7 +413,7 @@ async fn capture_codex_project(
     let scope = ObservationScopeV1::Project {
         project_id: project_id.clone(),
     };
-    let source_deferred = admit_codex_project_rollouts(
+    let admitted = admit_codex_project_rollouts(
         ctx.facade,
         &source,
         cg.project_root(),
@@ -418,7 +426,9 @@ async fn capture_codex_project(
         drain_host_observation_projections(ctx.facade, &scope, ctx.cancellation).await?;
     Ok(TranscriptCaptureOutcome {
         messages_upserted,
-        source_deferred,
+        source_deferred: admitted.deferred,
+        observations_committed: admitted.observations_committed,
+        exact_duplicate: admitted.exact_duplicate,
         ..TranscriptCaptureOutcome::default()
     })
 }
@@ -445,6 +455,8 @@ fn cursor_capture_outcome(
     TranscriptCaptureOutcome {
         messages_upserted: stats.messages_upserted,
         source_deferred: stats.source_deferred,
+        observations_committed: stats.observations_committed,
+        exact_duplicate: stats.exact_duplicate,
         ..TranscriptCaptureOutcome::default()
     }
 }
diff --git a/crates/tracedecay-session-runtime/src/session_sync.rs b/crates/tracedecay-session-runtime/src/session_sync.rs
index adfb143bc0..57125baf99 100644
--- a/crates/tracedecay-session-runtime/src/session_sync.rs
+++ b/crates/tracedecay-session-runtime/src/session_sync.rs
@@ -638,7 +638,6 @@ impl DaemonSessionSyncService {
     async fn await_import_history(
         &self,
         context: &SessionSyncProjectContext,
-        project_sessions: &RegisteredGlobalDbLeaseV1,
         request: &SessionSyncRequestV1,
     ) -> Result<
         crate::session_temporal_refresh_scheduler::history::SessionHistoricalIngestProgress,
@@ -673,22 +672,14 @@ impl DaemonSessionSyncService {
                 return Err(Some(interruption));
             }
         };
-        if let (Some(project), Some(user)) = settled
-            && matches!(
-                context.project_refresh.serving_status().state,
-                SessionProjectionServingState::Current
-            )
-            && matches!(
-                context.user_refresh.serving_status().state,
-                SessionProjectionServingState::Current
-            )
-            && self
-                .projection_store_is_current(project_sessions, request)
-                .await?
-            && self
-                .projection_store_is_current(&context.user_sessions, request)
-                .await?
-        {
+        // Only the historical frontier is decided here. Projection currency is
+        // `await_import_projection`'s gate, which waits for the projection
+        // workers and then re-checks these same serving states and stores. This
+        // gate does not wait for them, so asserting them here reports
+        // `session_history_not_current` for a history that is current and whose
+        // projection has simply not drained yet, and skips the stage that would
+        // have waited for it.
+        if let (Some(project), Some(user)) = settled {
             Ok(
                 crate::session_temporal_refresh_scheduler::history::SessionHistoricalIngestProgress {
                     stats: project.stats.merge(user.stats),
diff --git a/crates/tracedecay-session-runtime/src/session_sync/work.rs b/crates/tracedecay-session-runtime/src/session_sync/work.rs
index 4b2ed150ce..80b21f6e6d 100644
--- a/crates/tracedecay-session-runtime/src/session_sync/work.rs
+++ b/crates/tracedecay-session-runtime/src/session_sync/work.rs
@@ -491,10 +491,7 @@ impl SessionSyncProjectContext {
         request: &SessionSyncRequestV1,
         project_sessions: RegisteredGlobalDbLeaseV1,
     ) -> SessionSyncWorkResult {
-        let history = match service
-            .await_import_history(self, &project_sessions, request)
-            .await
-        {
+        let history = match service.await_import_history(self, request).await {
             Ok(progress) => Some(progress),
             Err(Some(interruption)) => {
                 return SessionSyncWorkResult::Interrupted(interruption);
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs
index 66a8dc6bee..b02eecaab6 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs
@@ -2,7 +2,7 @@
 
 use std::collections::{HashMap, VecDeque};
 use std::path::{Path, PathBuf};
-use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError};
+use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError, Weak};
 
 use tokio::sync::Notify;
 use tokio::sync::futures::OwnedNotified;
@@ -233,6 +233,12 @@ pub struct CodexJsonlAdmissionProgress {
     pub frames_rejected_before_decode: u64,
     pub frames_refused: u64,
     pub frames_persisted: u64,
+    /// This pass resumed from a durable source cursor instead of opening the
+    /// rollout for the first time. With `frames_persisted == 0` it is the only
+    /// evidence that separates an already-admitted rollout from an empty one,
+    /// so a caller can report the replay as a duplicate rather than as a pass
+    /// that captured nothing.
+    pub resumed: bool,
 }
 
 /// Admit a Codex rollout for one exact project identity.
@@ -651,6 +657,44 @@ async fn shared_session_meta_with_provenance(
     }
 }
 
+/// Serializes the read-cursor-then-admit window for one rollout in one scope.
+///
+/// The MCP hook route (`admit_codex_project_rollouts`) and the daemon's project
+/// catch-up sweep (`ingest::project_provider::run_codex`) both reach
+/// [`try_admit_codex_jsonl_observations`] for the same rollout under the same
+/// scope, and both read the source cursor before they write. Interleaved, the
+/// loser reads a cursor the winner has not published yet, re-reads frames the
+/// winner has already committed, and re-submits the same observation ids with
+/// its own independently captured repository provenance. The store's replay
+/// verification refuses that second provenance as `observation repository
+/// provenance collision`, which reaches the host as a retryable
+/// `authority_write_failed` infrastructure error rather than the duplicate it
+/// is. Serialized, the loser reads the advanced cursor, persists nothing, and
+/// reports the `resumed` replay its caller renders as an exact duplicate.
+///
+/// The gate is per process. Cross-process writers still meet at the store's
+/// own transaction, which is what the replay verification is there for.
+type CodexAdmissionGate = Arc<tokio::sync::Mutex<()>>;
+type CodexAdmissionGates =
+    Mutex<HashMap<(ObservationScopeV1, PathBuf), Weak<tokio::sync::Mutex<()>>>>;
+
+static CODEX_ADMISSION_GATES: OnceLock<CodexAdmissionGates> = OnceLock::new();
+
+/// ponytail: linear sweep of live gates per acquisition; keyed eviction if a
+/// scope ever admits enough rollouts at once for the sweep to show up.
+fn codex_admission_gate(scope: &ObservationScopeV1, path: &Path) -> CodexAdmissionGate {
+    let gates = CODEX_ADMISSION_GATES.get_or_init(|| Mutex::new(HashMap::new()));
+    let mut gates = gates.lock().unwrap_or_else(PoisonError::into_inner);
+    gates.retain(|_, gate| gate.strong_count() > 0);
+    let key = (scope.clone(), path.to_path_buf());
+    if let Some(gate) = gates.get(&key).and_then(Weak::upgrade) {
+        return gate;
+    }
+    let gate = CodexAdmissionGate::new(tokio::sync::Mutex::new(()));
+    gates.insert(key, Arc::downgrade(&gate));
+    gate
+}
+
 async fn try_admit_codex_jsonl_observations(
     path: &Path,
     admission_scope: CodexObservationAdmission<'_>,
@@ -686,6 +730,11 @@ async fn try_admit_codex_jsonl_observations(
         cancellation,
     };
     let scope = admission_scope.scope();
+    // Held across the cursor read and the admit below: both are one pass over
+    // this rollout, and a peer that interleaves between them re-submits what
+    // this pass is about to commit.
+    let gate = codex_admission_gate(&scope, path);
+    let _admitting = gate.lock().await;
     if let Some(target) = admission
         .get_source_cursor(&ordinary_source, &scope)
         .await
@@ -895,6 +944,7 @@ async fn admit_codex_jsonl_page(
         frames_rejected_before_decode: progress.frames_rejected_before_decode,
         frames_refused: progress.frames_refused,
         frames_persisted: progress.frames_persisted,
+        resumed: progress.resumed,
     })
 }
 
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs
index bdbd6126fa..195997fa0e 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs
@@ -799,20 +799,32 @@ mod goal_event_tests {
         );
     }
 
+    /// A pass yields on the source the capture window left mid-file, not on
+    /// every source it finishes.
+    ///
+    /// `MAX_CAPTURE_WINDOW` is the cooperative bound that owns a resumable
+    /// cursor: it stops inside one rollout and the next pass resumes at that
+    /// byte offset. Yielding once per *finished* rollout has no such cursor, so
+    /// the discovery frontier can never commit and every later pass re-reads
+    /// every rollout it already exhausted, which is quadratic in the rollouts a
+    /// project has. The profile-scope loop has always been bounded this way.
     #[tokio::test]
-    async fn project_provider_yields_between_dated_rollouts_and_converges_without_loss() {
+    async fn project_provider_yields_on_a_deferred_rollout_and_converges_without_loss() {
         crate::runtime::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority();
         let temp = tempfile::tempdir().unwrap();
         let home = temp.path().canonicalize().unwrap();
         let project = home.join("project");
         std::fs::create_dir_all(&project).unwrap();
+        // The newest rollout alone exceeds one capture window, so pass 0 must
+        // stop inside it; the two behind it fit in the pass that finishes it.
+        let deferring_messages =
+            crate::runtime::jsonl_observation_admission::MAX_CAPTURE_WINDOW + 44;
         let rollouts = [
-            (("2026", "09", "03"), "session-newest"),
-            (("2026", "09", "02"), "session-middle"),
-            (("2026", "09", "01"), "session-oldest"),
+            (("2026", "09", "03"), "session-newest", deferring_messages),
+            (("2026", "09", "02"), "session-middle", 3),
+            (("2026", "09", "01"), "session-oldest", 3),
         ];
-        let messages_per_rollout = 3;
-        for (date, session_id) in rollouts {
+        for (date, session_id, messages_per_rollout) in rollouts {
             let directory = home
                 .join(".codex/sessions")
                 .join(date.0)
@@ -826,7 +838,7 @@ mod goal_event_tests {
             })];
             lines.extend((0..messages_per_rollout).map(|ordinal| {
                 json!({
-                    "timestamp": format!("{}-{}-{}T12:00:0{}.000Z", date.0, date.1, date.2, ordinal + 1),
+                    "timestamp": format!("{}-{}-{}T12:00:01.{:03}Z", date.0, date.1, date.2, ordinal),
                     "type": "event_msg",
                     "payload": {
                         "type": "user_message",
@@ -852,7 +864,22 @@ mod goal_event_tests {
         };
         let admission = MemoryHostAdmission::default();
         let cancellation = ObservationCancellation::default();
-        let failure_ceiling = rollouts.len().saturating_add(1);
+        // Pass 0 stops inside the newest rollout at the capture window; pass 1
+        // resumes it from the stored byte offset and, still inside its byte
+        // budget, finishes the two rollouts behind it.
+        let expected_passes: [(&[&str], usize); 2] = [
+            (
+                &["session-newest"],
+                crate::runtime::jsonl_observation_admission::MAX_CAPTURE_WINDOW,
+            ),
+            (
+                &["session-newest", "session-middle", "session-oldest"],
+                (deferring_messages + 1)
+                    - crate::runtime::jsonl_observation_admission::MAX_CAPTURE_WINDOW
+                    + 8,
+            ),
+        ];
+        let failure_ceiling = expected_passes.len().saturating_add(1);
         let mut completed_after = None;
 
         for pass_index in 0..failure_ceiling {
@@ -877,20 +904,30 @@ mod goal_event_tests {
             let after = admission.observations();
             let admitted_this_pass = &after[before..];
             assert!(
-                pass_index < rollouts.len(),
+                pass_index < expected_passes.len(),
                 "coverage did not complete within the fixture-derived ceiling"
             );
-            let expected_session = rollouts[pass_index].1;
+            let (expected_sessions, expected_admitted) = expected_passes[pass_index];
             assert_eq!(
                 admitted_this_pass.len(),
-                messages_per_rollout + 1,
-                "pass {pass_index} must finish exactly {expected_session}"
+                expected_admitted,
+                "pass {pass_index} must admit one capture window of {expected_sessions:?}"
+            );
+            assert_eq!(
+                admitted_this_pass
+                    .iter()
+                    .map(|stored| {
+                        let envelope: CanonicalObservationEnvelopeV1 =
+                            serde_json::from_value(stored.observation().payload().clone()).unwrap();
+                        envelope.relations().session_id().as_str().to_owned()
+                    })
+                    .collect::<BTreeSet<_>>(),
+                expected_sessions
+                    .iter()
+                    .map(|session_id| (*session_id).to_owned())
+                    .collect::<BTreeSet<_>>(),
+                "pass {pass_index} admitted the wrong rollouts"
             );
-            assert!(admitted_this_pass.iter().all(|stored| {
-                let envelope: CanonicalObservationEnvelopeV1 =
-                    serde_json::from_value(stored.observation().payload().clone()).unwrap();
-                envelope.relations().session_id().as_str() == expected_session
-            }));
 
             let coverage = read_host_provider_coverage(&admission, &scope, "codex")
                 .await
@@ -902,11 +939,14 @@ mod goal_event_tests {
             assert_eq!(coverage, Some(HostProviderCoverage::Partial));
         }
 
-        assert_eq!(completed_after, Some(rollouts.len()));
+        assert_eq!(completed_after, Some(expected_passes.len()));
         let observations = admission.observations();
         assert_eq!(
             observations.len(),
-            rollouts.len() * (messages_per_rollout + 1)
+            rollouts
+                .iter()
+                .map(|(_, _, messages)| messages + 1)
+                .sum::<usize>()
         );
         let envelopes = observations
             .iter()
@@ -925,10 +965,10 @@ mod goal_event_tests {
             admitted_sessions.iter().cloned().collect::<BTreeSet<_>>(),
             rollouts
                 .iter()
-                .map(|(_, session_id)| (*session_id).to_owned())
+                .map(|(_, session_id, _)| (*session_id).to_owned())
                 .collect::<BTreeSet<_>>()
         );
-        for (_, session_id) in rollouts {
+        for (_, session_id, messages_per_rollout) in rollouts {
             assert_eq!(
                 admitted_sessions
                     .iter()
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs
index 4c36450be6..4bc895f9b2 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/cursor.rs
@@ -178,6 +178,46 @@ fn cursor_admission_record_id(
     Ok((identity.into_primary(), retry_eligible))
 }
 
+/// What one Cursor ingest pass did to the sources it scanned, independently of
+/// what its own projection drain happened to catch.
+///
+/// The projection queue is shared per scope and consumed on projection, so the
+/// project catch-up sweep in [`crate::runtime::ingest::project_provider`] can
+/// drain the rows this pass just admitted before this pass drains them itself.
+/// Projection-output counts alone therefore cannot answer whether the pass's
+/// transcript is durable, and both of the states below report zero outputs:
+///
+/// - `observations_committed > 0`: this pass persisted new observations. They
+///   are durable at admission; which drainer materializes them is not the
+///   host's question.
+/// - `fully_replayed()`: every scanned source resumed at its stored cursor
+///   with nothing new to persist, so its observations were already durable.
+#[derive(Debug, Default, Clone, Copy)]
+struct CursorSourceAdmissionTally {
+    scanned: u64,
+    replayed: u64,
+    observations_committed: u64,
+}
+
+impl CursorSourceAdmissionTally {
+    fn record(&mut self, progress: &JsonlObservationAdmissionProgress) {
+        self.scanned = self.scanned.saturating_add(1);
+        self.observations_committed = self
+            .observations_committed
+            .saturating_add(progress.frames_persisted);
+        if progress.resumed && progress.frames_persisted == 0 {
+            self.replayed = self.replayed.saturating_add(1);
+        }
+    }
+
+    /// True only when at least one source was scanned and every one of them
+    /// was a pure replay. One source with new frames makes the pass a commit,
+    /// not a duplicate.
+    const fn fully_replayed(self) -> bool {
+        self.scanned > 0 && self.scanned == self.replayed
+    }
+}
+
 // Cursor JSONL admission chokepoint: the whole per-file admission future is
 // boxed here so the per-file sweep loop no longer pins each call, keeping the
 // debug poll frame bounded through the deep ingest recursion chain.
@@ -575,6 +615,7 @@ pub async fn try_ingest_cursor_transcript_event_capped_with_admission(
         "sessions.hosts.cursor.discover_blocking",
         run_blocking_transcript_section(|| source.transcript_paths(&project_root))
     );
+    let mut admitted = CursorSourceAdmissionTally::default();
     for path in paths {
         let context = cursor_observation_context(&source.event, &path, false);
         let progress = admit_cursor_jsonl_observations(
@@ -587,6 +628,7 @@ pub async fn try_ingest_cursor_transcript_event_capped_with_admission(
             &ObservationCancellation::default(),
         )
         .await?;
+        admitted.record(&progress);
         budget.record_progress(progress.bytes_consumed, progress.source_deferred);
     }
     let mut stats = drain_cursor_observation_projections(
@@ -597,6 +639,10 @@ pub async fn try_ingest_cursor_transcript_event_capped_with_admission(
     .await?;
     stats.bytes_consumed = budget.consumed();
     stats.source_deferred |= budget.deferred();
+    stats.observations_committed = admitted.observations_committed;
+    stats.exact_duplicate |= stats.messages_upserted == 0
+        && stats.observations_committed == 0
+        && admitted.fully_replayed();
     Ok(stats)
 }
 
@@ -719,6 +765,7 @@ pub async fn try_ingest_cursor_user_transcript_event_capped_with_admission(
         "sessions.hosts.cursor.discover_blocking",
         run_blocking_transcript_section(|| source.transcript_paths(&placeholder))
     );
+    let mut admitted = CursorSourceAdmissionTally::default();
     for path in paths {
         let context = cursor_observation_context(&source.event, &path, true);
         let progress = admit_cursor_jsonl_observations(
@@ -731,6 +778,7 @@ pub async fn try_ingest_cursor_user_transcript_event_capped_with_admission(
             &ObservationCancellation::default(),
         )
         .await?;
+        admitted.record(&progress);
         budget.record_progress(progress.bytes_consumed, progress.source_deferred);
     }
     let mut stats = drain_cursor_observation_projections(
@@ -741,6 +789,10 @@ pub async fn try_ingest_cursor_user_transcript_event_capped_with_admission(
     .await?;
     stats.bytes_consumed = budget.consumed();
     stats.source_deferred |= budget.deferred();
+    stats.observations_committed = admitted.observations_committed;
+    stats.exact_duplicate |= stats.messages_upserted == 0
+        && stats.observations_committed == 0
+        && admitted.fully_replayed();
     Ok(stats)
 }
 
@@ -820,6 +872,7 @@ async fn admit_cursor_sweep_observations_with_session_ids(
         "sessions.hosts.cursor.discover_blocking",
         run_blocking_transcript_section(|| source.transcript_paths(project_root))
     );
+    let mut admitted = CursorSourceAdmissionTally::default();
     for path in paths {
         if cancellation.is_cancelled() {
             return Err(TranscriptIngestError::Cancelled { provider: "cursor" });
@@ -847,18 +900,20 @@ async fn admit_cursor_sweep_observations_with_session_ids(
             cancellation,
         )
         .await?;
+        admitted.record(&progress);
         budget.record_progress(progress.bytes_consumed, progress.source_deferred);
     }
     if cancellation.is_cancelled() {
         return Err(TranscriptIngestError::Cancelled { provider: "cursor" });
     }
-    let outcome = projection::drain_cursor_observation_projections_with_sessions(
+    let mut outcome = projection::drain_cursor_observation_projections_with_sessions(
         admission,
         &scope,
         cancellation,
     )
     .await
     .map(|stats| stats.into_sweep_outcome(budget.consumed(), budget.deferred()))?;
+    outcome.stats.observations_committed = admitted.observations_committed;
     persist_host_provider_coverage(
         admission,
         &scope,
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor/projection.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor/projection.rs
index 79f1ae795f..099fa77bd8 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/cursor/projection.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/cursor/projection.rs
@@ -19,6 +19,18 @@ pub struct CursorTranscriptIngestStats {
     pub messages_upserted: u64,
     pub bytes_consumed: u64,
     pub source_deferred: bool,
+    /// Observations this pass durably admitted. `messages_upserted` counts
+    /// only what this pass's own projection drain materialized, and the
+    /// projection queue is shared per scope, so a peer drainer can consume
+    /// these rows first. Admission is the commit; keep it accounted for
+    /// separately from whoever projects it.
+    pub observations_committed: u64,
+    /// This pass changed nothing because its observations were already
+    /// durable: every scanned source resumed at its stored cursor with no new
+    /// frame to persist, or the projection drain met only exact duplicates.
+    /// Distinguishes an already-committed replay from a pass that committed
+    /// nothing at all; both report `messages_upserted == 0`.
+    pub exact_duplicate: bool,
 }
 
 #[derive(Debug, Default)]
@@ -102,6 +114,7 @@ pub async fn try_ingest_cursor_user_sweep_capped_with_admission<S: BuildHasher>(
 pub(in crate::runtime) struct CursorProjectionDrainStats {
     pub session_ids: Vec<String>,
     pub messages_upserted: u64,
+    pub exact_duplicates: u64,
     pub source_deferred: bool,
 }
 
@@ -141,6 +154,7 @@ pub(in crate::runtime) async fn drain_cursor_observation_projections_with_sessio
     Ok(CursorProjectionDrainStats {
         session_ids: outcome.session_ids,
         messages_upserted: outcome.projected_outputs,
+        exact_duplicates: outcome.exact_duplicates,
         source_deferred: outcome.deferred,
     })
 }
@@ -152,6 +166,8 @@ impl CursorProjectionDrainStats {
             messages_upserted: self.messages_upserted,
             bytes_consumed: 0,
             source_deferred: self.source_deferred,
+            observations_committed: 0,
+            exact_duplicate: self.messages_upserted == 0 && self.exact_duplicates > 0,
         }
     }
 
diff --git a/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs
index 6d2c90c07f..b548c3bc95 100644
--- a/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs
+++ b/crates/tracedecay-sessions/src/runtime/hosts/cursor/tests.rs
@@ -354,3 +354,189 @@ fn user_scope_selects_one_physical_authority_for_a_mirrored_session() {
         Some("session-mirrored")
     );
 }
+
+/// Fixture for the replayed-ingest journey: one project-scoped Cursor hook
+/// event whose transcript already carries two records.
+fn cursor_replay_fixture() -> (tempfile::TempDir, String, ProjectId) {
+    // Production installs the process-wide capture authorities during daemon
+    // bootstrap; capture refuses with a typed `BackgroundResourceUnavailable`
+    // without them.
+    crate::runtime::observation::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority();
+    let project = tempfile::tempdir().unwrap();
+    let transcript = project.path().join("cursor-replayed.jsonl");
+    std::fs::write(
+        &transcript,
+        concat!(
+            "{\"role\":\"user\",\"message\":{\"content\":[{\"type\":\"text\",\"text\":\"Edit the shared file.\"}]}}\n",
+            "{\"role\":\"assistant\",\"message\":{\"content\":[{\"type\":\"text\",\"text\":\"Saved src/lib.rs.\"}]}}\n"
+        ),
+    )
+    .unwrap();
+    let event = json!({
+        "session_id": "session-replayed",
+        "conversation_id": "conversation-replayed",
+        "generation_id": "generation-replayed",
+        "transcript_path": transcript,
+        "workspace_roots": [project.path()],
+    })
+    .to_string();
+    let project_id = ProjectId::new("project.cursor-replayed").unwrap();
+    (project, event, project_id)
+}
+
+/// A pass that admits observations and then loses the projection queue to a
+/// peer drainer still committed those observations.
+///
+/// This is the production interleaving on a slow runner: the explicit hook
+/// ingest admits the transcript, the project catch-up sweep's scheduler tick
+/// drains the scope-wide projection queue, and the ingest's own drain then
+/// finds nothing left. Reporting only the projections this pass drained itself
+/// turns a real commit into a terminal, non-retryable `accepted_for_replay`.
+#[tokio::test]
+async fn cursor_ingest_reports_its_commit_when_a_peer_drains_the_projection_queue() {
+    let (_project, event, project_id) = cursor_replay_fixture();
+    let admission = crate::admission::test_support::MemoryHostAdmission::default();
+    let scope = ObservationScopeV1::Project {
+        project_id: project_id.clone(),
+    };
+
+    // Admit exactly as the hook ingest does, then let a peer empty the queue
+    // before the ingest's own drain can run.
+    let transcript: PathBuf = serde_json::from_str::<Value>(&event).unwrap()["transcript_path"]
+        .as_str()
+        .map(PathBuf::from)
+        .unwrap();
+    let source_event: Value = serde_json::from_str(&event).unwrap();
+    let context = cursor_observation_context(&source_event, &transcript, false);
+    let progress = admit_cursor_jsonl_observations(
+        "session-replayed",
+        &transcript,
+        &context,
+        &admission,
+        &scope,
+        None,
+        &ObservationCancellation::default(),
+    )
+    .await
+    .unwrap();
+    assert!(
+        progress.frames_persisted > 0,
+        "the admit persists the transcript frames: {progress:?}"
+    );
+    let peer = projection::drain_cursor_observation_projections(
+        &admission,
+        &scope,
+        &ObservationCancellation::default(),
+    )
+    .await
+    .unwrap();
+    assert!(
+        peer.messages_upserted > 0,
+        "the peer drainer takes the queued rows: {peer:?}"
+    );
+
+    // The ingest now re-scans an exhausted source against an empty queue.
+    let stats = try_ingest_cursor_transcript_event_capped_with_admission(
+        &event, project_id, &admission, None,
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        stats.messages_upserted, 0,
+        "the peer already projected these rows: {stats:?}"
+    );
+    assert!(
+        stats.observations_committed > 0 || stats.exact_duplicate,
+        "an ingest whose observations are durable must not look like a pass that captured nothing: {stats:?}"
+    );
+}
+
+/// A pass whose observations a peer drainer already projected must not report
+/// the same zero-change accounting as a pass that captured nothing.
+///
+/// The daemon's project catch-up sweep drains the whole Cursor projection
+/// queue for a scope, not just the rows it admitted itself, and projection
+/// consumes the queue row. So an explicit hook ingest that admitted on a
+/// deferred first call can find the queue empty on its next call even though
+/// its own observations are durably committed. Reported as an unqualified
+/// zero, the admission completes as `accepted_for_replay`: terminal,
+/// non-retryable, and proving nothing.
+#[tokio::test]
+async fn replayed_cursor_ingest_reports_an_exact_duplicate_not_a_bare_replay() {
+    let (_project, event, project_id) = cursor_replay_fixture();
+    let admission = crate::admission::test_support::MemoryHostAdmission::default();
+
+    let committed = try_ingest_cursor_transcript_event_capped_with_admission(
+        &event,
+        project_id.clone(),
+        &admission,
+        None,
+    )
+    .await
+    .unwrap();
+    assert!(
+        committed.messages_upserted > 0,
+        "the first pass admits and projects the transcript: {committed:?}"
+    );
+    assert_eq!(
+        committed.observations_committed, 2,
+        "admission is the commit and is accounted for independently of whichever \
+         drainer projects it: {committed:?}"
+    );
+    assert!(
+        !committed.exact_duplicate,
+        "a pass that committed rows is not a duplicate: {committed:?}"
+    );
+
+    // Same event again: the source cursor is at end of file and the projection
+    // queue this scope shares with the catch-up sweep is already empty.
+    let replayed = try_ingest_cursor_transcript_event_capped_with_admission(
+        &event, project_id, &admission, None,
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        replayed.messages_upserted, 0,
+        "an already-projected replay upserts nothing: {replayed:?}"
+    );
+    assert!(
+        !replayed.source_deferred,
+        "nothing is left to defer: {replayed:?}"
+    );
+    assert!(
+        replayed.exact_duplicate,
+        "a replay of already-durable observations is an exact duplicate, not a bare accepted-for-replay: {replayed:?}"
+    );
+}
+
+/// The duplicate verdict is evidence, not a default: a source this pass has
+/// never opened carries no proof that anything was committed before.
+#[tokio::test]
+async fn first_cursor_ingest_of_an_empty_source_is_never_an_exact_duplicate() {
+    crate::runtime::observation::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority();
+    let project = tempfile::tempdir().unwrap();
+    let transcript = project.path().join("cursor-empty.jsonl");
+    std::fs::write(&transcript, "").unwrap();
+    let event = json!({
+        "session_id": "session-empty",
+        "transcript_path": transcript,
+        "workspace_roots": [project.path()],
+    })
+    .to_string();
+    let admission = crate::admission::test_support::MemoryHostAdmission::default();
+
+    let stats = try_ingest_cursor_transcript_event_capped_with_admission(
+        &event,
+        ProjectId::new("project.cursor-empty").unwrap(),
+        &admission,
+        None,
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(stats.messages_upserted, 0);
+    assert!(
+        !stats.exact_duplicate,
+        "a first-ever scan proves no prior commit: {stats:?}"
+    );
+}
diff --git a/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs b/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs
index 0d2029f659..24f02e9a60 100644
--- a/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs
+++ b/crates/tracedecay-sessions/src/runtime/ingest/project_provider.rs
@@ -260,7 +260,7 @@ impl<'a> ProjectProviderRun<'a> {
         let mut deferred = discovery.is_truncated();
         let mut frontier_committable = true;
         let mut outcome = ProviderRunOutcome::bounded(TranscriptIngestStats::default(), 0, false);
-        for (path_index, path) in discovery.paths.iter().enumerate() {
+        for path in &discovery.paths {
             if remaining == 0 {
                 deferred = true;
                 frontier_committable = false;
@@ -286,10 +286,15 @@ impl<'a> ProjectProviderRun<'a> {
                     frontier_committable &=
                         !progress.source_deferred && progress.bytes_consumed <= remaining;
                     remaining = remaining.saturating_sub(progress.bytes_consumed);
-                    if progress.bytes_consumed > 0
-                        && (progress.source_deferred
-                            || path_index.saturating_add(1) < discovery.paths.len())
-                    {
+                    // Only a source the admission left mid-window ends the
+                    // pass: it owns the next one, and no discovery frontier may
+                    // commit past it. An exhausted source must not, or a pass
+                    // admits at most one source however much budget is left,
+                    // never commits a frontier, and the next pass rediscovers
+                    // and re-reads every source it already finished. The byte
+                    // budget above is the pass bound here, exactly as it is in
+                    // the profile-scope loop.
+                    if progress.source_deferred {
                         deferred = true;
                         frontier_committable = false;
                         break;
@@ -922,6 +927,8 @@ mod tests {
                 messages_upserted: 3,
                 bytes_consumed: 4,
                 source_deferred: true,
+                observations_committed: 0,
+                exact_duplicate: false,
             },
             session_ids: BTreeSet::from(["shared-session".to_string()]),
         };
diff --git a/crates/tracedecay-sessions/src/runtime/ingest/user.rs b/crates/tracedecay-sessions/src/runtime/ingest/user.rs
index 43904b8849..f2e15c29b3 100644
--- a/crates/tracedecay-sessions/src/runtime/ingest/user.rs
+++ b/crates/tracedecay-sessions/src/runtime/ingest/user.rs
@@ -779,6 +779,8 @@ mod cursor_tests {
                 messages_upserted: 3,
                 bytes_consumed: 4,
                 source_deferred: true,
+                observations_committed: 0,
+                exact_duplicate: false,
             },
             session_ids: BTreeSet::from(["shared-session".to_string()]),
         };
diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs
index 22ff24262b..468022fd84 100644
--- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs
+++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs
@@ -276,6 +276,11 @@ pub(in crate::runtime) struct JsonlObservationAdmissionProgress {
     pub frames_rejected_before_decode: u64,
     pub frames_refused: u64,
     pub frames_persisted: u64,
+    /// This pass resumed from a durable source cursor rather than opening the
+    /// source for the first time. With `frames_persisted == 0` that is the
+    /// only evidence a caller has that the source was *already* admitted,
+    /// versus never carrying anything: both report zero new frames.
+    pub resumed: bool,
     pub io: crate::runtime::source::JsonlIoAccounting,
 }
 
@@ -2280,6 +2285,7 @@ pub(in crate::runtime) async fn admit_jsonl_observations<State: Clone>(
     let mut progress = JsonlObservationAdmissionProgress {
         bytes_consumed: raw.read_through.saturating_sub(raw.start_offset),
         source_deferred: raw.deferred.is_some(),
+        resumed: had_expected_cursor,
         io: if shared_page_hit {
             JsonlIoAccounting::default()
         } else {
diff --git a/crates/tracedecay/src/daemon/production_harness/lcm_preserved_profile_journey_test.rs b/crates/tracedecay/src/daemon/production_harness/lcm_preserved_profile_journey_test.rs
index bb17b3c310..8ad630e212 100644
--- a/crates/tracedecay/src/daemon/production_harness/lcm_preserved_profile_journey_test.rs
+++ b/crates/tracedecay/src/daemon/production_harness/lcm_preserved_profile_journey_test.rs
@@ -531,40 +531,37 @@ fn assert_window_side(label: &str, returned: &[String], in_window: bool, payload
     }
 }
 
-/// The complement of the 12-hour window, once temporal convergence can serve
-/// it.
+/// One window read, retried until the projection can serve that window.
 ///
-/// A `stale` outcome is the projection reporting it has not caught up to those
-/// generations yet; reading absence out of it would let the window assertions
-/// pass on lag instead of on a window decision.
-async fn wait_for_pre_window_search(
+/// Retrieval is never blocked on convergence, which is the admission this
+/// journey asserts: a window the projection has not caught up to is answered
+/// with typed staleness instead of waiting for it. So any single read here can
+/// land in a lag window that background ingest opened after an earlier read of
+/// the same window was served, and reading absence out of that would let the
+/// window assertions pass on lag instead of on a window decision. Every window
+/// read retries under the same convergence budget the discovery wait uses.
+///
+/// The returned elapsed time is the served call alone, so the product search
+/// budget still measures one answer and not the wait in front of it.
+async fn converged_window_read(
     harness: &ProductionProjectCompositionHarnessV1,
     project: &Path,
-    origin: i64,
-    since: i64,
-) -> Value {
+    label: &str,
+    tool: &str,
+    arguments: Value,
+) -> (Duration, Value) {
     let deadline = Instant::now() + CONVERGENCE_WAIT;
     loop {
-        let payload = answered(
-            harness,
-            project,
-            "tracedecay_message_search",
-            json!({
-                "query": DIRECT_USER_QUERY,
-                "message_type": "direct_user",
-                "since": origin,
-                "until": since - 1,
-                "limit": SESSION_REPLAYS,
-                "format": "json",
-            }),
-        )
-        .await;
-        if payload["outcome"] != json!("stale") {
-            return payload;
+        let (elapsed, response) = timed_call(harness, project, tool, arguments.clone()).await;
+        let payload = retained_payload(&resolved(harness, project, tool, response).await);
+        // `lcm_grep` names typed staleness on `status`, `message_search` on
+        // `outcome`; a served page carries "stale" on neither.
+        if payload["status"] != json!("stale") && payload["outcome"] != json!("stale") {
+            return (elapsed, payload);
         }
         assert!(
             Instant::now() < deadline,
-            "the pre-window search never left typed staleness: {payload}"
+            "{label} never left typed staleness: {payload}"
         );
         tokio::time::sleep(Duration::from_millis(200)).await;
     }
@@ -765,9 +762,10 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval()
         "known TraceDecay worktree must return its correlated session: {sessions_for}"
     );
 
-    let (search_elapsed, search) = timed_call(
+    let (search_elapsed, search_payload) = converged_window_read(
         &harness,
         &project,
+        "direct-user 12-hour message_search",
         "tracedecay_message_search",
         json!({
             "query": DIRECT_USER_QUERY,
@@ -783,8 +781,6 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval()
         search_elapsed,
         SEARCH_BUDGET,
     );
-    let search_payload =
-        retained_payload(&resolved(&harness, &project, "tracedecay_message_search", search).await);
     assert_ne!(
         search_payload["status"],
         json!("error"),
@@ -807,7 +803,21 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval()
     // The exclusion above must be a window decision, not an empty corpus: the
     // complementary query over the same span returns exactly the replays the
     // 12-hour window drops, every one of them.
-    let excluded_search = wait_for_pre_window_search(&harness, &project, origin, since).await;
+    let (_, excluded_search) = converged_window_read(
+        &harness,
+        &project,
+        "pre-window direct-user search",
+        "tracedecay_message_search",
+        json!({
+            "query": DIRECT_USER_QUERY,
+            "message_type": "direct_user",
+            "since": origin,
+            "until": since - 1,
+            "limit": SESSION_REPLAYS,
+            "format": "json",
+        }),
+    )
+    .await;
     let excluded_sessions = message_hit_session_ids(&excluded_search);
     assert_window_side(
         "pre-window direct-user search",
@@ -823,9 +833,10 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval()
         );
     }
 
-    let (grep_elapsed, grep) = timed_call(
+    let (grep_elapsed, grep_payload) = converged_window_read(
         &harness,
         &project,
+        "direct-user 12-hour lcm_grep",
         "tracedecay_lcm_grep",
         json!({
             "query": DIRECT_USER_QUERY,
@@ -837,8 +848,6 @@ async fn preserved_profile_lcm_discovery_converges_without_blocking_retrieval()
     )
     .await;
     assert_under_budget("direct-user 12-hour lcm_grep", grep_elapsed, SEARCH_BUDGET);
-    let grep_payload =
-        retained_payload(&resolved(&harness, &project, "tracedecay_lcm_grep", grep).await);
     for hit in grep_hits(&grep_payload) {
         let snippet = hit["snippet"].as_str().unwrap_or_default();
         assert!(
diff --git a/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs b/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs
index 8c3a2499c6..946f6bf383 100644
--- a/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs
+++ b/crates/tracedecay/src/daemon/project_open_owners/code_index_reads/ignored_dependency_admission_tests.rs
@@ -386,10 +386,18 @@ async fn latest(
     project_root: &Path,
 ) -> LatestCompleteCodeIndexV1 {
     // Lightweight publication precedes complete-generation seating. Demand
-    // that complete state before using its imports as admission evidence.
-    tokio::time::timeout(Duration::from_secs(5), async {
+    // that complete state before using its imports as admission evidence. The
+    // seat is background work behind the scheduler mutex; under a loaded CI
+    // runner it has taken over 5 s, so the bound is a minute.
+    // Poll the dashboard projection alone while the owner is busy: the
+    // query-admission read (`latest_complete_fresh`) leaves a coalesced wake
+    // behind whenever it finds the worker holding the scheduler with an
+    // expired proof, and polling it every 25 ms re-armed a no-op pass faster
+    // than the ladder could settle to `Fresh` (CI run 35419627712: one
+    // minute of `Verifying`, then 0.3 s on the retry). Read the generation
+    // only once the ladder has settled.
+    tokio::time::timeout(Duration::from_mins(1), async {
         loop {
-            let _ = registry.latest_complete_fresh(project_root).await;
             if registry
                 .dashboard_freshness(project_root)
                 .await
diff --git a/crates/tracedecay/tests/common/mod.rs b/crates/tracedecay/tests/common/mod.rs
index 8ea4bdd407..28bbc596c4 100644
--- a/crates/tracedecay/tests/common/mod.rs
+++ b/crates/tracedecay/tests/common/mod.rs
@@ -1089,6 +1089,13 @@ pub fn spawn_tracedecay_daemon_with(
     spawn_tracedecay_daemon_process(&home, &binary, configure)
 }
 
+/// How long a replacement daemon waits for a stopped predecessor's endpoint to
+/// stop accepting before reporting it as still live.
+///
+/// Generous on purpose: the wait only costs time when a predecessor is
+/// genuinely still reachable, and a real leak still fails rather than hangs.
+const PREDECESSOR_DAEMON_VACATE_TIMEOUT: Duration = Duration::from_secs(10);
+
 fn spawn_tracedecay_daemon_process(
     home: &Path,
     binary: &Path,
@@ -1111,17 +1118,42 @@ fn spawn_tracedecay_daemon_process(
             })
             .is_some_and(|address| TcpStream::connect(address).is_ok())
     };
-    #[cfg(unix)]
-    assert!(
-        std::os::unix::net::UnixStream::connect(&socket_path).is_err(),
-        "refusing to replace a live test daemon at {}",
-        socket_path.display()
-    );
-    #[cfg(not(unix))]
-    assert!(
-        !portable_daemon_connectable(),
-        "refusing to replace a live test daemon recorded at {}",
-        authority_path.display()
+    // Stopping a predecessor daemon is asynchronous with respect to its
+    // endpoint: `kill` plus `wait` reaps the PID the harness spawned, but the
+    // kernel keeps the listening socket alive while *any* duplicate of that
+    // descriptor survives, including one a subprocess inherited across `fork`
+    // and still holds because it has not reached its own `exec` yet. Asserting
+    // instantaneously therefore reports an ordinary teardown tail as a live
+    // daemon, which is what `init_project_fixture` journeys (spawn, init, drop,
+    // spawn again) hit on a loaded runner. Wait a bounded time for the endpoint
+    // to stop accepting; a daemon that keeps accepting still fails with the
+    // same refusal.
+    poll_until(
+        Instant::now() + PREDECESSOR_DAEMON_VACATE_TIMEOUT,
+        Duration::from_millis(25),
+        || {
+            #[cfg(unix)]
+            let live = std::os::unix::net::UnixStream::connect(&socket_path).is_ok();
+            #[cfg(not(unix))]
+            let live = portable_daemon_connectable();
+            (!live).then_some(())
+        },
+        || {
+            #[cfg(unix)]
+            {
+                format!(
+                    "refusing to replace a live test daemon at {}",
+                    socket_path.display()
+                )
+            }
+            #[cfg(not(unix))]
+            {
+                format!(
+                    "refusing to replace a live test daemon recorded at {}",
+                    authority_path.display()
+                )
+            }
+        },
     );
 
     let mut command = Command::new(binary);
diff --git a/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs b/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs
index 2cec3b2c8c..bc1388df7f 100644
--- a/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs
+++ b/crates/tracedecay/tests/daemon_suite/code_index_ignored_dependencies_test/flight_tests.rs
@@ -283,8 +283,39 @@ async fn coalesced_publication_failure_preserves_the_scheduler_error_family() {
             &fixture.path().canonicalize().expect("canonical fixture"),
         );
     let pointer_path = scoped_store.join("active-code-generation-v1.json");
-    let pointer_bytes = std::fs::read(&pointer_path).expect("read active pointer");
-    std::fs::write(&pointer_path, b"{").expect("corrupt active pointer");
+    // Every production writer of the active pointer reads it, edits it in
+    // memory and renames a temporary over it while holding the exclusive
+    // generation-store lock. Corrupting the file without that lock races an
+    // in-flight read-modify-write whose rename then restores a valid pointer,
+    // and this owner publishes instead of failing closed. The racer is the
+    // background pass tail: it releases the background admission permit this
+    // owner then takes (registry/mount.rs, "release the background admission
+    // permit before HeadOpening / graph work") and keeps attaching the
+    // generation's text artifact afterwards, so neither the held admission
+    // nor the held scheduler mutex proves the store is quiet. Taking the
+    // store lock does: being granted it means no writer is mid-transaction,
+    // and any writer that starts after it is released reads the corruption
+    // under the lock and refuses instead of overwriting it.
+    let pointer_bytes = {
+        use tracedecay_code_index_retention::code_index_generations::try_acquire_code_generation_store_lock;
+
+        let store_lock = tokio::time::timeout(Duration::from_secs(5), async {
+            loop {
+                if let Some(lock) = try_acquire_code_generation_store_lock(&scoped_store)
+                    .expect("generation store lock")
+                {
+                    break lock;
+                }
+                tokio::time::sleep(Duration::from_millis(2)).await;
+            }
+        })
+        .await
+        .expect("no generation-store writer is mid-transaction");
+        let pointer_bytes = std::fs::read(&pointer_path).expect("read active pointer");
+        std::fs::write(&pointer_path, b"{").expect("corrupt active pointer");
+        drop(store_lock);
+        pointer_bytes
+    };
     owner_control.release();
     hold.release();
 
diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs
index fdf85beb3d..8acf0b25e8 100644
--- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs
+++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/graph_analysis_test.rs
@@ -3108,7 +3108,7 @@ fn symbol_facts(symbols: &Value) -> Value {
             }
         })
         .collect::<Vec<_>>();
-    facts.sort_by(|left, right| left.to_string().cmp(&right.to_string()));
+    facts.sort_by_key(ToString::to_string);
     Value::Array(facts)
 }
 
diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs
index c394a8c651..09534aa0b3 100644
--- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs
+++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_search_test.rs
@@ -534,15 +534,19 @@ async fn production_codex_hook_ingest_survives_message_search_reopen() {
     )
     .expect("production Codex hook ingest JSON");
     assert_eq!(ingest["completed"], true, "{ingest}");
-    // The composition's background Codex catch-up may admit the rollout
-    // before the hook pass reaches it, in which case the hook truthfully
-    // reports zero new bytes. Either path must leave the rollout durable and
-    // searchable, which the retrieval assertions below verify directly.
+    // The composition's background Codex catch-up may admit the rollout before
+    // the hook pass reaches it, in which case the hook persists no new frames
+    // and reports the rollout as an exact duplicate. Both terminals prove the
+    // transcript is durable; `accepted_for_replay` proves neither a commit nor
+    // a duplicate and must not be reported for a rollout that is on disk and
+    // admitted. Either path must also leave the rollout searchable, which the
+    // retrieval assertions below verify directly.
     assert!(
-        ingest["admission"]["status"]
-            .as_str()
-            .is_some_and(|status| status != "unavailable" && status != "unknown"),
-        "real Codex hook ingest was refused: {ingest}"
+        matches!(
+            ingest["admission"]["status"].as_str(),
+            Some("committed" | "exact_duplicate")
+        ),
+        "real Codex hook ingest proved neither a commit nor a duplicate: {ingest}"
     );
 
     let initial = production_codex_message_search(&harness, &project).await;
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 a02f6328ed..f397d6df9d 100644
--- a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs
+++ b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs
@@ -1036,9 +1036,17 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() {
             .expect("registered daemon ingest response text"),
     )
     .expect("registered daemon ingest payload");
-    assert_eq!(
-        payload["status"], "committed",
-        "registered daemon ingest did not commit: {response}"
+    // The daemon's project catch-up sweep drains the whole Cursor projection
+    // queue for this scope, so it can project the observations this ingest
+    // admitted on an earlier deferred pass. Both terminal states below prove
+    // the transcript is durable; only `accepted_for_replay` would not.
+    assert!(
+        matches!(
+            payload["status"].as_str(),
+            Some("committed" | "exact_duplicate")
+        ),
+        "registered daemon ingest did not commit: {response}\ndaemon log:\n{}",
+        std::fs::read_to_string(&daemon_log).expect("read isolated advisory daemon log"),
     );
 
     // Codex records a turn in its rollout, not in the Stop event, so the
diff --git a/crates/tracedecay/tests/transport_acceptance_suite/graph_rebuild_status_test.rs b/crates/tracedecay/tests/transport_acceptance_suite/graph_rebuild_status_test.rs
index 4b5eda1bee..6e741f8222 100644
--- a/crates/tracedecay/tests/transport_acceptance_suite/graph_rebuild_status_test.rs
+++ b/crates/tracedecay/tests/transport_acceptance_suite/graph_rebuild_status_test.rs
@@ -279,12 +279,15 @@ async fn wait_for_background_refresh(
 ///
 /// The batch only has to keep one refresh observable across a few status polls.
 /// At 768 files it instead indexed 98,304 symbols into 455 million lexical
-/// units and 645 MB on disk, which on a four-core runner takes ~61s to commit
-/// and then pushes the reopen past the composition harness's own 20s publish
-/// gate: no `RECEIPT_TIMEOUT` can rescue that, the journey simply cannot finish.
-/// 96 files still take seconds, so `partial_refresh_in_progress` is sampled
-/// many times over at [`POLL_INTERVAL`], and every later open stays inside its
-/// gate.
+/// units and 645 MB on disk, which on a four-core runner takes ~61s to commit:
+/// no `RECEIPT_TIMEOUT` can rescue that, the journey simply cannot finish. 96
+/// files of 16 symbols still take seconds, so `partial_refresh_in_progress` is
+/// sampled many times over at [`POLL_INTERVAL`]; the 128 symbols a file used to
+/// carry bought no extra samples and cost eight times the commit.
+///
+/// The batch is also retired in the offline commit before the first reopen, so
+/// its weight is paid by the refresh it exists for and not again by two reopens
+/// bounded by a publish gate this journey cannot raise.
 const REFRESH_BATCH_FILES: u32 = 96;
 
 fn install_background_batch(isolation_root: &Path, project: &Path) {
@@ -292,7 +295,7 @@ fn install_background_batch(isolation_root: &Path, project: &Path) {
     fs::create_dir_all(&staging).expect("background batch staging directory");
     for file_index in 0..REFRESH_BATCH_FILES {
         let mut source = String::new();
-        for symbol_index in 0..128_u32 {
+        for symbol_index in 0..16_u32 {
             writeln!(
                 source,
                 "pub fn refresh_probe_{file_index:04}_{symbol_index:03}(input: u32) -> u32 {{ input + {symbol_index} }}"
@@ -360,12 +363,25 @@ async fn background_refresh_and_reopen_report_only_servable_generations_inner()
     );
     harness.shutdown().await;
 
+    // The batch has done its only job: one background refresh stayed
+    // observable across many status polls. Leaving it installed makes every
+    // later reopen re-index the whole batch inside
+    // `ProductionProjectCompositionHarnessV1::open`'s fixed 20s publish gate, a
+    // budget this journey neither controls nor asserts on: on a contended
+    // four-core runner that reopen exhausts the gate and the open fails before
+    // any reopen assertion runs. Retiring the batch in the same offline commit
+    // keeps both reopen assertions exact -- a source change the closed daemon
+    // never saw, then a quiet checkout -- at the cost they actually need.
+    fs::remove_dir_all(project.join("src/refresh_batch")).expect("retire the background batch");
     fs::write(
         project.join("src/after_reopen.rs"),
         "pub fn after_reopen() -> &'static str { \"current\" }\n",
     )
     .expect("post-shutdown source");
-    commit_all(&project, "change source while daemon is closed");
+    commit_all(
+        &project,
+        "retire the batch and change source while daemon is closed",
+    );
     let reopened_revision = head(&project);
 
     let reopened = ProductionProjectCompositionHarnessV1::open(isolation.path(), [project.clone()])
diff --git a/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs b/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs
index 34cc736911..daca6a9303 100644
--- a/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs
+++ b/crates/tracedecay/tests/transport_acceptance_suite/typed_terminal_restart_acceptance/transport_boundaries.rs
@@ -105,25 +105,77 @@ fn http_mount(home: &Path) -> HttpMount {
     }
 }
 
+/// Posts to the daemon's HTTP mount, repeating while the daemon answers with a
+/// pre-admission problem whose own retry directive is `after_delay`.
+///
+/// The authority record can be published before a restarted daemon has opened
+/// the project, and the reply for that window is a typed, retryable
+/// `unavailable`, not a verdict on the project. The first observation a
+/// journey asserts on is the first one the daemon *admitted*, which is what a
+/// production client that honours the directive sees.
+fn post_application_once_admitted(
+    mount: &HttpMount,
+    project_id: &str,
+    route: &str,
+    body: &Value,
+) -> (u16, Value) {
+    let deadline = Instant::now() + AUTHORITY_TIMEOUT;
+    loop {
+        let (status, payload) = post_application(mount, project_id, route, body, None);
+        let problem = [&payload, &payload["value"], &payload["data"]]
+            .into_iter()
+            .map(|candidate| &candidate["problem"])
+            .find(|problem| problem.is_object())
+            .filter(|problem| problem["terminality"] == "pre_admission")
+            .filter(|problem| problem["retry"] == "after_delay");
+        match problem {
+            Some(problem) if Instant::now() < deadline => {
+                let millis = problem["retry_after_millis"].as_u64().unwrap_or(250);
+                std::thread::sleep(Duration::from_millis(millis));
+            }
+            _ => return (status, payload),
+        }
+    }
+}
+
 /// Opens the exact project through the same daemon-owned route as a production
 /// CLI client and returns the identity the daemon admitted. HTTP cannot infer
 /// this identity locally: its route accepts only the daemon's public ID.
 fn admitted_project_id(home: &Path, project: &Path) -> String {
     let project_arg = project.to_string_lossy().into_owned();
-    let output = tracedecay_command_with_home(home)
-        .current_dir(project)
-        .args([
-            "tool",
-            "--project",
-            project_arg.as_str(),
-            "storage_status",
-            "--args",
-            r#"{"include_details":false}"#,
-            "--json",
-        ])
-        .stdin(Stdio::null())
-        .output()
-        .expect("read daemon-admitted project identity");
+    let deadline = Instant::now() + AUTHORITY_TIMEOUT;
+    let output = loop {
+        let output = tracedecay_command_with_home(home)
+            .current_dir(project)
+            .args([
+                "tool",
+                "--project",
+                project_arg.as_str(),
+                "storage_status",
+                "--args",
+                r#"{"include_details":false}"#,
+                "--json",
+            ])
+            .stdin(Stdio::null())
+            .output()
+            .expect("read daemon-admitted project identity");
+        // Right after `init` the daemon can still be mounting the project's
+        // query authority; it says so with a pre-admission problem whose own
+        // retry directive is `after_delay`. Honour that directive, as a
+        // production client would, instead of treating it as a verdict.
+        let problem = serde_json::from_slice::<Value>(&output.stdout)
+            .ok()
+            .map(|envelope| envelope["problem"].clone())
+            .filter(|problem| problem["terminality"] == "pre_admission")
+            .filter(|problem| problem["retry"] == "after_delay");
+        match problem {
+            Some(problem) if !output.status.success() && Instant::now() < deadline => {
+                let millis = problem["retry_after_millis"].as_u64().unwrap_or(250);
+                std::thread::sleep(Duration::from_millis(millis));
+            }
+            _ => break output,
+        }
+    };
     assert!(
         output.status.success(),
         "storage_status failed while admitting the fixture project\nstdout:\n{}\nstderr:\n{}",
@@ -598,12 +650,11 @@ fn reset_required_survives_http_mcp_and_rust_sdk_across_restart() {
 
     let storage_status_body = json!({ "include_details": false });
 
-    let (http_status, http_body) = post_application(
+    let (http_status, http_body) = post_application_once_admitted(
         &mount,
         &identity,
         STORAGE_STATUS_ROUTE,
         &storage_status_body,
-        None,
     );
     super::assert_reset_required(
         &problem_envelope(&http_body, "HTTP reset required"),