fix(code-index): isolate workers and keep the reader charge

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

@@ -38,6 +38,45 @@ use super::{
     publication_authority_is_terminal, retained_noop_requires_follow_up_wake,
 };
 
+/// Runtime that polls code-index workers, separate from the daemon's serving
+/// runtime.
+///
+/// The first reconcile of a cold checkout is the strict-readiness critical
+/// path. Sharing the daemon runtime left that worker unpolled for minutes
+/// while project-open tasks occupied every serving thread: the pass log then
+/// showed a 226s queue delay with the gates themselves taking 15µs, and the
+/// dogfood deadline expired still inside text projection, graph never seated.
+/// A process-wide runtime keeps indexing scheduled as soon as the wake is
+/// posted. The future is already boxed, so the stack only has to poll it.
+///
+/// Tests do not use this. The suite mounts many registries in one process;
+/// a few shared threads then hold a waiter behind every other mount until
+/// the per-test deadline. Each test already has a runtime that is not the
+/// daemon's serving runtime.
+#[cfg(not(test))]
+fn code_index_worker_runtime() -> Result<&'static tokio::runtime::Runtime, CodeIndexSchedulerErrorV1>
+{
+    static RUNTIME: OnceLock<Result<tokio::runtime::Runtime, String>> = OnceLock::new();
+    match RUNTIME.get_or_init(|| {
+        let workers = std::thread::available_parallelism()
+            .map_or(2, usize::from)
+            .clamp(2, 4);
+        tokio::runtime::Builder::new_multi_thread()
+            .thread_name("td-code-index")
+            .worker_threads(workers)
+            .max_blocking_threads(workers.saturating_add(8))
+            .thread_stack_size(8 * 1024 * 1024)
+            .enable_all()
+            .build()
+            .map_err(|error| error.to_string())
+    }) {
+        Ok(runtime) => Ok(runtime),
+        Err(error) => Err(CodeIndexSchedulerErrorV1::Identity(format!(
+            "code-index worker runtime failed to start: {error}"
+        ))),
+    }
+}
+
 impl CodeIndexSchedulerRegistryV1 {
     #[cfg(test)]
     pub fn open_worktree(
@@ -2332,10 +2371,16 @@ impl CodeIndexSchedulerRegistryV1 {
                 let _ = result;
             }
         });
+        #[cfg(test)]
         let task = tokio::spawn(hotpath::future!(
             worker_loop,
             label = "daemon.code_index.scheduler_worker"
         ));
