fix(code-index): stamp continuations before the seat looks idle

crates/tracedecay-code-index-retention/src/code_index_generations.rs

@@ -778,6 +778,25 @@ pub fn plan_code_generation_retention_with_verification(
     )
 }
 
+/// A store directory that does not exist yet has nothing to collect. The
+/// sealer creates that directory on open; until then the census is the same
+/// unpublished plan an empty directory with no pointer produces.
+fn unpublished_store_plan(
+    vector_readable_sources: &BTreeSet<CodeGenerationId>,
+) -> CodeGenerationRetentionPlanV1 {
+    CodeGenerationRetentionPlanV1 {
+        active_generation_id: None,
+        vector_readable_sources: vector_readable_sources.clone(),
+        superseded_generations: Vec::new(),
+        collectable_generations: Vec::new(),
+        collectable_text_artifacts: Vec::new(),
+        collectable_generation_segments: GenerationSegmentCensusV1::NoneFound,
+        text_artifact_inventory_bytes: 0,
+        verification: GenerationDigestVerificationV1::Full,
+        active_pointer: None,
+    }
+}
+
 /// Recover any bounded prior apply, then build the next fully verified
 /// collection unit while preserving the caller's cancellation authority.
 ///
@@ -795,6 +814,25 @@ pub fn prepare_next_code_generation_retention_cancellable(
     if observe_cancel(is_cancelled) {
         return Err(CodeGenerationRetentionErrorV1::Cancelled);
     }
+    // The serving seat can name a generation before the scoped store directory
+    // exists: cold open creates it inside the worker, and a waiter that only
+    // saw `latest_generation_id` plans against a path canonicalize then
+    // reports as `Storage(NotFound)`. That is an unpublished store, the same
+    // typed state as a directory with no pointer, not a storage failure the
+    // failure ceiling then retries.
+    match std::fs::metadata(store_root) {
+        Ok(metadata) if metadata.is_dir() => {}
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Ok(unpublished_store_plan(vector_readable_sources));
+        }
+        Ok(_) => {
+            return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
+                "code-generation store '{}' is not a directory",
+                store_root.display()
+            )));
+        }
+        Err(error) => return Err(storage(error)),
+    }
     recover_code_generation_retention_cancellable(
         store_root,
         vector_readable_sources,

crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs

@@ -3118,3 +3118,22 @@ fn recovery_completes_a_committed_rewrite_that_never_reached_the_pointer() {
     plan_code_generation_retention(fixture.store.path(), &BTreeSet::new())
         .expect("a recovered store must stay plannable");
 }
+
+#[test]
+fn missing_store_is_an_unpublished_plan_not_a_storage_failure() {
+    let missing = std::env::temp_dir().join(format!(
+        "tracedecay-missing-code-store-{}",
+        std::process::id()
+    ));
+    assert!(!missing.exists());
+    let plan = prepare_next_code_generation_retention_cancellable(
+        &missing,
+        &BTreeSet::new(),
+        &|| false,
+        None,
+    )
+    .expect("a store that has not been opened is unpublished");
+    assert_eq!(plan.active_generation_id, None);
+    assert!(plan.collectable_generations.is_empty());
+    assert!(!plan.has_collectable_work());
+}

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

@@ -2109,6 +2109,21 @@ impl CodeIndexSchedulerRegistryV1 {
         }
     }
 
+    /// Stamp a continuation while `reconcile_in_progress` still reports this pass.
+    ///
+    /// A seat waiter that sees the pass counter at zero treats the owner as
+    /// idle. Noting `BusyFollowUp` after that drop is the race: the waiter
+    /// samples an empty slot, then loses to the stamp and burns the failure
+    /// ceiling. The guard lives only for the note.
+    fn note_visible_worker_continuation(
+        passes: &Arc<AtomicUsize>,
+        pending_wake: &PendingWakeV1,
+        wake: &tokio::sync::Notify,
+    ) {
+        let _visible = super::ReconcilePassGuard::enter(passes);
+        Self::note_worker_continuation(pending_wake, wake);
+    }
+
     /// Claim the pending wake as one reconcile's arrival, at the instant the
     /// scheduler dequeues it.
     ///

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

