fix(admission): make NotApplicable a terminal no-op

CHANGELOG.md

@@ -1710,6 +1710,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- *(admission)* `NotApplicable` is a terminal no-op in the shared replay-pass
+  decision. A closed status that leaves the spool unchanged now stops until
+  the next kick instead of entering the retryable backoff arm.
+
 - *(code-index)* the background worker consults the typed publication-authority
   park instead of a loop-local bool, so a park it has not yet observed still
   stops reconcile. Branch publication handles `NotApplicable` as a closed

crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs

@@ -841,6 +841,10 @@ impl ProfileHostAdmissionReplayWorker {
                         // Non-retryable failure: stop until the next explicit kick.
                         break;
                     }
+                    ReplayPassDecision::TerminalNoop => {
+                        consecutive_retryable = 0;
+                        break;
+                    }
                     ReplayPassDecision::Requeue => {
                         consecutive_retryable = 0;
                     }
@@ -1391,6 +1395,43 @@ mod tests {
         registry.shutdown().await;
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn not_applicable_pending_record_does_not_back_off() {
+        let temp = tempfile::TempDir::new().unwrap();
+        let profile_root = temp.path().join("profile");
+        std::fs::create_dir_all(&profile_root).unwrap();
+        let db_path = tracedecay_sessions::runtime::user_sessions_db_path(&profile_root);
+        let (runtime, _) =
+            tracedecay_host_admission::HostAdmissionRuntime::open_for_database(&db_path).unwrap();
+        let broker = Arc::new(tracedecay_host_admission::HostAdmissionBroker::new(runtime));
+        broker.admit("test:pending", b"pending").await.unwrap();
+        let registry = ProfileHostAdmissionReplayRegistry::default();
+        let pass_override = Arc::new(|| {
+            Box::pin(async { HostAdmissionOutcome::not_applicable("code_index_not_applicable") })
+                as std::pin::Pin<Box<dyn std::future::Future<Output = HostAdmissionOutcome> + Send>>
+        });
+
+        registry
+            .ensure_with_pass_override(&db_path, &profile_root, &broker, pass_override)
+            .await;
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while registry.pass_count(&db_path).await == 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("profile replay must attempt the not-applicable record");
+        tokio::time::sleep(Duration::from_millis(150)).await;
+        let passes = registry.pass_count(&db_path).await;
+        assert!(passes >= 1);
+        assert_eq!(registry.backoff_count(&db_path).await, 0);
+        tokio::time::sleep(Duration::from_millis(150)).await;
+
+        assert_eq!(registry.pass_count(&db_path).await, passes);
+        assert_eq!(registry.backoff_count(&db_path).await, 0);
+        registry.shutdown().await;
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn shutdown_cancels_and_joins_an_in_flight_pass() {
         let temp = tempfile::TempDir::new().unwrap();

crates/tracedecay-host-admission/src/replay.rs

@@ -7,7 +7,7 @@
 
 use std::time::Duration;
 
-use tracedecay_sessions::admission::HostAdmissionOutcome;
+use tracedecay_sessions::admission::{HostAdmissionOutcome, HostAdmissionStatus};
 
 const MAX_BACKOFF: Duration = Duration::from_secs(2);
 const INITIAL_BACKOFF: Duration = Duration::from_millis(25);
@@ -28,6 +28,7 @@ pub fn replay_backoff(attempt: u32, shift_cap: u32) -> Duration {
 }
 
 /// How a worker should proceed after one replay pass.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum ReplayPassDecision {
     /// The spool shrank and more work remains. Yield and re-run immediately.
     ProgressPending,
@@ -37,15 +38,33 @@ pub enum ReplayPassDecision {
     Stop,
     /// Re-evaluate the work condition without backoff.
     Requeue,
+    /// Closed `NotApplicable` left the spool unchanged. Stop until the next
+    /// kick without backoff and without a failure log.
+    TerminalNoop,
 }
 
 /// Classify one replay pass from its pending-count delta and outcome.
+///
+/// `NotApplicable` already closes the replay record. It must not enter the
+/// retryable backoff arm, even when `is_replay_progress` is true and the
+/// spool did not shrink: that arm is for work that may succeed later. A shrink
+/// with records still pending continues immediately; a drained spool requeues;
+/// an unchanged spool stops until the next external kick.
 pub fn classify_replay_pass(
     pending_before: usize,
     pending_after: usize,
     outcome: &HostAdmissionOutcome,
 ) -> ReplayPassDecision {
     let made_progress = pending_after < pending_before;
+    if outcome.status == HostAdmissionStatus::NotApplicable {
+        if made_progress && pending_after > 0 {
+            return ReplayPassDecision::ProgressPending;
+        }
+        if pending_after == 0 {
+            return ReplayPassDecision::Requeue;
+        }
+        return ReplayPassDecision::TerminalNoop;
+    }
     if made_progress && pending_after > 0 {
         ReplayPassDecision::ProgressPending
     } else if !made_progress
@@ -58,3 +77,42 @@ pub fn classify_replay_pass(
         ReplayPassDecision::Requeue
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{ReplayPassDecision, classify_replay_pass};
+    use tracedecay_sessions::admission::HostAdmissionOutcome;
+
+    #[test]
+    fn not_applicable_is_a_terminal_noop_unless_the_spool_actually_moves() {
+        let closed = HostAdmissionOutcome::not_applicable("code_index_not_applicable");
+        let mut flagged_retryable = closed.clone();
+        flagged_retryable.retryable = true;
+
+        assert_eq!(
+            classify_replay_pass(2, 2, &closed),
+            ReplayPassDecision::TerminalNoop
+        );
+        assert_eq!(
+            classify_replay_pass(2, 3, &flagged_retryable),
+            ReplayPassDecision::TerminalNoop
+        );
+        assert_eq!(
+            classify_replay_pass(2, 1, &closed),
+            ReplayPassDecision::ProgressPending
+        );
+        assert_eq!(
+            classify_replay_pass(1, 0, &closed),
+            ReplayPassDecision::Requeue
+        );
+
+        assert_eq!(
+            classify_replay_pass(2, 2, &HostAdmissionOutcome::accepted_for_replay()),
+            ReplayPassDecision::Backoff
+        );
+        assert_eq!(
+            classify_replay_pass(2, 2, &HostAdmissionOutcome::spool_corrupted()),
+            ReplayPassDecision::Stop
+        );
+    }
+}

crates/tracedecay-mcp/src/server/project_host_admission_replay.rs

@@ -203,6 +203,10 @@ impl ProjectHostAdmissionReplayWorker {
                         );
                         break;
                     }
+                    ReplayPassDecision::TerminalNoop => {
+                        consecutive_retryable = 0;
+                        break;
+                    }
                     ReplayPassDecision::Requeue => {
                         consecutive_retryable = 0;
                         if self.dirty.load(Ordering::Acquire) || pending_after > 0 {
@@ -275,6 +279,42 @@ mod tests {
         task.shutdown().await;
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn not_applicable_pending_record_stops_without_backoff() {
+        let temp = tempfile::TempDir::new().unwrap();
+        let (runtime, _) = tracedecay_host_admission::HostAdmissionRuntime::open(
+            temp.path(),
+            tracedecay_host_admission::SpoolBounds::default(),
+        )
+        .unwrap();
+        let broker = Arc::new(tracedecay_host_admission::HostAdmissionBroker::new(runtime));
+        broker.admit("test:pending", b"pending").await.unwrap();
+        let passes = Arc::new(AtomicUsize::new(0));
+        let passes_for_run = Arc::clone(&passes);
+        let pass: PassFn = Arc::new(move || {
+            let passes = Arc::clone(&passes_for_run);
+            Box::pin(async move {
+                passes.fetch_add(1, Ordering::AcqRel);
+                HostAdmissionOutcome::not_applicable("code_index_not_applicable")
+            })
+        });
+        let task = ProjectHostAdmissionReplayTask::start(broker, pass);
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while task.pass_count() == 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("project replay must attempt the not-applicable record");
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        assert_eq!(passes.load(Ordering::Acquire), 1);
+        assert_eq!(task.pass_count(), 1);
+        assert_eq!(task.backoff_count(), 0);
+        task.shutdown().await;
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn dropping_task_aborts_an_in_flight_pass_without_an_arc_cycle() {
         let temp = tempfile::TempDir::new().unwrap();

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()),
-    )
-}