+        #[cfg(not(test))]
+        let task = code_index_worker_runtime()?.spawn(hotpath::future!(
+            worker_loop,
+            label = "daemon.code_index.scheduler_worker"
+        ));
         entry.insert(MountedCodeIndexWorktreeV1 {
             project_id,
             repository_id,

crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs

@@ -2645,10 +2645,17 @@ impl LatestCodeTextGenerationV1 {
         control: &dyn CodeIndexExecutionControlV1,
     ) -> Result<Box<CodeTextCloneSuccessorBuildV1>, RetrievalPortError> {
         let generation_id = &self.metadata.manifest().generation_id;
+        // The successor's working set stays at
+        // `CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1`. The charge is at least
+        // the reader budget so publication can transfer it onto the reader
+        // component. A fresh reader admission at that boundary is refused
+        // when graph replay is already on the RSS watermark.
+        let publication_charge = CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1
+            .max(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1);
         let reservation = self.text_artifact_store.reserve_resident_memory(
             generation_id,
             "code-text-clone-successor",
-            CLONE_SUCCESSOR_MEMORY_BUDGET_BYTES_V1,
+            publication_charge,
         )?;
         let artifacts_root = code_text_artifacts_root(self.text_artifact_store.store_root());
         ensure_private_text_artifacts_root(&artifacts_root)?;
@@ -3292,15 +3299,11 @@ impl LatestCodeTextGenerationV1 {
             &sealed_identity,
             control,
         )?;
-        // The builder and source are gone, so its transient reservation no
-        // longer owns bytes. Release it before sampling the reader admission;
-        // the reader guard then carries the still-live unmodeled baseline.
-        drop(build_reservation);
-        let reader_reservation = store.reserve_resident_memory(
-            &self.metadata.manifest().generation_id,
-            "code-text-artifact-reader",
-            CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1,
-        )?;
+        // The reader is a smaller charge of the bytes this build already
+        // holds. Dropping the build reservation and reserving the reader
+        // again is a gap: an overlapping graph replay can sit on the process
+        // RSS watermark, and the new admission is then refused forever.
+        let reader_reservation = reader_charge_from_held_reservation(build_reservation)?;
         let final_path = code_text_artifact_path(store.store_root(), &descriptor)
             .map_err(text_artifact_unavailable)?;
         let reader = CodeLexicalArtifactReaderV1::open_content_addressed(
@@ -3405,19 +3408,19 @@ impl LatestCodeTextGenerationV1 {
             .finish(source_receipt, control)
             .map_err(map_text_artifact_error)?;
         drop(build.builder.take());
-        drop(build.build_reservation.take());
+        let held_reservation = build.build_reservation.take().ok_or_else(|| {
+            RetrievalPortError::Contract(
+                "clone-successor resident-memory charge disappeared before publication".to_owned(),
+            )
+        })?;
         let descriptor = self.text_artifact_store.publish_with_prior(
             &build.staging_path,
             &self.metadata.manifest().generation_id,
             &build.sealed_identity,
             Some(&build.prior_descriptor),
             control,
         )?;
-        let reader_reservation = self.text_artifact_store.reserve_resident_memory(
-            &self.metadata.manifest().generation_id,
-            "code-text-artifact-reader",
-            CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1,
-        )?;
+        let reader_reservation = reader_charge_from_held_reservation(held_reservation)?;
         let final_path =
             code_text_artifact_path(self.text_artifact_store.store_root(), &descriptor)
                 .map_err(text_artifact_unavailable)?;
@@ -3553,7 +3556,37 @@ impl LatestCodeTextGenerationV1 {
         build.builder = Some(builder);
         Ok(())
     }
+}
+
+/// Name a held publication charge as the reader and shrink it to the reader
+/// budget. The bytes never leave the ledger, so the handoff is not a new
+/// admission and measured RSS cannot refuse a charge that was already held.
+fn reader_charge_from_held_reservation(
+    mut held: ResidentMemoryReservationV1,
+) -> Result<ResidentMemoryReservationV1, RetrievalPortError> {
+    let reader_budget =
+        u64::try_from(CODE_LEXICAL_ARTIFACT_QUERY_CACHE_BUDGET_BYTES_V1).map_err(|_| {
+            RetrievalPortError::Contract("text-artifact reader budget exceeds u64".to_owned())
+        })?;
+    if held.reserved_bytes() < reader_budget {
+        return Err(RetrievalPortError::Contract(format!(
+            "text-artifact publication charge {} is below the reader budget {reader_budget}",
+            held.reserved_bytes()
+        )));
+    }
+    let component = ResidentMemoryComponentIdV1::new("code-text-artifact-reader")
+        .map_err(|error| RetrievalPortError::Contract(error.to_string()))?;
+    held.transfer_component(component, reader_budget)
+        .map_err(|error| {
+            RetrievalPortError::Contract(format!(
+                "text-artifact reader charge could not take over its held reservation: reserved {} measured {}",
+                error.reserved_bytes, error.measured_bytes
+            ))
+        })?;
+    Ok(held)
+}
 
+impl LatestCodeTextGenerationV1 {
     fn install_artifact_owners(
         &self,
         reader: CodeLexicalArtifactReaderV1,

crates/tracedecay-runtime-core/src/resident_memory.rs

@@ -1134,6 +1134,57 @@ impl ProcessResidentMemoryV1 {
         Ok(())
     }
 
+    /// Move one reservation's contribution onto `to_component` and keep only
+    /// `measured_bytes`, under the same lock.
+    ///
+    /// [`Self::reserve`] re-checks measured RSS. Dropping a charge and
+    /// reserving again is a gap: an overlapping consumer can sit on the
+    /// watermark and the new admission is refused even though these bytes
+    /// were already held. The ledger move does not ask for a new admission.
+    fn transfer_component(
+        &self,
+        from: &ResidentMemoryKeyV1,
+        to_component: ResidentMemoryComponentIdV1,
+        reserved_bytes: u64,
+        measured_bytes: u64,
+    ) -> Result<(), ResidentMemoryAdjustmentFailureV1> {
+        if measured_bytes > reserved_bytes {
+            return Err(ResidentMemoryAdjustmentFailureV1 {
+                reserved_bytes,
+                measured_bytes,
+            });
+        }
+        let mut state = self.lock_state();
+        let remove_source = {
+            let Some(charge) = state.charges.get_mut(from) else {
+                return Err(ResidentMemoryAdjustmentFailureV1 {
+                    reserved_bytes,
+                    measured_bytes,
+                });
+            };
+            if *charge < reserved_bytes {
+                return Err(ResidentMemoryAdjustmentFailureV1 {
+                    reserved_bytes: *charge,
+                    measured_bytes,
+                });
+            }
+            *charge -= reserved_bytes;
+            *charge == 0
+        };
+        if remove_source {
+            state.charges.remove(from);
+        }
+        let released_bytes = reserved_bytes - measured_bytes;
+        state.used_bytes -= released_bytes;
+        hotpath::gauge!("runtime_core.resident.used_bytes").set(state.used_bytes as f64);
+        if measured_bytes > 0 {
+            let mut to = from.clone();
+            to.component = to_component;
+            *state.charges.entry(to).or_default() += measured_bytes;
+        }
+        Ok(())
+    }
+
     fn release(&self, key: &ResidentMemoryKeyV1, reserved_bytes: u64) {
         if reserved_bytes == 0 {
             return;
@@ -1230,6 +1281,29 @@ impl ResidentMemoryReservationV1 {
         self.reserved_bytes = measured_bytes;
         Ok(())
     }
+
+    /// Keep this charge and name it `component`, shrinking to `measured_bytes`
+    /// when the held amount is larger.
+    ///
+    /// The bytes stay in the ledger for the whole move. Callers that instead
+    /// drop this reservation and [`ProcessResidentMemoryV1::reserve`] the
+    /// destination open a gap where measured RSS can refuse a charge that
+    /// was already admitted.
+    pub fn transfer_component(
+        &mut self,
+        component: ResidentMemoryComponentIdV1,
+        measured_bytes: u64,
+    ) -> Result<(), ResidentMemoryAdjustmentFailureV1> {
+        self.authority.transfer_component(
+            &self.key,
+            component,
+            self.reserved_bytes,
+            measured_bytes,
+        )?;
+        self.key.component = component;
+        self.reserved_bytes = measured_bytes;
+        Ok(())
+    }
 }
 
 impl Drop for ResidentMemoryReservationV1 {

crates/tracedecay-runtime-core/src/resident_memory/tests.rs

@@ -350,6 +350,57 @@ fn reservation_tracks_exact_identity_and_releases_on_drop() {
     assert_eq!(authority.snapshot().used_bytes, 0);
 }
 
+#[test]
+fn transfer_keeps_retained_bytes_when_a_new_admission_cannot() {
+    let authority = Arc::new(ProcessResidentMemoryV1::new(bytes(100)));
+    let build = key(
+        "project-a",
+        "worktree-a",
+        "generation-a",
+        "code-text-artifact-build",
+    );
+    let reader = key(
+        "project-a",
+        "worktree-a",
+        "generation-a",
+        "code-text-artifact-reader",
+    );
+    let mut held = authority
+        .reserve(build.clone(), bytes(80))
+        .expect("build reservation");
+    let _neighbor = authority
+        .reserve(
+            key("project-a", "worktree-a", "generation-a", "graph"),
+            bytes(20),
+        )
+        .expect("neighbor fills the ceiling");
+    let denied = authority
+        .reserve(reader.clone(), bytes(30))
+        .expect_err("a fresh reader admission does not fit beside the held build charge");
+    assert!(matches!(
+        denied,
+        ResidentMemoryAdmissionFailureV1::ReservationCeiling { .. }
+    ));
+
+    held.transfer_component(reader.component, 30)
+        .expect("the held charge moves without a new admission");
+    assert_eq!(held.key(), &reader);
+    assert_eq!(held.reserved_bytes(), 30);
+    let snapshot = authority.snapshot();
+    assert_eq!(snapshot.used_bytes, 50);
+    assert_eq!(snapshot.charge_for(&build), 0);
+    assert_eq!(snapshot.charge_for(&reader), 30);
+
+    let grown = held.transfer_component(reader.component, 40);
+    assert!(grown.is_err(), "a transfer cannot grow the held charge");
+    assert_eq!(held.reserved_bytes(), 30);
+    assert_eq!(authority.snapshot().charge_for(&reader), 30);
+
+    drop(held);
+    assert_eq!(authority.snapshot().used_bytes, 20);
+    assert_eq!(authority.snapshot().charge_for(&reader), 0);
+}
+
 #[test]
 fn rejection_reports_final_used_requested_and_limit_bytes() {
     let authority = Arc::new(ProcessResidentMemoryV1::new(bytes(100)));

crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs

@@ -1026,7 +1026,7 @@ async fn cursor_cas_lost_on_a_partially_covered_window_replays_the_tail() {
         })
     )
     .unwrap();
-    let len = u64::try_from(std::fs::metadata(&path).unwrap().len()).unwrap();
+    let len = std::fs::metadata(&path).unwrap().len();
     let spy = SeamSpyAdmission::default();
     spy.script_peer_covers_batch_prefix();