Clear released snapshot-alias pages before their first KV use

Snapshot rows virtually alias KV pages. When the last snapshot owning a
page is released, the page returns to the free list still holding row
bytes. A request that later receives it as a partially filled tail page
reads those bytes past its last token; in FP8 KV some decode as NaN,
masked attention turns 0 x NaN into non-finite logits, and the commit
reports INVALID_SPECULATION (device error 6) under KV pressure.

The scheduler now marks a page when its last alias is released. When
commit first moves a marked page from a request's reservation into its
KV pages, the submission carries that page id and the mark is cleared.
Every executor zeroes the listed pages in each local cache plane with
cuMemsetD8Async on the model stream, after the input copies and before
the step's graph. Tensor-parallel PREPARE frames carry the list to every
rank (handshake layout 7), and peers validate it before READY. Pages are
not cleared at release, where inbound host restores could still be
writing, or at reservation, which may be returned unused. Inbound
restore pages are overwritten whole and never enter a clear list;
reservations, cancellations, and snapshot reuse keep the mark. The
change is independent of model and cache scheme.

Validation:
- Scheduler unit test: released alias pages are listed exactly once,
  when a request first commits them, and survive cancellation and
  snapshot reuse.
- TP wire tests: cleared pages round-trip through PREPARE, and ids
  outside the cache are rejected.
- CUDA test on an RTX PRO 6000 (SM120) using real VMM aliases across
  three planes: FP8 NaN bytes written through released rows land in
  freed KV pages and stay inside their alias ranges. clear_pages zeroes
  exactly the freed pages and leaves other pages and the live row
  byte-identical.
- Helios lib tests after rebasing onto 585159c: 1259 passed. The 9
  failures are the frontend, engram, and package-cache tests that fail on
  unmodified cc696a6 in the test environment; 585159c itself was not
  rerun.
- helios check with all targets and flare release build pass on
  585159c.
- Stock Qwen3.8 Flash-Next TP2 under KV pressure
  (--memory-limit 49.5G, 143 alternating 16k/32k prompts): stock
  passes 143/143. This model's QSA attention does not read partially
  filled FP8 tail pages, so the run does not reproduce #116.

