feat: define DeepSeek V4.1 text architecture and execution contract

THIRD_PARTY_NOTICES.md

@@ -437,6 +437,33 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 SOFTWARE.
 ```
 
+## SGLang DeepSeek-V4.1 serving contracts
+
+The V4.1 execution and performance-model logic in
+`src/aisimulate_core/sdk/deepseek_v41.py`, `sdk/models/deepseek_v41.py`
+(with the same `src/aisimulate_core/` prefix), and repository-root
+`crates/core/src/perfmodel/operators/dsv41.rs` is informed by and modified from
+SGLang's serving architecture at immutable commit
+`1aa0e962b206102b7c439a4a0c4981cfec6e87bc`:
+
+- `python/sglang/srt/models/deepseek_v4.py` and `deepseek_v2.py`
+- `python/sglang/srt/layers/engram.py`
+- `python/sglang/srt/layers/attention/dsv4/compressor.py`
+- `python/sglang/srt/layers/attention/dsv4/dsv41_sparse.py`
+- `python/sglang/srt/layers/attention/deepseek_v4_backend.py`
+- `python/sglang/srt/mem_cache/deepseek_v4_memory_pool.py`
+- `python/sglang/kernels/ops/attention/dsv4_attn_metadata_kernels.py`
+
+Source: https://github.com/sgl-project/sglang/tree/1aa0e962b206102b7c439a4a0c4981cfec6e87bc
+Copyright 2023-2024 SGLang Team and SGLang contributors. Licensed under Apache-2.0; its terms are
+reproduced in the repository `LICENSE`. These are analytical adaptations,
+not a copy of the model execution implementation. The modified analytical
+scoring/storage adaptations and their independently written regression cases
+also appear in `python/aisimulate/tests/unit/sdk/models/test_deepseek_v41.py`,
+Rust operator/spec unit tests, `docs/deepseek-v41.md`, and
+`docs/deepseek-v41-storage.md`. They distinguish candidate masking from scoring
+and physical FlashMLA cache payload from logical FP4 values.
+
 ## DeepSeek model configuration files
 
 The following model configuration files are copied from, or formatting-only
@@ -448,8 +475,16 @@ adaptations of, the named DeepSeek model repositories:
 | `src/aiconfigurator_core/model_configs/deepseek-ai--DeepSeek-V3_config.json` | `deepseek-ai/DeepSeek-V3@e815299b0bcbac849fa540c768ef21845365c9eb` |
 | `src/aiconfigurator_core/model_configs/deepseek-ai--DeepSeek-V3.2_config.json` | `deepseek-ai/DeepSeek-V3.2@c69397ecfd1fd142e90e3fbad51f4c7e40b9f3d3` |
 | `src/aiconfigurator_core/model_configs/deepseek-ai--DeepSeek-V4-Flash_config.json` | `deepseek-ai/DeepSeek-V4-Flash@60d8d70770c6776ff598c94bb586a859a38244f1` |
+| `src/aisimulate_core/model_configs/deepseek-ai--DeepSeek-V4.1-Flash_config.json` | `deepseek-ai/DeepSeek-V4.1-Flash@fb2764a5cf321eaa5070ca8f9e892818f477c16d` |
 | `src/aiconfigurator_core/model_configs/deepseek-ai--DeepSeek-V4-Pro_config.json` | `deepseek-ai/DeepSeek-V4-Pro@b5968e9190ef611bbf34a7229255be88a0e937c1` |
 
+The V4.1 descriptor and performance formulas in `src/aisimulate_core/sdk/deepseek_v41.py`,
+`src/aisimulate_core/sdk/models/deepseek_v41.py`, and repository-root
+`crates/core/src/perfmodel/operators/dsv41.rs` are AISimulate performance-model
+adaptations of the architecture described by `inference/model.py` and
+`DeepSeek_V41_Tech_Report.pdf` at the same V4.1 revision (modified; no model execution code).
+Source: https://huggingface.co/deepseek-ai/DeepSeek-V4.1-Flash/tree/fb2764a5cf321eaa5070ca8f9e892818f477c16d
+
 Upstream repositories:
 https://huggingface.co/deepseek-ai
 

crates/core/src/engine/cache/radix_cache.rs

@@ -43,6 +43,7 @@ impl KvPageId {
 /// Manages free / allocated pages for the simulated SGLang KV cache.
 ///
 /// SGLang's paged allocator owns and frees whole pages in production.
+#[derive(Clone)]
 pub struct PagePool {
     next_fresh: usize,
     free: Vec<KvPageId>,
@@ -157,6 +158,7 @@ impl PagePool {
 }
 
 /// A single node in the radix tree.
+#[derive(Clone)]
 pub struct TreeNode {
     /// Children keyed by the first complete page on the child edge.
     pub children: FxHashMap<LocalBlockHash, NodeId>,
@@ -195,6 +197,7 @@ pub struct RadixCache {
 }
 
 /// Optional forecast metadata; physical ownership and leaf eligibility stay in the radix tree.
+#[derive(Clone)]
 struct BeladyLeaves {
     oracle: BeladyOracle,
     cursor: usize,
@@ -205,6 +208,23 @@ struct BeladyLeaves {
 }
 
 impl RadixCache {
+    /// Snapshot allocator and radix metadata for a fallible admission transaction.
+    /// This is not a second owner of live request leases; only one cache state may commit.
+    pub(crate) fn admission_checkpoint(&self) -> Self {
+        Self {
+            nodes: self.nodes.clone(),
+            root: self.root,
+            page_pool: self.page_pool.clone(),
+            page_size: self.page_size,
+            #[cfg(test)]
+            test_now: self.test_now,
+            evictable_leaves: self.evictable_leaves.clone(),
+            belady: self.belady.clone(),
+            evictable_size: self.evictable_size,
+            protected_size: self.protected_size,
+        }
+    }
+
     pub fn new(total_tokens: usize, page_size: usize) -> Self {
         assert!(page_size >= 1, "page_size must be >= 1");
         let mut nodes = SlotMap::with_key();

crates/core/src/engine/common/perf_model.rs

@@ -42,6 +42,22 @@ impl std::fmt::Debug for PerfModel {
 }
 
 impl PerfModel {
+    pub(crate) fn prefill_batch_validation_can_fail(&self) -> bool {
+        match self {
+            Self::Polynomial => false,
+            Self::External { timing } => timing.prefill_batch_validation_can_fail(),
+        }
+    }
+
+    pub fn validate_prefill_batch(&self, requests: &[(usize, usize)]) -> Result<()> {
+        if let PerfModel::External { timing } = self {
+            timing
+                .validate_prefill_batch(requests)
+                .context("external prefill batch cannot be represented")?;
+        }
+        Ok(())
+    }
+
     /// Predict prefill time in milliseconds.
     ///
     /// Callers always pass all parameters; each variant uses what it needs:
@@ -101,17 +117,17 @@ impl PerfModel {
                 )
                 .context("external decode prediction failed")?,
         };
-        // Validate before the floor: f64::max would hide NaN or -infinity.
+        // Reject non-finite provider output before it reaches the scheduler.
         ensure!(
             time.is_finite(),
             "decode timing provider returned non-finite duration {time}ms"
         );
-        // Token-emitting decode steps should not collapse onto the same timestamp.
-        let result = time.max(1.0);
         tracing::trace!(
-            "Decode time prediction: batch_size={batch_size}, active_kv_tokens={active_kv_tokens}, context_length={context_length}, time={result:.2}ms"
+            "Decode time prediction: batch_size={batch_size}, active_kv_tokens={active_kv_tokens}, context_length={context_length}, time={time:.2}ms"
         );
-        Ok(result)
+        // The polynomial applies its own historical floor. External measurements
+        // can legitimately be below one millisecond and must retain that value.
+        Ok(time)
     }
 }
 
@@ -206,6 +222,46 @@ mod tests {
         assert_eq!(model.predict_decode_time(9, 0, 128, 0).unwrap(), 9.0);
     }
 
+    #[test]
+    fn external_decode_preserves_finite_values_and_rejects_non_finite() {
+        struct FixedDecode(f64);
+        impl TimingModel for FixedDecode {
+            fn predict_prefill_ms(&self, _: usize, _: usize, _: usize) -> anyhow::Result<f64> {
+                Ok(0.0)
+            }
+            fn predict_decode_ms(
+                &self,
+                _: usize,
+                _: usize,
+                _: usize,
+                _: usize,
+            ) -> anyhow::Result<f64> {
+                Ok(self.0)
+            }
+        }
+        // Finite values retain the provider's submillisecond resolution;
+        // modeled_duration_ms still rejects negative durations downstream.
+        for value in [0.6, 0.0, -1.0] {
+            let model = PerfModel::External {
+                timing: Arc::new(FixedDecode(value)),
+            };
+            assert_eq!(model.predict_decode_time(1, 128, 128, 1024).unwrap(), value);
+        }
+        // Main's finite-time boundary must reject NaN and both infinities before
+        // any duration normalization or scheduling can hide invalid output.
+        for value in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
+            let model = PerfModel::External {
+                timing: Arc::new(FixedDecode(value)),
+            };
+            let error = model.predict_decode_time(1, 128, 128, 1024).unwrap_err();
+            assert!(
+                error
+                    .to_string()
+                    .contains("decode timing provider returned non-finite duration")
+            );
+        }
+    }
+
     #[test]
     fn external_prefill_errors_propagate_with_context() {
         let error = PerfModel::External {

crates/core/src/engine/kv_manager/sglang_backend.rs

@@ -32,7 +32,38 @@ pub(crate) struct RadixRequestLease {
     last_node: Option<NodeId>,
 }
 
+/// Metadata used only with the matching KV-manager admission checkpoint.
+pub(crate) struct RadixLeaseCheckpoint {
+    pages: Vec<KvPageId>,
+    materialized_tokens: usize,
+    cached_tokens: usize,
+    admission_reused_tokens: usize,
+    page_hashes_len: usize,
+    last_node: Option<NodeId>,
+}
+
 impl RadixRequestLease {
+    pub(crate) fn admission_checkpoint(&self) -> RadixLeaseCheckpoint {
+        RadixLeaseCheckpoint {
+            pages: self.pages.clone(),
+            materialized_tokens: self.materialized_tokens,
+            cached_tokens: self.cached_tokens,
+            admission_reused_tokens: self.admission_reused_tokens,
+            page_hashes_len: self.page_hashes.len(),
+            last_node: self.last_node,
+        }
+    }
+
+    /// Restore only after the matching manager checkpoint has been rolled back.
+    pub(crate) fn restore_admission(&mut self, checkpoint: RadixLeaseCheckpoint) {
+        self.pages = checkpoint.pages;
+        self.materialized_tokens = checkpoint.materialized_tokens;
+        self.cached_tokens = checkpoint.cached_tokens;
+        self.admission_reused_tokens = checkpoint.admission_reused_tokens;
+        self.page_hashes.truncate(checkpoint.page_hashes_len);
+        self.last_node = checkpoint.last_node;
+    }
+
     #[cfg(test)]
     pub(crate) fn pages(&self) -> &[KvPageId] {
         &self.pages
@@ -135,6 +166,15 @@ pub struct SglangKvManager {
     /// block hash so router events reflect logical block visibility, not
     /// transient slot ownership.
     block_hash_refcounts: FxHashMap<SequenceHash, usize>,
+    pending_admission_events: Option<Vec<KvEvent>>,
+}
+
+#[must_use = "admission must be committed or rolled back"]
+pub(crate) struct SglangAdmissionCheckpoint {
+    cache: RadixCache,
+    next_event_id: u64,
+    page_to_block_hash: Vec<Option<SequenceHash>>,
+    block_hash_refcounts: FxHashMap<SequenceHash, usize>,
 }
 
 pub struct DecodeTokenReservation {
@@ -227,6 +267,7 @@ impl SglangKvManager {
             next_event_id: 0,
             page_to_block_hash,
             block_hash_refcounts: FxHashMap::default(),
+            pending_admission_events: None,
         }
     }
 
@@ -243,6 +284,50 @@ impl SglangKvManager {
         self.cache.set_belady_oracle(oracle);
     }
 
+    /// Hold speculative events until the timing provider accepts the admitted batch.
+    pub(crate) fn begin_admission(&mut self) -> SglangAdmissionCheckpoint {
+        assert!(
+            self.pending_admission_events.is_none(),
+            "nested KV admission"
+        );
+        self.pending_admission_events = Some(Vec::new());
+        SglangAdmissionCheckpoint {
+            cache: self.cache.admission_checkpoint(),
+            next_event_id: self.next_event_id,
+            page_to_block_hash: self.page_to_block_hash.clone(),
+            block_hash_refcounts: self.block_hash_refcounts.clone(),
+        }
+    }
+
+    pub(crate) fn commit_admission(&mut self, _checkpoint: SglangAdmissionCheckpoint) {
+        for event in self
+            .pending_admission_events
+            .take()
+            .expect("active KV admission")
+        {
+            self.publish_event(event);
+        }
+    }
+
+    pub(crate) fn rollback_admission(&mut self, checkpoint: SglangAdmissionCheckpoint) {
+        self.pending_admission_events
+            .take()
+            .expect("active KV admission");
+        self.cache = checkpoint.cache;
+        self.next_event_id = checkpoint.next_event_id;
+        self.page_to_block_hash = checkpoint.page_to_block_hash;
+        self.block_hash_refcounts = checkpoint.block_hash_refcounts;
+        self.log_trace("admission_rollback", 0);
+    }
+
+    fn publish_event(&mut self, event: KvEvent) {
+        if let Some(events) = self.pending_admission_events.as_mut() {
+            events.push(event);
+        } else if let Err(error) = self.kv_event_publishers.publish(event, None) {
+            tracing::warn!("Failed to publish SGLang KV event: {error}");
+        }
+    }
+
     #[cfg(test)]
     pub fn cache_mut(&mut self) -> &mut RadixCache {
         &mut self.cache
@@ -999,9 +1084,7 @@ impl SglangKvManager {
         };
         self.next_event_id += 1;
 
-        if let Err(e) = self.kv_event_publishers.publish(event, None) {
-            tracing::warn!("Failed to publish SGLang KV event: {e}");
-        }
+        self.publish_event(event);
 
         hashed_blocks
     }
@@ -1042,9 +1125,7 @@ impl SglangKvManager {
         };
         self.next_event_id += 1;
 
-        if let Err(e) = self.kv_event_publishers.publish(event, None) {
-            tracing::warn!("Failed to publish SGLang KV remove event: {e}");
-        }
+        self.publish_event(event);
     }
 }