@@ -1168,7 +1168,20 @@ impl CodeIndexSchedulerRegistryV1 {
                 // A successor-only retained projection holds no pass guard of
                 // its own; keeping the worker's guard through graph seat would
                 // report rebuild_in_flight for clone backfill that is not
-                // exact/lexical work.
+                // exact/lexical work. Stamp the continuation this projection
+                // already owes before that drop, so idle means the slot is set.
+                if let Some(outcome) = published_text_projection_outcome.as_ref() {
+                    let schedule_continuation = match outcome {
+                        PublishedTextProjectionOutcomeV1::Finished => graph_text
+                            .as_ref()
+                            .is_some_and(LatestCodeTextGenerationV1::text_projection_needs_work),
+                        PublishedTextProjectionOutcomeV1::Unfinished => true,
+                        PublishedTextProjectionOutcomeV1::Shutdown => false,
+                    };
+                    if schedule_continuation {
+                        Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
+                    }
+                }
                 if retained_text_projection.is_none() || retained_projection_successor_only {
                     drop(reconcile_pass.take());
                 }
@@ -1400,8 +1413,13 @@ impl CodeIndexSchedulerRegistryV1 {
                     // all and never published the successor generation. The
                     // `retained_graph_head_recovery_attempted` guard above is
                     // now false for every later pass, so this cannot spin
-                    // another retained-recovery Noop.
-                    Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
+                    // another retained-recovery Noop. The optional-graph drop
+                    // may already have happened; re-enter the pass for the stamp.
+                    Self::note_visible_worker_continuation(
+                        &worker_reconcile_in_progress,
+                        &worker_pending_wake,
+                        &worker_wake,
+                    );
                 }
                 // A recovered revision-7 verified head already serves its
                 // native graph from the retained text owner, and that owner
@@ -1548,8 +1566,12 @@ impl CodeIndexSchedulerRegistryV1 {
                                 if roster_refusal_rebuild {
                                     // One pass, claimed from the scheduler, so
                                     // a refusal that keeps reproducing cannot
-                                    // spin this worker.
-                                    Self::note_worker_continuation(
+                                    // spin this worker. Stamp while a pass
+                                    // guard is held: the step guard above has
+                                    // already dropped, and an idle read must
+                                    // not sample the empty slot.
+                                    Self::note_visible_worker_continuation(
+                                        &worker_reconcile_in_progress,
                                         &worker_pending_wake,
                                         &worker_wake,
                                     );
@@ -1714,7 +1736,15 @@ impl CodeIndexSchedulerRegistryV1 {
                                 .as_ref()
                                 .is_some_and(LatestCodeTextGenerationV1::text_projection_needs_work)
                             {
-                                Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
+                                // Stamped before the optional-graph guard drop
+                                // when this outcome was already known. Re-enter
+                                // so a reader that cleared the slot during
+                                // graph cannot sample the stamp as idle.
+                                Self::note_visible_worker_continuation(
+                                    &worker_reconcile_in_progress,
+                                    &worker_pending_wake,
+                                    &worker_wake,
+                                );
                             }
                             // Large text projections can outlive the bounded
                             // source proof established before publication. The
@@ -1784,7 +1814,11 @@ impl CodeIndexSchedulerRegistryV1 {
                                 "the publication's text owner did not finish its projection; \
                                  the sealed generation stays unseated until it does"
                             );
-                            Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
+                            Self::note_visible_worker_continuation(
+                                &worker_reconcile_in_progress,
+                                &worker_pending_wake,
+                                &worker_wake,
+                            );
                         }
                     }
                     // Keep the pass lifetime around the post-projection source
@@ -1965,7 +1999,11 @@ impl CodeIndexSchedulerRegistryV1 {
                             if text_latest.text_projection_needs_work()
                                 && !text_latest.query_owners_are_ready()
                             {
-                                Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
+                                Self::note_visible_worker_continuation(
+                                    &worker_reconcile_in_progress,
+                                    &worker_pending_wake,
+                                    &worker_wake,
+                                );
                             }
                         }
                         Ok(Err(error)) => {
@@ -1994,7 +2032,27 @@ impl CodeIndexSchedulerRegistryV1 {
                 }
                 // The source proof and serving witness are now published as
                 // one lifecycle. Optional receipts do not keep source
-                // verification in flight.
+                // verification in flight. Stamp a clone-backfill continuation
+                // before the drop: a seat waiter that sees the counter at zero
+                // must already observe the slot, or it races the failure ceiling.
+                if clone_backfill_waiting_for_source
+                    && matches!(
+                        &result,
+                        Ok((Ok(CodeIndexReconcileOutcomeV1::Noop(_)), _, _))
+                    )
+                    && worker_text_generation
+                        .read()
+                        .unwrap_or_else(std::sync::PoisonError::into_inner)
+                        .is_some()
+                    && worker_source_freshness
+                        .ready_without_stat(&worker_project_root, &worker_shutting_down)
+                {
+                    Self::note_visible_worker_continuation(
+                        &worker_reconcile_in_progress,
+                        &worker_pending_wake,
+                        &worker_wake,
+                    );
+                }
                 drop(reconcile_pass.take());
                 if let Ok((Ok(outcome), _, _)) = &result {
                     // A pass that ran to a terminal outcome proves neither the
@@ -2046,12 +2104,8 @@ impl CodeIndexSchedulerRegistryV1 {
                             );
                         }
                         worker_serving_generation_changed.send_replace(());
-                        // The retained slice was checked before reconciliation
-                        // renewed this proof. Preserve its wake now that source
-                        // is current, without requiring another query arrival.
-                        if clone_backfill_waiting_for_source {
-                            Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
-                        }
+                        // The clone-backfill continuation was stamped before
+                        // this pass dropped `reconcile_in_progress`.
                     }
                 } else {
                     // Surface bounded non-terminal failure without new project-path data.
@@ -2269,7 +2323,6 @@ impl CodeIndexSchedulerRegistryV1 {
                             PublishedTextProjectionOutcomeV1::Unfinished
                         }
                     };
-                    drop(reconcile_pass.take());
                     match outcome {
                         PublishedTextProjectionOutcomeV1::Finished
                             if !retained_head_recovered_without_complete_replay
@@ -2281,7 +2334,11 @@ impl CodeIndexSchedulerRegistryV1 {
                             // full replay can proceed without overlapping it.
                             // A clone-fingerprint backfill changed no owner
                             // the seat reads, so it owes no such pass.
-                            Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
+                            Self::note_visible_worker_continuation(
+                                &worker_reconcile_in_progress,
+                                &worker_pending_wake,
+                                &worker_wake,
+                            );
                         }
                         PublishedTextProjectionOutcomeV1::Finished => {
                             let installed_owner_still_needs_work = worker_text_generation
@@ -2292,7 +2349,11 @@ impl CodeIndexSchedulerRegistryV1 {
                                     LatestCodeTextGenerationV1::text_projection_needs_work,
                                 );
                             if installed_owner_still_needs_work {
-                                Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
+                                Self::note_visible_worker_continuation(
+                                    &worker_reconcile_in_progress,
+                                    &worker_pending_wake,
+                                    &worker_wake,
+                                );
                             }
                         }
                         PublishedTextProjectionOutcomeV1::Shutdown => {
@@ -2313,9 +2374,16 @@ impl CodeIndexSchedulerRegistryV1 {
                             // stopped short would sleep until an unrelated
                             // arrival, exactly as the inline slice's own
                             // follow-up notify prevented.
-                            Self::note_worker_continuation(&worker_pending_wake, &worker_wake);
+                            Self::note_visible_worker_continuation(
+                                &worker_reconcile_in_progress,
+                                &worker_pending_wake,
+                                &worker_wake,
+                            );
                         }
                     }
+                    // The continuation is already in the slot. This drop is the
+                    // first moment the pass looks idle.
+                    drop(reconcile_pass.take());
                 }
                 if worker_shutting_down.load(Ordering::Acquire) {
                     tracing::info!(

crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/projector.rs

@@ -119,11 +119,11 @@ impl SessionTemporalRefreshProjector for CanonicalSessionTemporalProjector {
                 // Empty remaining range is a durable no-op: terminalize with an
                 // empty complete progress batch instead of deferring forever.
                 Ok(None) => canonical_noop_complete_effect(&recovery),
-                Err(error) if error.is_storage() => Err(
-                    SessionTemporalRefreshProjectorError::retryable(format!(
+                Err(error) if error.is_storage() => {
+                    Err(SessionTemporalRefreshProjectorError::retryable(format!(
                         "source_busy: {error}"
-                    )),
-                ),
+                    )))
+                }
                 Err(_) => Err(SessionTemporalRefreshProjectorError::terminal(
                     "projector_failed",
                 )),

crates/tracedecay-session-temporal-store/src/query.rs

@@ -246,8 +246,7 @@ pub(super) async fn read_observations(
                 // ceiling. Split until a single observation remains; that
                 // observation is then a typed storage failure, not a retry
                 // that looks like a busy source.
-                if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1
-                {
+                if observation_prefetch_exceeded_materialization_limit(&error) && chunk.len() > 1 {
                     let mid = start + chunk.len() / 2;
                     pending.push((mid, end));
                     pending.push((start, mid));
@@ -289,11 +288,18 @@ fn observation_prefetch_exceeded_materialization_limit(error: &SessionStoreError
     }
 }
 
+/// The error `read_observation` raises for an id the store does not hold, reused
+/// by callers that resolve prefetched observations out of a batch map.
+pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError {
+    storage_message(
+        PERSIST_OPERATION,
+        format!("source observation {} is missing", observation_id.as_str()),
+    )
+}
+
 #[cfg(test)]
 mod tests {
-    use super::{
-        PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage,
-    };
+    use super::{PERSIST_OPERATION, observation_prefetch_exceeded_materialization_limit, storage};
 
     #[test]
     fn materialization_limit_is_the_prefetch_split_signal() {
@@ -308,15 +314,8 @@ mod tests {
         assert!(observation_prefetch_exceeded_materialization_limit(
             &exceeded
         ));
-        assert!(!observation_prefetch_exceeded_materialization_limit(&locked));
+        assert!(!observation_prefetch_exceeded_materialization_limit(
+            &locked
+        ));
     }
 }
-
-/// The error `read_observation` raises for an id the store does not hold, reused
-/// by callers that resolve prefetched observations out of a batch map.
-pub(super) fn missing_observation(observation_id: &CanonicalObservationIdV1) -> SessionStoreError {
-    storage_message(
-        PERSIST_OPERATION,
-        format!("source observation {} is missing", observation_id.as_str()),
-    )
-}