test(code-index): fence the expired-proof read with the scheduler

crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry/test_gates.rs

@@ -554,6 +554,21 @@ impl CodeIndexSchedulerRegistryV1 {
             .map(|worktree| Arc::clone(&worktree.serving_source_witness))
     }
 
+    /// One mounted root's scheduler mutex, the lock every step that renews the
+    /// source proof must hold, so a test can age that proof and read it back
+    /// without a pass tail re-proving it in between.
+    #[cfg(test)]
+    pub(crate) async fn scheduler_for_root(
+        &self,
+        project_root: &Path,
+    ) -> Option<Arc<std::sync::Mutex<super::super::CodeIndexWorktreeSchedulerV1>>> {
+        let project_root = project_root.canonicalize().ok()?;
+        let mounted = self.mounted.lock().await;
+        mounted
+            .get(&project_root)
+            .map(|worktree| Arc::clone(&worktree.scheduler))
+    }
+
     /// The shared source-freshness fence for one mounted root, so tests can
     /// age its bounded proof instead of waiting the bound out in wall clock.
     #[cfg(test)]

crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/mod.rs

@@ -1265,6 +1265,55 @@ async fn drain_clone_backfill(registry: &CodeIndexSchedulerRegistryV1, path: &Pa
     }
 }
 
+/// Hold one mounted root's scheduler mutex until released, so no worker step
+/// can renew the source proof meanwhile.
+///
+/// The admission permit and the pass counter cannot fence this. The worker
+/// releases the permit after source reconciliation and drops its pass guard
+/// before the graph tail, whose renewing steps
+/// (`reconcile_retained_text_generation_with` and the serving swap's
+/// `currency_witness_for_sealed_snapshot`) take a guard only once a blocking
+/// thread reaches their closure. Both signals read idle in that gap while a
+/// renewal is already committed to run. Every renewing step takes this mutex
+/// and no read does.
+struct HeldSchedulerV1 {
+    release: Option<tokio::sync::oneshot::Sender<()>>,
+    held: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl HeldSchedulerV1 {
+    async fn release(mut self) {
+        drop(self.release.take());
+        if let Some(held) = self.held.take() {
+            held.await.expect("scheduler holder task");
+        }
+    }
+}
+
+async fn hold_scheduler_for_root(
+    registry: &CodeIndexSchedulerRegistryV1,
+    project_root: &Path,
+) -> HeldSchedulerV1 {
+    let scheduler = registry
+        .scheduler_for_root(project_root)
+        .await
+        .expect("mounted scheduler");
+    let (release, released) = tokio::sync::oneshot::channel();
+    let (acquired, holding) = tokio::sync::oneshot::channel();
+    let held = tokio::task::spawn_blocking(move || {
+        let _scheduler = scheduler
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        acquired.send(()).expect("report the held scheduler");
+        let _ = released.blocking_recv();
+    });
+    holding.await.expect("acquire the scheduler mutex");
+    HeldSchedulerV1 {
+        release: Some(release),
+        held: Some(held),
+    }
+}
+
 /// Hold the background worker out of a new pass, then wait for the in-flight
 /// pass to finish, and keep the admission permit.
 ///

crates/tracedecay-code-index-runtime/src/code_index_scheduler/tests/reconcile.rs

@@ -26,16 +26,17 @@ use super::{
     ALPHA_LIB_V1, GitFixture, RETAINED_REVISION_0, SERVING_SEAT_FAILURE_CEILING,
     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, 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,
+    hold_scheduler_for_root, 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,
+    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::{
@@ -2694,10 +2695,10 @@ async fn long_text_projection_renews_source_before_seating_and_noop_follow_up_se
     //
     // A pass that re-proves the seat rebinds the admission clock, so an
     // unfenced window between ageing the proof and reading it is a race with
-    // the worker, not an expiry test. Hold the single background admission
-    // across both: that parks the worker at its dequeue point and its
-    // acquisition already waited for any pass in flight to finish.
+    // the worker, not an expiry test. The admission permit alone does not
+    // close it (see `hold_scheduler_for_root`).
     let admission = quiesced_background_reconcile_admission(&registry, fixture.path()).await;
+    let scheduler = hold_scheduler_for_root(&registry, fixture.path()).await;
     {
         let mut state = source_freshness
             .state
@@ -2716,6 +2717,7 @@ async fn long_text_projection_renews_source_before_seating_and_noop_follow_up_se
             .is_none(),
         "the expired proof declines before the worker renews it"
     );
+    scheduler.release().await;
     drop(admission);
     assert_eq!(
         wait_until_serving_seat(&registry, fixture.path(), Duration::from_secs(10), || {