perf(kv): keep exact-state recording off inference path

crates/skippy-cache/src/disk_tier.rs

@@ -565,6 +565,17 @@ impl PrefixDiskTier {
         }))
     }
 
+    /// Mark an entry as recently used without mapping or verifying its payload.
+    pub fn touch(&mut self, page_id: &str) -> bool {
+        let Some(entry) = self.entries.get_mut(page_id) else {
+            return false;
+        };
+        self.use_clock = self.use_clock.saturating_add(1);
+        entry.last_used_secs = now_secs();
+        entry.use_sequence = self.use_clock;
+        true
+    }
+
     pub fn contains(&self, page_id: &str) -> bool {
         self.entries.contains_key(page_id)
     }

crates/skippy-cache/src/exact_state.rs

@@ -206,6 +206,21 @@ impl<E: Clone + serde::Serialize> ExactStateCache<E> {
         self.misses.note_identity_mismatch();
     }
 
+    /// Mark an existing page as recently used without reconstructing or
+    /// replacing its payload.
+    ///
+    /// Page identities include the complete prefix identity, so an existing
+    /// entry is already the checkpoint the caller intends to record. This
+    /// lets record paths avoid exporting and re-hashing the same state.
+    pub fn touch(&mut self, page_id: &str) -> bool {
+        self.clock = self.clock.saturating_add(1);
+        if let Some(entry) = self.entries.get_mut(page_id) {
+            entry.last_used = self.clock;
+            return true;
+        }
+        self.disk.as_mut().is_some_and(|disk| disk.touch(page_id))
+    }
+
     pub fn disk_contains(&self, page_id: &str) -> bool {
         self.disk
             .as_ref()
@@ -515,6 +530,36 @@ mod tests {
         assert_eq!(cache.stats().entries, 1);
     }
 
+    #[test]
+    fn touching_existing_page_refreshes_lru_without_replacing_payload() {
+        let mut cache = ExactStateCache::new(2, 0);
+        cache.record(
+            "first".to_string(),
+            2,
+            ExactStatePayload::full_state(vec![1, 2]),
+            (),
+        );
+        cache.record(
+            "second".to_string(),
+            2,
+            ExactStatePayload::full_state(vec![3, 4]),
+            (),
+        );
+
+        assert!(cache.touch("first"));
+        assert!(!cache.touch("missing"));
+        cache.record(
+            "third".to_string(),
+            2,
+            ExactStatePayload::full_state(vec![5, 6]),
+            (),
+        );
+
+        assert!(cache.lookup("first").is_some());
+        assert!(cache.lookup("second").is_none());
+        assert!(cache.lookup("third").is_some());
+    }
+
     #[test]
     fn cached_token_counts_are_bounded_sorted_and_deduplicated() {
         let mut cache = ExactStateCache::new(4, 0);

crates/skippy-server/Cargo.toml

@@ -28,6 +28,7 @@ futures-util = "0.3"
 skippy-runtime = { path = "../skippy-runtime", version = "0.76.0-rc4"  }
 skippy-tokenizer = { path = "../skippy-tokenizer", version = "0.76.0-rc4" }
 model-artifact = { path = "../model-artifact", version = "0.76.0-rc4" }
+mesh-llm-events = { path = "../mesh-llm-events", version = "0.76.0-rc4" }
 mesh-native-serving-plugin-api = { path = "../mesh-native-serving-plugin-api", version = "0.76.0-rc4" }
 skippy-protocol = { path = "../skippy-protocol", version = "0.76.0-rc4"  }
 skippy-cache = { path = "../skippy-cache", version = "0.76.0-rc4"  }

crates/skippy-server/src/frontend/local_generation/token_generation.rs

@@ -600,6 +600,18 @@ impl StageOpenAiBackend {
                     "skippy.exact_cache.reconstruct_blocks".to_string(),
                     json!(restored.reconstruct_blocks),
                 );
+                attrs.insert(
+                    "skippy.exact_cache.lookup_ms".to_string(),
+                    json!(restored.lookup_ms),
+                );
+                attrs.insert(
+                    "skippy.exact_cache.kv_import_ms".to_string(),
+                    json!(restored.kv_import_ms),
+                );
+                attrs.insert(
+                    "skippy.exact_cache.recurrent_import_ms".to_string(),
+                    json!(restored.recurrent_import_ms),
+                );
                 self.telemetry
                     .emit("stage.openai_kv_lookup_decision", attrs);
             }
@@ -1088,13 +1100,10 @@ impl StageOpenAiBackend {
                     "skippy.exact_cache.recorded_tokens".to_string(),
                     json!(record.token_count),
                 );
-                attrs.insert(
-                    "skippy.exact_cache.stored".to_string(),
-                    json!(record.stored),
-                );
+                attrs.insert("skippy.exact_cache.queued".to_string(), json!(true));
                 self.telemetry
                     .emit("stage.openai_kv_record_decision", attrs);
-                record.stored
+                true
             }
             Ok(None) => false,
             Err(error) => {

crates/skippy-server/src/kv_integration/config.rs

@@ -27,6 +27,7 @@ pub fn configure_kv_disk_cache(config: KvDiskCacheConfig) -> Result<(), KvDiskCa
 }
 
 use anyhow::Result;
