fix(code-index): fence stale progress boundaries and repeated refusals

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

@@ -189,6 +189,7 @@ pub struct CodeIndexBuildProgressSlotStateV1 {
     generation_id: Option<CodeGenerationId>,
     pub(super) owner_epoch: u64,
     progress_epoch: u64,
+    last_progress_micros: Option<i64>,
     snapshot: Option<Arc<CodeIndexBuildProgressV1>>,
 }
 
@@ -198,27 +199,61 @@ impl CodeIndexBuildProgressSlotStateV1 {
         self.progress_epoch = self.progress_epoch.saturating_add(1).max(1);
         self.generation_id = Some(generation_id);
         self.snapshot = None;
+        self.last_progress_micros = None;
         self.owner_epoch
     }
 
+    /// A discarded or replaced staging file starts a different committed
+    /// sequence, even if the sealed generation and scheduler owner survive.
+    /// Only the head-opening singleflight may reset it, after it has verified
+    /// that a new staging builder was actually created. The next boundary
+    /// keeps the monotonic publication epoch and reports its real counters.
+    pub(super) fn begin_staging_attempt(
+        &mut self,
+        generation_id: &CodeGenerationId,
+        owner_epoch: u64,
+    ) -> bool {
+        if self.generation_id.as_ref() != Some(generation_id) || self.owner_epoch != owner_epoch {
+            return false;
+        }
+        self.snapshot = None;
+        true
+    }
+
     pub(super) fn publish(
         &mut self,
         generation_id: &CodeGenerationId,
         owner_epoch: u64,
         mut snapshot: CodeIndexBuildProgressV1,
     ) -> bool {
-        if self.generation_id.as_ref() != Some(generation_id) || self.owner_epoch != owner_epoch {
+        if self.generation_id.as_ref() != Some(generation_id)
+            || self.owner_epoch != owner_epoch
+            || snapshot.generation_id != generation_id.as_str()
+        {
             return false;
         }
-        // A progress producer can timestamp its snapshot before another
-        // producer publishes. Preserve the most recent observed timestamp
-        // even when the later lock acquisition delivers an older sample.
-        // Replacing the generation clears the snapshot and starts a new owner.
         if let Some(current) = self.snapshot.as_ref() {
-            snapshot.last_progress_micros = snapshot
-                .last_progress_micros
-                .max(current.last_progress_micros);
+            // Concurrent phase/report producers may complete out of order.
+            // A delayed report must not undo already committed boundaries or
+            // withdraw a ready text owner from the same generation.
+            if snapshot.committed_pages < current.committed_pages
+                || snapshot.committed_chunks < current.committed_chunks
+                || snapshot.committed_imports < current.committed_imports
+                || snapshot.committed_payload_bytes < current.committed_payload_bytes
+                || (current.phase == CodeIndexBuildPhaseV1::Ready
+                    && snapshot.phase != CodeIndexBuildPhaseV1::Ready)
+            {
+                return false;
+            }
         }
+        // The next producer may have assembled its report before an earlier
+        // producer acquired the lock. Keep the owner's latest observed time
+        // even across a verified staging restart (which clears the snapshot).
+        // Replacing the owner clears this timestamp authority.
+        if let Some(latest) = self.last_progress_micros {
+            snapshot.last_progress_micros = snapshot.last_progress_micros.max(latest);
+        }
+        self.last_progress_micros = Some(snapshot.last_progress_micros);
         self.progress_epoch = self.progress_epoch.saturating_add(1).max(1);
         snapshot.progress_epoch = self.progress_epoch;
 
@@ -233,6 +268,9 @@ impl CodeIndexBuildProgressSlotStateV1 {
         else {
             return;
         };
+        if snapshot.blocked_reason == Some(reason) {
+            return;
+        }
         let mut snapshot = snapshot.as_ref().clone();
         snapshot.blocked_reason = Some(reason);
         let _ = self.publish(&generation_id, self.owner_epoch, snapshot);
@@ -252,6 +290,7 @@ impl CodeIndexBuildProgressSlotStateV1 {
             return false;
         };
         if current.phase == CodeIndexBuildPhaseV1::Ready
+            || current.blocked_reason == Some(reason)
             || expected_progress_epoch.is_some_and(|epoch| current.progress_epoch != epoch)
         {
             return false;
@@ -2693,6 +2732,17 @@ impl LatestCodeTextGenerationV1 {
                 let Some(current) = slot.snapshot() else {
                     return;
                 };
+                // A retry that re-enters an unchanged phase has not made
+                // progress. Keep the last committed observation epoch and
+                // elapsed sample until a new phase or batch boundary arrives.
+                if current.phase == phase
+                    && current.current_batch_pages == current_batch_pages
+                    && current.current_batch_payload_bytes == current_batch_payload_bytes
+                    && !(phase == CodeIndexBuildPhaseV1::Ready
+                        && current.blocked_reason.is_some())
+                {
+                    return;
+                }
                 let mut snapshot = current.as_ref().clone();
                 snapshot.phase = phase;
                 if !phase_has_file_rate_estimate(phase) {
@@ -3220,6 +3270,10 @@ impl LatestCodeTextGenerationV1 {
         } else {
             None
         };
+        // A missing/incompatible staging file is a new durable attempt for
+        // this sealed generation. Its committed counters legitimately start
+        // at zero; the old attempt must not fence that fresh boundary.
+        let mut fresh_staging_attempt = resumed.is_none();
         let mut builder = match resumed {
             Some(builder) => builder,
             None => match self.carry_parent_text_artifact(
@@ -3262,6 +3316,7 @@ impl LatestCodeTextGenerationV1 {
                 Err(VerifiedSealedLexicalCursorRestoreErrorV1::IncompatiblePosition) => {
                     drop(builder);
                     store.discard_incompatible_staging(&staging_path, control)?;
+                    fresh_staging_attempt = true;
                     builder = CodeLexicalArtifactBuilderV1::create_with_memory_budget(
                         &staging_path,
                         metadata,
@@ -3288,6 +3343,15 @@ impl LatestCodeTextGenerationV1 {
             staging_path,
             _build_reservation: build_reservation,
         };
+        if fresh_staging_attempt
+            && !self
+                .text_progress_slot
+                .write()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .begin_staging_attempt(&generation_id, self.text_progress_owner_epoch)
+        {
+            return Err(RetrievalPortError::GenerationMismatch);
+        }
         self.publish_text_progress_boundary(
             &initialized,
             &progress,

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

@@ -3041,6 +3041,248 @@ fn progress_timestamp_never_regresses_within_one_owner() {
     assert_eq!(slot.snapshot().unwrap().last_progress_micros, 42);
 }
 
+#[test]
+fn delayed_progress_cannot_undo_committed_boundaries_or_ready() {
+    let generation = CodeGenerationId::new("generation.stale-progress").expect("generation");
+    let mut slot = super::super::CodeIndexBuildProgressSlotStateV1::default();
+    let owner = slot.replace_generation(generation.clone());
+
+    let mut committed = progress_snapshot_for_generation(&generation, 6);
+    committed.committed_imports = 3;
+    committed.last_progress_micros = 600;
+    assert!(slot.publish(&generation, owner, committed));
+    let accepted = slot.snapshot().expect("accepted committed progress");
+
+    for counter in ["pages", "chunks", "imports", "payload"] {
+        let mut delayed = accepted.as_ref().clone();
+        delayed.last_progress_micros = 700;
+        match counter {
+            "pages" => delayed.committed_pages -= 1,
+            "chunks" => delayed.committed_chunks -= 1,
+            "imports" => delayed.committed_imports -= 1,
+            "payload" => delayed.committed_payload_bytes -= 1,
+            _ => unreachable!(),
+        }
+        assert!(
+            !slot.publish(&generation, owner, delayed),
+            "delayed {counter} cannot overwrite committed progress"
+        );
+        assert_eq!(
+            slot.snapshot().expect("progress after stale report").as_ref(),
+            accepted.as_ref(),
+            "refusing stale {counter} must preserve the epoch and last progress time"
+        );
+    }
+
+    let mut wrong_generation = accepted.as_ref().clone();
+    wrong_generation.generation_id = "generation.foreign".to_owned();
+    assert!(!slot.publish(&generation, owner, wrong_generation));
+    assert_eq!(
+        slot.snapshot().expect("correct generation still owns progress").as_ref(),
+        accepted.as_ref()
+    );
+
+    let mut ready = progress_snapshot_for_generation(&generation, 7);
+    ready.committed_imports = 3;
+    ready.phase = CodeIndexBuildPhaseV1::Ready;
+    ready.last_progress_micros = 650;
+    assert!(slot.publish(&generation, owner, ready));
+    let settled = slot.snapshot().expect("ready progress");
+
+    let mut late_build = settled.as_ref().clone();
+    late_build.phase = CodeIndexBuildPhaseV1::BulkCommit;
+    late_build.last_progress_micros = 700;
+    assert!(!slot.publish(&generation, owner, late_build));
+    assert_eq!(
+        slot.snapshot().expect("ready progress remains").as_ref(),
+        settled.as_ref(),
+        "a delayed build phase must not withdraw ready"
+    );
+
+    // A remounted owner starts a new committed sequence even when its
+    // generation identity is the same. The old owner must stay fenced off.
+    let replacement_owner = slot.replace_generation(generation.clone());
+    let fresh = progress_snapshot_for_generation(&generation, 1);
+    assert!(slot.publish(&generation, replacement_owner, fresh));
+    assert_eq!(slot.snapshot().expect("new owner").committed_pages, 1);
+    assert!(!slot.publish(
+        &generation,
+        owner,
+        progress_snapshot_for_generation(&generation, 99)
+    ));
+}
+
+#[test]
+fn a_new_staging_attempt_can_truthfully_restart_committed_progress() {
+    let generation = CodeGenerationId::new("generation.staging-retry").expect("generation");
+    let mut slot = super::super::CodeIndexBuildProgressSlotStateV1::default();
+    let owner = slot.replace_generation(generation.clone());
+
+    let mut old = progress_snapshot_for_generation(&generation, 12);
+    old.last_progress_micros = 500;
+    assert!(slot.publish(&generation, owner, old));
+    let old_epoch = slot.snapshot().expect("first staging attempt").progress_epoch;
+
+    // Missing/incompatible SQLite staging is a verified new build attempt,
+    // not a delayed report from an older commit boundary. Without resetting
+    // the slot, rejecting counter regressions would leave status frozen.
+    assert!(slot.begin_staging_attempt(&generation, owner));
+    let mut restarted = progress_snapshot_for_generation(&generation, 0);
+    // A wall-clock adjustment or delayed sample must not rewind the
+    // timestamp even though a verified new staging attempt reset its counts.
+    restarted.last_progress_micros = 400;
+    restarted.phase = CodeIndexBuildPhaseV1::SourceScan;
+    assert!(slot.publish(&generation, owner, restarted));
+    let restarted = slot.snapshot().expect("new staging attempt");
+    assert_eq!(restarted.committed_pages, 0);
+    assert_eq!(restarted.phase, CodeIndexBuildPhaseV1::SourceScan);
+    assert_eq!(
+        restarted.last_progress_micros, 500,
+        "staging restart must not rewind the same owner's last progress timestamp"
+    );
+    assert!(
+        restarted.progress_epoch > old_epoch,
+        "a new verified staging attempt must produce a newer report epoch"
+    );
+
+    let mut resumed = progress_snapshot_for_generation(&generation, 2);
+    resumed.last_progress_micros = 700;
+    assert!(slot.publish(&generation, owner, resumed));
+    assert_eq!(slot.snapshot().expect("resumed commits").committed_pages, 2);
+
+    let foreign = CodeGenerationId::new("generation.foreign-stage").expect("foreign");
+    assert!(!slot.begin_staging_attempt(&foreign, owner));
+    assert!(!slot.begin_staging_attempt(&generation, owner + 1));
+    assert_eq!(slot.snapshot().expect("unchanged latest").committed_pages, 2);
+}
+
+#[test]
+fn retrying_an_unchanged_phase_does_not_fabricate_status_progress() {
+    let fixture = GitFixture::new(&[(
+        "src/lib.rs",
+        "pub fn caller() { callee(); }\npub fn callee() {}\n",
+    )]);
+    let store = TempDir::new().expect("store root");
+    let mut scheduler = scheduler(
+        &fixture,
+        store.path().to_path_buf(),
+        Arc::new(SharedCodeIndexBytePoolV1::default()),
+    );
+    published(scheduler.reconcile_now().expect("publish"));
+    let latest = scheduler.latest_complete().expect("latest generation");
+    assert!(
+        !latest.advance_text_serving(0).expect("open text projection"),
+        "a zero-work admission must leave the projection pending"
+    );
+    let before = build_progress_snapshot(&scheduler);
+    assert_ne!(before.phase, CodeIndexBuildPhaseV1::Ready);
+
+    latest.text.publish_text_progress_phase(
+        before.phase,
+        before.current_batch_pages,
+        before.current_batch_payload_bytes,
+    );
+    let repeated = build_progress_snapshot(&scheduler);
+    assert_eq!(
+        repeated.progress_epoch, before.progress_epoch,
+        "a retry reporting no new phase or batch boundary must not feign progress"
+    );
+    assert_eq!(repeated.last_progress_micros, before.last_progress_micros);
+    assert_eq!(repeated.committed_pages, before.committed_pages);
+
+    let next_phase = if before.phase == CodeIndexBuildPhaseV1::Verification {
+        CodeIndexBuildPhaseV1::GraphPublication
+    } else {
+        CodeIndexBuildPhaseV1::Verification
+    };
+    latest.text.publish_text_progress_phase(
+        next_phase,
+        before.current_batch_pages,
+        before.current_batch_payload_bytes,
+    );
+    let transitioned = build_progress_snapshot(&scheduler);
+    assert_eq!(transitioned.phase, next_phase);
+    assert!(transitioned.progress_epoch > before.progress_epoch);
+    latest.text.publish_text_progress_phase(
+        next_phase,
+        before.current_batch_pages,
+        before.current_batch_payload_bytes,
+    );
+    let unchanged = build_progress_snapshot(&scheduler);
+    assert_eq!(unchanged.progress_epoch, transitioned.progress_epoch);
+    assert_eq!(unchanged.last_progress_micros, transitioned.last_progress_micros);
+
+    let mut advances = 0;
+    while !latest.advance_text_serving(8).expect("resume durable text build") {
+        advances += 1;
+        assert!(advances < 1_000, "text build failed to finish");
+    }
+    let ready = build_progress_snapshot(&scheduler);
+    assert_eq!(ready.phase, CodeIndexBuildPhaseV1::Ready);
+    assert!(ready.progress_epoch > transitioned.progress_epoch);
+
+    // Ready is the one idempotent-looking phase that may still be a real
+    // state change: success must clear a previously published refusal.
+    scheduler
+        .build_progress_slot()
+        .write()
+        .expect("progress slot")
+        .block_current(CodeIndexBuildBlockedReasonV1::ResidentMemory);
+    let blocked = build_progress_snapshot(&scheduler);
+    assert_eq!(
+        blocked.blocked_reason,
+        Some(CodeIndexBuildBlockedReasonV1::ResidentMemory)
+    );
+    latest.text.publish_text_progress_phase(
+        CodeIndexBuildPhaseV1::Ready,
+        blocked.current_batch_pages,
+        blocked.current_batch_payload_bytes,
+    );
+    let cleared = build_progress_snapshot(&scheduler);
+    assert_eq!(cleared.blocked_reason, None);
+    assert!(cleared.progress_epoch > blocked.progress_epoch);
+}
+
+#[test]
+fn repeated_blocked_reason_is_not_reported_as_new_progress() {
+    let generation = CodeGenerationId::new("generation.blocked-report").expect("generation");
+    let mut slot = super::super::CodeIndexBuildProgressSlotStateV1::default();
+    let owner = slot.replace_generation(generation.clone());
+    assert!(slot.publish(
+        &generation,
+        owner,
+        progress_snapshot_for_generation(&generation, 2)
+    ));
+    slot.block_current(CodeIndexBuildBlockedReasonV1::ResidentMemory);
+    let first = slot.snapshot().expect("first refusal");
+    assert_eq!(
+        first.blocked_reason,
+        Some(CodeIndexBuildBlockedReasonV1::ResidentMemory)
+    );
+
+    slot.block_current(CodeIndexBuildBlockedReasonV1::ResidentMemory);
+    assert!(!slot.block_owned_text(
+        &generation,
+        owner,
+        CodeIndexBuildBlockedReasonV1::ResidentMemory,
+        Some(first.progress_epoch)
+    ));
+    assert_eq!(
+        slot.snapshot().expect("repeated refusal").progress_epoch,
+        first.progress_epoch,
+        "retries without a new refusal or committed work must not inflate the observation epoch"
+    );
+
+    let mut committed = progress_snapshot_for_generation(&generation, 3);
+    committed.last_progress_micros = first.last_progress_micros + 1;
+    assert!(slot.publish(&generation, owner, committed));
+    assert_eq!(
+        slot.snapshot().expect("committed progress").blocked_reason,
+        None,
+        "durable progress clears the previous blocked verdict"
+    );
+}
+
 #[test]
 fn progress_owner_epoch_rejects_original_a_after_a_b_a_replacement() {
     let generation_a = CodeGenerationId::new("generation.progress-a").expect("generation A");