Limitations: the #116 end-to-end reproduction on dense Qwen3.8-27B has
not been run. Exporting qwen3.8-27b at 6fa01a4 failed in "Warm compiled
forwards" (compiled forward has unbound static Inductor kernels). No
throughput comparison has been measured. Gemma 4 (#61) is untested.
The CUDA alias test and the TP2 run predate the rebase onto 585159c
and were not repeated.

Addresses #116.

Design-Signed-Off-By: Cursor Agent (Auto); design-review=a38f

Author: mark-yong

docs/runtime-package-design.md

@@ -605,6 +605,9 @@ and page capacities that all ranks can back.
 An owned snapshot excludes all conflicting page ids; overlapping exclusions
 are reference-counted. A snapshot fitting entirely in execution-allocation
 alignment padding excludes no pages. Execution rows never alias KV pages.
+A page whose last exclusion is released still holds snapshot row bytes. Before
+a request first writes it as KV, every rank clears that page's local cache
+planes on the model stream ahead of the step's graph.
 
 Admission reserves the request's KV pages. System, prompt, and terminal snapshots
 acquire backing at their capture boundaries, evicting unpinned checkpoints when

helios/src/cuda.rs

@@ -24,6 +24,7 @@ type HostFree = unsafe extern "C" fn(Handle) -> c_int;
 type Upload = unsafe extern "C" fn(u64, *const c_void, usize, Handle) -> c_int;
 type Download = unsafe extern "C" fn(Handle, u64, usize, Handle) -> c_int;
 type Download2D = unsafe extern "C" fn(*const Copy2D, Handle) -> c_int;
+type Fill = unsafe extern "C" fn(u64, u8, usize, Handle) -> c_int;
 type EventCreate = unsafe extern "C" fn(*mut Handle, c_uint) -> c_int;
 type EventRecord = unsafe extern "C" fn(Handle, Handle) -> c_int;
 type EventElapsed = unsafe extern "C" fn(*mut f32, Handle, Handle) -> c_int;
@@ -158,6 +159,7 @@ struct Driver {
     upload: Upload,
     download: Download,
     download_2d: Download2D,
+    fill: Fill,
     event_create: EventCreate,
     event_record: EventRecord,
     event_elapsed: EventElapsed,
@@ -222,6 +224,7 @@ impl Driver {
                 upload: symbol!("cuMemcpyHtoDAsync_v2"),
                 download: symbol!("cuMemcpyDtoHAsync_v2"),
                 download_2d: symbol!("cuMemcpy2DAsync_v2"),
+                fill: symbol!("cuMemsetD8Async"),
                 event_create: symbol!("cuEventCreate"),
                 event_record: symbol!("cuEventRecord"),
                 event_elapsed: symbol!("cuEventElapsedTime"),
@@ -910,6 +913,13 @@ impl Stream {
         })
     }
 
+    /// No queued or concurrent work may access the range until this stream passes the fill.
+    pub(crate) unsafe fn zero(&self, device: u64, nbytes: usize) -> Result<()> {
+        self.driver.check("cuMemsetD8Async", unsafe {
+            (self.driver.fill)(device, 0, nbytes, self.handle as Handle)
+        })
+    }
+
     pub(crate) fn synchronize(&self) -> Result<()> {
         self.driver.check("cuStreamSynchronize", unsafe {
             (self.driver.stream_synchronize)(self.handle as Handle)

helios/src/execution.rs

@@ -20,7 +20,7 @@ mod threaded;
 pub use rank_stream::RankStream;
 pub use ranks::RankPeers;
 mod transport;
-pub(crate) use transport::validate_speculative_survival;
+pub(crate) use transport::{clear_pages, validate_speculative_survival};
 mod worker;
 pub use parallel::{TensorParallelExecutor, serve_rank};
 pub(crate) use ranks::RankExecutor;

helios/src/execution/parallel.rs

@@ -13,7 +13,7 @@ mod wire;
 use std::{time::Instant, collections::VecDeque};
 
 use super::{ExecutionManifest, FrameCompletion, GraphExecutor, OwnedUpload, RankStream, Upload};
-use crate::{Error, Result, Shutdown, cache::Fence};
+use crate::{Error, Result, Shutdown, cache::{Fence, PageId}};
 use wire::*;
 
 struct Peer {
@@ -25,6 +25,7 @@ struct Pending {
     graph: usize,
     grammar: bool,
     uploads: Vec<OwnedUpload>,
+    cleared_pages: Vec<PageId>,
     dispatched: Instant,
     committed: bool,
     queued: bool,
@@ -222,6 +223,7 @@ impl TensorParallelExecutor {
         &mut self,
         graph: usize,
         uploads: &[Upload<'_>],
+        cleared_pages: &[PageId],
         grammar: bool,
     ) -> Result<Option<Fence>> {
         if let Some(error) = &self.poison {
@@ -233,7 +235,7 @@ impl TensorParallelExecutor {
         let barrier = grammar || !matches!(self.local.manifest().graphs.get(graph).map(|g| g.kind),
             Some(super::GraphKind::Decode | super::GraphKind::Speculative));
         if !self.can_submit(barrier) { return Ok(None); }
-        let Some(fence) = self.local.validate_submission(graph, uploads, grammar)? else {
+        let Some(fence) = self.local.validate_submission(graph, uploads, cleared_pages, grammar)? else {
             return Ok(None);
         };
         let queued = !self.pending.is_empty();
@@ -242,10 +244,11 @@ impl TensorParallelExecutor {
             uploads: uploads.iter().map(|u| OwnedUpload {
                 buffer: u.buffer.to_owned(), offset: u.offset, bytes: u.bytes.to_vec(),
             }).collect(),
+            cleared_pages: cleared_pages.to_vec(),
             ready: vec![false; self.peers.len()],
             completions: (0..=self.peers.len()).map(|_| None).collect(),
         });
-        let result = wire::prepare(self.local.manifest(), fence, graph, grammar, uploads, |tag, bytes| {
+        let result = wire::prepare(self.local.manifest(), fence, graph, grammar, uploads, cleared_pages, |tag, bytes| {
             for peer in &mut self.peers { peer.channel.send(tag, bytes)?; }
             Ok(())
         });
@@ -319,13 +322,14 @@ impl TensorParallelExecutor {
             if frame.committed { continue; }
             if !frame.ready.iter().all(|ready| *ready) { break; }
             let uploads = frame.uploads();
-            if self.local.validate_submission(frame.graph, &uploads, frame.grammar)? != Some(frame.fence) {
+            if self.local.validate_submission(frame.graph, &uploads, &frame.cleared_pages, frame.grammar)?
+                != Some(frame.fence) {
                 return Err(invalid("rank-zero reservation changed before commit"));
             }
             for peer in &mut self.peers {
                 peer.channel.send(COMMIT, &identity(frame.fence, frame.graph))?;
             }
-            if self.local.submit(frame.graph, &uploads, frame.grammar)? != Some(frame.fence) {
+            if self.local.submit(frame.graph, &uploads, &frame.cleared_pages, frame.grammar)? != Some(frame.fence) {
                 return Err(invalid("rank-zero submission changed after coordinated validation"));
             }
             frame.committed = true;
@@ -544,7 +548,7 @@ pub unsafe fn serve_rank(
             }
             executor.finish_startup()?;
             channel.send(STARTED, &[])?;
-            let mut prepared: Option<(Fence, usize, bool, Vec<OwnedUpload>)> = None;
+            let mut prepared: Option<(Fence, usize, bool, Vec<OwnedUpload>, Vec<PageId>)> = None;
             let mut prepared_at: Option<Instant> = None;
             let mut started = VecDeque::new();
             loop {
@@ -568,30 +572,30 @@ pub unsafe fn serve_rank(
                         }
                         PREPARE | PREPARE_MORE => {
                             if prepared.is_some() { return Err(invalid("rank already has a prepared frame")); }
-                            let (fence, graph, grammar, owned) =
+                            let (fence, graph, grammar, owned, cleared) =
                                 receive_prepare(&mut channel, executor.manifest(), tag, bytes)?;
                             let uploads: Vec<_> = owned.iter().map(|upload| Upload {
                                 buffer: &upload.buffer, offset: upload.offset, bytes: &upload.bytes,
                             }).collect();
-                            if executor.validate_submission(graph, &uploads, grammar)? != Some(fence) {
+                            if executor.validate_submission(graph, &uploads, &cleared, grammar)? != Some(fence) {
                                 return Err(invalid("rank submission sequence or capacity differs"));
                             }
                             channel.send(READY, &identity(fence, graph))?;
-                            prepared = Some((fence, graph, grammar, owned));
+                            prepared = Some((fence, graph, grammar, owned, cleared));
                             prepared_at = Some(Instant::now());
                         }
                         COMMIT => {
                             let (fence, graph) = read_identity(&bytes)?;
                             prepared_at = None;
-                            let (expected, expected_graph, grammar, owned) = prepared.take()
+                            let (expected, expected_graph, grammar, owned, cleared) = prepared.take()
                                 .ok_or_else(|| invalid("rank commit arrived without preparation"))?;
                             if fence != expected || graph != expected_graph {
                                 return Err(invalid("rank commit identity differs"));
                             }
                             let uploads: Vec<_> = owned.iter().map(|upload| Upload {
                                 buffer: &upload.buffer, offset: upload.offset, bytes: &upload.bytes,
                             }).collect();
-                            if executor.submit(graph, &uploads, grammar)? != Some(fence) {
+                            if executor.submit(graph, &uploads, &cleared, grammar)? != Some(fence) {
                                 return Err(invalid("rank lost its prepared submission"));
                             }
                             started.push_back(Instant::now());

helios/src/execution/parallel/tests.rs

@@ -226,7 +226,7 @@ fn prepare_packets_preserve_large_and_partial_uploads_on_both_transports() {
     manifest.graphs[0].input_bindings[2].nbytes = buffer.nbytes;
     manifest.input_frame_bytes = 16 + buffer.nbytes;
     manifest.validate().unwrap();
-    assert_eq!(u64::from_le_bytes(layout(&manifest)[..8].try_into().unwrap()), 6);
+    assert_eq!(u64::from_le_bytes(layout(&manifest)[..8].try_into().unwrap()), 7);
     for (sender, receiver) in [crate::execution::rank_stream::test_pair().unwrap(),
                                (client.into(), server.into())] {
         let mut sender = Channel::new(sender).unwrap();
@@ -239,7 +239,7 @@ fn prepare_packets_preserve_large_and_partial_uploads_on_both_transports() {
                             buffer: "tokens", offset: 8 + index * 80 * 1024 * 1024 / parts, bytes,
                         }).collect();
                     let mut packets = 0;
-                    wire::prepare(&manifest, Fence(7), 0, true, &uploads, |tag, data| {
+                    wire::prepare(&manifest, Fence(7), 0, true, &uploads, &[], |tag, data| {
                         assert!(data.len() <= 64 * 1024 * 1024);
                         assert_eq!(tag, if packets == 0 { PREPARE_MORE } else { PREPARE });
                         packets += 1;
@@ -249,9 +249,10 @@ fn prepare_packets_preserve_large_and_partial_uploads_on_both_transports() {
                     sender.send(DRAIN, &[]).unwrap();
                 });
                 let (tag, packet) = receiver.wait(false).unwrap();
-                let (fence, graph, grammar, owned) =
+                let (fence, graph, grammar, owned, cleared) =
                     receive_prepare(&mut receiver, &manifest, tag, packet).unwrap();
                 assert_eq!((fence, graph, grammar), (Fence(7), 0, true));
+                assert!(cleared.is_empty());
                 assert!(owned.len() > parts);
                 let mut offset = 8;
                 for upload in &owned {
@@ -281,16 +282,53 @@ fn small_prepare_preserves_one_packet_and_upload_extents() {
     let (sender, receiver) = crate::execution::rank_stream::test_pair().unwrap();
     let mut sender = Channel::new(sender).unwrap();
     let mut receiver = Channel::new(receiver).unwrap();
-    wire::prepare(&manifest, Fence(3), 0, false, &input, |tag, data| {
+    wire::prepare(&manifest, Fence(3), 0, false, &input, &[], |tag, data| {
         assert_eq!(tag, PREPARE);
         sender.send(tag, data)
     }).unwrap();
     let (tag, packet) = receiver.wait(false).unwrap();
-    let (fence, graph, grammar, owned) = receive_prepare(&mut receiver, &manifest, tag, packet).unwrap();
+    let (fence, graph, grammar, owned, cleared) = receive_prepare(&mut receiver, &manifest, tag, packet).unwrap();
     assert_eq!((fence, graph, grammar), (Fence(3), 0, false));
     assert_eq!(owned.len(), 1);
     assert_eq!(owned[0].offset, 8);
     assert_eq!(owned[0].bytes, bytes);
+    assert!(cleared.is_empty());
+}
+
+#[test]
+fn prepare_carries_cleared_pages_to_peers_within_the_cache() {
+    use crate::cache::PageId;
+    let mut manifest = super::super::tests::manifest();
+    let plane = |name: &str| flare_selene::Plane { name: name.into(), address: 0, unit_bytes: 256 };
+    manifest.cache = Some(flare_selene::CacheGeometry {
+        page_tokens: 64,
+        page_capacity: 16,
+        row_capacity: 2,
+        row: plane("rows"),
+        planes: vec![plane("keys"), plane("values")],
+        snapshot_pages: vec![8..16],
+    });
+    let bytes = 7_i64.to_le_bytes();
+    let input = [Upload { buffer: "tokens", offset: 8, bytes: &bytes }];
+    let (sender, receiver) = crate::execution::rank_stream::test_pair().unwrap();
+    let mut sender = Channel::new(sender).unwrap();
+    let mut receiver = Channel::new(receiver).unwrap();
+    let pages = [PageId(9), PageId(10), PageId(15)];
+    wire::prepare(&manifest, Fence(4), 0, false, &input, &pages, |tag, data| {
+        assert_eq!(tag, PREPARE);
+        sender.send(tag, data)
+    }).unwrap();
+    let (tag, packet) = receiver.wait(false).unwrap();
+    let (fence, _, _, owned, cleared) = receive_prepare(&mut receiver, &manifest, tag, packet).unwrap();
+    assert_eq!(fence, Fence(4));
+    assert_eq!(owned.len(), 1);
+    assert_eq!(cleared, pages);
+    wire::prepare(&manifest, Fence(5), 0, false, &[], &[PageId(16)], |tag, data| {
+        sender.send(tag, data)
+    }).unwrap();
+    let (tag, packet) = receiver.wait(false).unwrap();
+    let error = receive_prepare(&mut receiver, &manifest, tag, packet).err().unwrap().to_string();
+    assert!(error.contains("outside the cache"), "{error}");
 }
 
 #[test]
@@ -543,7 +581,7 @@ fn logprob_uploads_budget_count_and_prefix_wire_bytes_separately() {
     let manifest = super::super::tests::with_logprobs(crate::execution::MAX_LOGPROB_TOKEN_IDS);
     let measure = |uploads: &[Upload<'_>]| {
         let mut size = 0;
-        wire::prepare(&manifest, Fence(1), 0, false, uploads, |_, bytes| {
+        wire::prepare(&manifest, Fence(1), 0, false, uploads, &[], |_, bytes| {
             size += bytes.len();
             Ok(())
         })
@@ -587,7 +625,7 @@ fn two_frame_window_retains_preparing_and_unconsumed_completions() {
     let mut manifest = crate::execution::tests::manifest();
     manifest.graphs[0].kind = crate::execution::GraphKind::Decode;
     let frame = |sequence| Pending {
-        fence: Fence(sequence), graph: 0, grammar: false, uploads: Vec::new(),
+        fence: Fence(sequence), graph: 0, grammar: false, uploads: Vec::new(), cleared_pages: Vec::new(),
         dispatched: Instant::now(), committed: false, queued: sequence > 1,
         ready: vec![false; 3], completions: (0..4).map(|_| None).collect(),
     };

helios/src/execution/parallel/wire.rs

@@ -7,7 +7,7 @@
 use std::io::{ErrorKind, Read, Write};
 use std::time::Duration;
 
-use crate::cache::Fence;
+use crate::cache::{Fence, PageId};
 use crate::execution::{ExecutionManifest, FrameCompletion, OwnedUpload, RankStream, Upload};
 use crate::{Error, ExecutionError, Failure, Result};
 
@@ -183,7 +183,7 @@ pub(super) fn validate_completion_packet(manifest: &ExecutionManifest) -> Result
 pub(super) fn layout(manifest: &ExecutionManifest) -> Vec<u8> {
     let mut bytes = Vec::new();
     for value in [
-        6,
+        7,
         manifest.completion_token_capacity,
         manifest.max_active,
         manifest.input_frame_bytes,
@@ -253,15 +253,21 @@ pub(super) fn layout(manifest: &ExecutionManifest) -> Vec<u8> {
 }
 
 /// Split validated uploads into bounded packets while preserving element alignment.
+/// The terminal packet carries the cleared page ids.
 pub(super) fn prepare(
     manifest: &ExecutionManifest,
     fence: Fence,
     graph: usize,
     grammar: bool,
     uploads: &[Upload<'_>],
+    cleared_pages: &[PageId],
     mut send: impl FnMut(u8, &[u8]) -> Result<()>,
 ) -> Result<()> {
-    let capacity = uploads.iter().fold(32usize, |size, upload| {
+    let clears = 8 * (1 + cleared_pages.len());
+    if 32 + clears > MAX_PACKET {
+        return Err(invalid("cleared pages exceed the rank control packet"));
+    }
+    let capacity = uploads.iter().fold(32 + clears, |size, upload| {
         size.saturating_add(24).saturating_add(upload.bytes.len())
     }).min(MAX_PACKET);
     let mut packet = Vec::with_capacity(capacity);
@@ -299,6 +305,13 @@ pub(super) fn prepare(
             bytes = &bytes[size..];
         }
     }
+    if packet.len() + clears > MAX_PACKET {
+        flush(&mut packet, &mut count, PREPARE_MORE)?;
+    }
+    number(&mut packet, cleared_pages.len());
+    for page in cleared_pages {
+        number(&mut packet, page.0 as usize);
+    }
     flush(&mut packet, &mut count, PREPARE)
 }
 
@@ -307,7 +320,7 @@ pub(super) fn receive_prepare(
     manifest: &ExecutionManifest,
     mut tag: u8,
     mut bytes: Vec<u8>,
-) -> Result<(Fence, usize, bool, Vec<OwnedUpload>)> {
+) -> Result<(Fence, usize, bool, Vec<OwnedUpload>, Vec<PageId>)> {
     let mut identity = None;
     let mut remaining = manifest.input_frame_bytes;
     let mut owned = Vec::new();
@@ -329,14 +342,28 @@ pub(super) fn receive_prepare(
         }
         identity = Some(received);
         let part = uploads(manifest, &mut reader, &mut remaining)?;
-        reader.end()?;
         if tag == PREPARE_MORE && part.is_empty() {
             return Err(invalid("empty rank prepare continuation"));
         }
         owned.extend(part);
         if tag == PREPARE {
-            return Ok((fence, graph, grammar, owned));
+            let count = reader.number()?;
+            let capacity = manifest.cache.as_ref().map_or(0, |cache| cache.page_capacity);
+            if count > reader.0.len() / 8 {
+                return Err(invalid("invalid rank cleared page count"));
+            }
+            let mut cleared = Vec::with_capacity(count);
+            for _ in 0..count {
+                let page = reader.number()?;
+                if page as u64 >= capacity {
+                    return Err(invalid("rank cleared page id is outside the cache"));
+                }
+                cleared.push(PageId(page as u32));
+            }
+            reader.end()?;
+            return Ok((fence, graph, grammar, owned, cleared));
         }
+        reader.end()?;
         (tag, bytes) = channel.wait(false)?;
     }
 }

helios/src/execution/ranks.rs

@@ -9,7 +9,7 @@
 use super::{
     ExecutionManifest, FrameCompletion, GraphExecutor, RankStream, TensorParallelExecutor, Upload,
 };
-use crate::{Result, Shutdown, cache::Fence};
+use crate::{Result, Shutdown, cache::{Fence, PageId}};
 
 pub enum RankPeers {
     Processes(Vec<RankStream>),
@@ -48,12 +48,13 @@ impl RankExecutor {
         &mut self,
         graph: usize,
         uploads: &[Upload<'_>],
+        cleared_pages: &[PageId],
         grammar: bool,
     ) -> Result<Option<Fence>> {
         match self {
-            Self::Local(executor) => executor.submit(graph, uploads, grammar),
-            Self::Parallel(executor) => executor.submit(graph, uploads, grammar),
-            Self::Threaded(executor) => executor.submit(graph, uploads, grammar),
+            Self::Local(executor) => executor.submit(graph, uploads, cleared_pages, grammar),
+            Self::Parallel(executor) => executor.submit(graph, uploads, cleared_pages, grammar),
+            Self::Threaded(executor) => executor.submit(graph, uploads, cleared_pages, grammar),
         }
     }