+use mesh_llm_events::{OutputEvent, emit_event};
 use skippy_cache::{
     ExactStateCache, PrefixCandidatePolicy, PrefixDiskTier, ResidentActivationCache,
     ResidentCacheConfig, ResidentPrefixCache,
@@ -38,8 +39,8 @@ use skippy_runtime::ModelInfo;
 use skippy_topology::{STAGE_RUNTIME_LLAMA_FAMILY_EXPECTATIONS, infer_family_capability};
 
 use super::{
-    ExactStateExtra, KvStageIntegration, StageKvMode, StagePrefixCachePayload, disk_budget,
-    disk_budget::NodeBudget,
+    EXACT_STATE_RECORD_CAPACITY, ExactStateExtra, KvStageIntegration, PendingExactStateRecord,
+    StageKvMode, StagePrefixCachePayload, disk_budget, disk_budget::NodeBudget,
 };
 
 impl KvStageIntegration {
@@ -78,16 +79,54 @@ impl KvStageIntegration {
         if let Some(opened) = disk {
             exact_states = exact_states.with_disk_tier(opened.tier);
         }
+        let exact_states = Arc::new(Mutex::new(exact_states));
+        let (exact_state_record_tx, exact_state_record_rx) =
+            std::sync::mpsc::sync_channel::<PendingExactStateRecord>(EXACT_STATE_RECORD_CAPACITY);
+        let worker_exact_states = exact_states.clone();
+        let inflight_records = Arc::new(Mutex::new(BTreeSet::new()));
+        let worker_inflight_records = inflight_records.clone();
+        let exact_state_records_queued = Arc::new(std::sync::atomic::AtomicU64::new(0));
+        let exact_state_records_dropped = Arc::new(std::sync::atomic::AtomicU64::new(0));
+        let exact_state_records_pending = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let worker_exact_state_records_pending = exact_state_records_pending.clone();
+        let worker_disk_budget_reservation = disk_budget_reservation.clone();
+        std::thread::Builder::new()
+            .name(format!("skippy-exact-cache-{}", config.stage_id))
+            .spawn(move || {
+                let _disk_budget_reservation = worker_disk_budget_reservation;
+                while let Ok(pending) = exact_state_record_rx.recv() {
+                    let page_id = pending.page_id.clone();
+                    worker_exact_states
+                        .lock()
+                        .unwrap_or_else(std::sync::PoisonError::into_inner)
+                        .record(
+                            pending.page_id,
+                            pending.token_count,
+                            pending.payload,
+                            pending.extra,
+                        );
+                    worker_inflight_records
+                        .lock()
+                        .unwrap_or_else(std::sync::PoisonError::into_inner)
+                        .remove(&page_id);
+                    worker_exact_state_records_pending
+                        .fetch_sub(1, std::sync::atomic::Ordering::Release);
+                }
+            })?;
         Ok(Some(Self {
             mode,
             payload,
             correctness_mode: false,
             trust_local_writes: true,
             candidate_policy,
-            inflight_records: Arc::new(Mutex::new(BTreeSet::new())),
+            inflight_records,
             resident: Arc::new(Mutex::new(ResidentPrefixCache::new(resident_config))),
             activations: Arc::new(Mutex::new(ResidentActivationCache::new(resident_config))),
-            exact_states: Arc::new(Mutex::new(exact_states)),
+            exact_states,
+            exact_state_record_tx,
+            exact_state_records_queued,
+            exact_state_records_dropped,
+            exact_state_records_pending,
             first_tokens: Arc::new(Mutex::new(BTreeMap::new())),
             replay_tokens: Arc::new(Mutex::new(BTreeMap::new())),
             split_prefill_tokens: Arc::new(Mutex::new(BTreeMap::new())),
@@ -97,6 +136,13 @@ impl KvStageIntegration {
     }
 }
 
+fn emit_warning(message: String) {
+    let _ = emit_event(OutputEvent::Warning {
+        message,
+        context: None,
+    });
+}
+
 /// Open the KV disk tier for this stage. Host configuration wins; legacy
 /// environment variables remain compatibility input when no host configured it.
 struct OpenedDiskTier {
@@ -107,17 +153,19 @@ struct OpenedDiskTier {
 fn open_disk_tier(config: &StageConfig) -> Option<OpenedDiskTier> {
     let root = disk_tier_root(config);
     if !has_valid_content_digest(config) {
-        eprintln!(
+        emit_warning(format!(
             "skippy: KV disk tier disabled for stage {}: no valid content digest",
             config.stage_id
-        );
+        ));
         return None;
     }
     let reservation = stage_disk_budget(&root, config)?;
     match PrefixDiskTier::open(&root, reservation.bytes()) {
         Ok(tier) => Some(OpenedDiskTier { tier, reservation }),
         Err(error) => {
-            eprintln!("skippy: KV disk tier unavailable, continuing without it: {error}");
+            emit_warning(format!(
+                "skippy: KV disk tier unavailable, continuing without it: {error}"
+            ));
             None
         }
     }
@@ -149,12 +197,12 @@ fn stage_disk_budget(root: &Path, config: &StageConfig) -> Option<disk_budget::B
     };
     let budget = disk_budget::resolve_node_budget(explicit, enabled, free_bytes);
     if let NodeBudget::InsufficientSpace { free_bytes } = budget {
-        eprintln!(
+        emit_warning(format!(
             "skippy: KV disk tier disabled for stage {}: only {:.1} GiB free on {}",
             config.stage_id,
             free_bytes as f64 / (1024.0 * 1024.0 * 1024.0),
             probe.display(),
-        );
+        ));
         return None;
     }
     let node_bytes = budget.bytes()?;

crates/skippy-server/src/kv_integration/disk_budget.rs

@@ -174,6 +174,16 @@ mod tests {
         );
     }
 
+    #[test]
+    fn reservation_clone_keeps_ownership_until_last_clone_drops() {
+        let reservation = BudgetReservation(Arc::new(ReservationInner { bytes: 600 }));
+        let worker_reservation = reservation.clone();
+        assert_eq!(Arc::strong_count(&reservation.0), 2);
+
+        drop(reservation);
+        assert_eq!(Arc::strong_count(&worker_reservation.0), 1);
+    }
+
     #[test]
     fn reservation_is_reclaimed_on_drop() {
         let first = reserve(1000, 600).expect("first reservation");