Bound GPU snapshot rows by memory, not a fixed count

Snapshot capacity was fixed at eight rows per concurrent request (256 at
concurrency 32) regardless of memory. On a long agent conversation the
GPU therefore evicted 94% of the checkpoints it created while 55-73% of
its KV pool sat free, and the host store refilled what the GPU dropped.

Capacity is now whatever the borrowable KV backing can alias, which the
planner already computes and caps by memory and virtual extent; rows and
request pages keep yielding to each other through the existing pressure
eviction. The per-request capture budget stays at most eight, so prefill
cuts are unchanged. No package re-export is needed. The design docs (core design, runtime
package, CLI) describe the memory-bounded capacity.

Validation: upstream CI's Linux commands: helios the same 1190 passed /
7 failed as main (main's DeepSeek tokenizer-fixture defect), selene and
selene-manager all passed. Tests: capacity grows with memory, the
per-request budget stays eight, cached rows yield backing to admission.
CPU measurements at 256 vs 1,350 rows: blocked snapshot search 7.2 vs
80.5 us, median leaf eviction 2.0 vs 13.6 us, reclaiming 128 rows 0.49
vs 1.98 ms. Not yet run on a GPU.

Design-Signed-Off-By: yatesdr; design-review=1c01

Author: yatesdr

docs/cli-design.md

@@ -1117,9 +1117,9 @@ value in preference to that recorded reserve.
 
 The GPU prefix cache is derived. It holds model states at instruction,
 prompt, and answer boundaries so that later requests repeating a prefix skip
-recomputing it. A finished exchange can retain its prompt and answer states, so the
-engine exposes up to eight snapshot slots per concurrent request, about four
-recent exchanges each. Snapshot slots share physical backing with KV pages and
+recomputing it. Immutable snapshot capacity is bounded by borrowable KV backing
+and the prepared virtual extent. The per-request capture budget is capped at
+eight. Snapshot slots share physical backing with KV pages and
 acquire it only when retaining a checkpoint. Unpinned checkpoints are evicted
 to admit requests. If active requests or transfers own all suitable backing,
 the engine skips optional checkpoint retention. There is no dedicated snapshot
@@ -1189,7 +1189,7 @@ every rank and have no configuration-file key. See
 | Tensor-parallel size | Count of `--gpus`, or the smallest recipe-supported size whose estimate fits the first GPUs with enough free memory |
 | Context limit | The smaller of the prepared context ceiling and loaded KV capacity after per-request draft-page storage; admission also checks available shared capacity |
 | KV pages | Remaining supply after the runtime memory limit or package reserve, engine headroom, weights, graphs, scratch, and execution rows; bounded by `--kv-cache-limit` and prepared virtual extents. Retained snapshot rows borrow page backing; the load display reports total KV capacity |
-| Prefix cache | Up to eight snapshot slots per `--max-concurrency` request, acquired on demand from shared KV backing and evicted under pressure |
+| Prefix cache | Immutable snapshot capacity bounded by borrowable KV backing and the prepared virtual extent, acquired on demand and evicted under pressure; per-request capture budget capped at eight |
 | Live state | One null row plus the recipe's execution rows per active request slot; GLM verification commits into one live row |
 | Decode graph capacities | 1 through `--max-concurrency`, rounded to the recipe's captured set |
 | Image feature arena | Prepared image concurrency times per-request feature capacity and embedding width; slots remain owned through request retirement |

docs/core-design.md

@@ -1395,10 +1395,10 @@ to earlier tokens cannot reuse a checkpoint for different tokens.
 
 At admission, the scheduler budgets newly captured cache snapshots from the
 loaded snapshot capacity divided by configured maximum concurrency, rounded
-down. A nonempty pool permits at least one capture per request; an empty pool
-permits none. The budget reserves a terminal capture and, when generation can
-continue beyond the prompt and the budget permits, a prompt capture. A single
-output token shares the prompt and terminal endpoint.
+down and capped at eight. A nonempty pool permits at least one capture per
+request; an empty pool permits none. The budget reserves a terminal capture and,
+when generation can continue beyond the prompt and the budget permits, a prompt
+capture. A single output token shares the prompt and terminal endpoint.
 
 The remaining budget selects message checkpoints within the uncached prompt
 suffix. If all eligible boundaries fit, retain them all. Otherwise, place token

docs/prefix-cache-offload.md

@@ -179,6 +179,8 @@ message is offered while the request still pins it; cancellation or error before
 that capture offers it at retirement. GPU captures, prefill cuts, and retention
 are unchanged; deferred points can be lost to GPU
 eviction before reuse, and are not announced to the pool until offered.
+Compaction also offers a deferred branch point as its heir and refreshes
+descendant offers with their new parent.
 `Queued` holds one GPU pin until `Stored` or `Deferred`; `Present`
 requires a committed local copy. `Retry` accepts no copy. The worker retains only
 checkpoint IDs for retries after store progress and discards IDs that have left

docs/runtime-package-design.md

@@ -584,8 +584,8 @@ the recipe's execution rows per request. KV pages receive the remaining budget,
 within their virtual maximum. The planner accounts for mapping alignment and
 requires at least 8,192 cached tokens per active request.
 
-Up to eight immutable snapshot slots per active request share physical backing
-with KV pages. At load, CUDA virtual mappings expose the same allocation chunks
+Immutable snapshot capacity is bounded by borrowable KV backing and the prepared
+virtual extent. At load, CUDA virtual mappings expose the same allocation chunks
 through the page planes and the snapshot part of the continuation range. Small
 interleaved runs of dense page planes keep each snapshot's conflicting page range
 close to its byte cost. Planes whose allocation granules span more pages than a

helios/src/runtime_package/memory.rs

@@ -294,7 +294,7 @@ fn capacities(
     if minimum_pages == 0 {
         return Err(invalid("KV cache limit leaves no physical pages"));
     }
-    let mut cached = active * 8;
+    let mut cached = u64::MAX;
     let mut maximum_pages = page_limit.map_or(u64::MAX, |limit| limit.maximum);
     for (rank, (package, &budget)) in packages.iter().zip(available).enumerate() {
         let rows: Vec<_> = package
@@ -1343,14 +1343,29 @@ mod tests {
         }
     }
 
+    #[test]
+    fn snapshot_capacity_grows_with_borrowable_kv_backing() {
+        let packages = [package(65536, 65536, 100000)];
+        let mut previous = 0;
+        for budget in [64 << 20, 128 << 20, 512 << 20] {
+            let (rows, pages, snapshots) =
+                capacities(&packages, &[budget], 4, 1, 64, None).unwrap();
+            assert_eq!(rows, 5 + snapshots);
+            assert_eq!(snapshots, pages.min(4096 - 5));
+            assert!(snapshots > previous.max(4 * 8));
+            previous = snapshots;
+        }
+    }
+
     #[test]
     fn lower_concurrency_and_verification_move_physical_budget_to_kv() {
         let p = [package(100 << 20, 2 << 20, 100000)];
         let budget = [80 << 30];
         let full = capacities(&p, &budget, 32, 6, 64, None).unwrap();
         let reduced = capacities(&p, &budget, 4, 6, 64, None).unwrap();
         let target_only = capacities(&p, &budget, 4, 1, 64, None).unwrap();
-        assert!(full.0 > reduced.0 && reduced.0 > target_only.0);
+        assert!(full.0 - full.2 > reduced.0 - reduced.2);
+        assert!(reduced.0 - reduced.2 > target_only.0 - target_only.2);
         assert!(full.1 < reduced.1 && reduced.1 < target_only.1);
         assert_eq!(reduced.0, 1 + 4 * 6 + reduced.2);
         for (rows, pages, snapshots) in [full, reduced, target_only] {

helios/src/scheduler.rs

@@ -1569,7 +1569,7 @@ impl RequestScheduler {
         let checkpoint_budget = if self.limits.snapshots == 0 {
             0
         } else {
-            (self.limits.snapshots as usize / self.limits.max_active).max(1)
+            (self.limits.snapshots as usize / self.limits.max_active).clamp(1, 8)
         };
         let endpoints = 1 + usize::from(spec.max_new_tokens > 1 && consumed < spec.prompt.len() as u64);
         let message_boundaries = checkpoint_boundaries(

helios/src/scheduler/boundary_tests.rs

@@ -149,6 +149,27 @@ fn checkpoint_targets_follow_token_distance_and_exclude_reused_prefixes() {
     assert!(checkpoint_boundaries(vec![10], 0, 100, 0).is_empty());
 }
 
+#[test]
+fn larger_snapshot_pools_keep_the_same_eight_capture_prefill_cuts() {
+    let mut previous = None;
+    for snapshots in [32, 1024] {
+        let mut s = scheduler(snapshots, false);
+        let mut request = spec(None, 700, 3);
+        request.message_boundaries = (1..700).collect();
+        let handle = s.admit(request).unwrap();
+        let boundaries = s.request(handle).unwrap().message_boundaries.clone();
+        assert_eq!(boundaries.len(), 6);
+        if let Some(previous) = previous {
+            assert_eq!(boundaries, previous);
+        }
+        previous = Some(boundaries);
+        finish(&mut s, handle);
+        assert_eq!(s.stats().message_checkpoints, 6);
+        assert_eq!(s.stats().prompt_checkpoints, 1);
+        assert_eq!(s.stats().snapshot_publications, 8);
+    }
+}
+
 #[test]
 fn small_snapshot_shares_prioritize_endpoints_and_allow_completion() {
     for snapshots in [0, 1, 2, 4, 8, 12] {
@@ -544,30 +565,32 @@ fn every_message_endpoint_can_restore_after_a_changed_following_turn() {
     let endpoints = vec![1, 7, 63, 64, 65, 129, 300, 514, 700];
     for page_tokens in [64, 128, 256, 2048] {
         for mtp in [false, true] {
-            let mut s = scheduler_with_concurrency(16, mtp, page_tokens, 1);
-            let mut request = spec(None, 901, 3);
-            request.message_boundaries = endpoints.clone();
-            let handle = s.admit(request).unwrap();
-            for &end in &endpoints {
-                stop_at_capture(&mut s, handle, end as u64);
-                assert_eq!(s.request(handle).unwrap().generated_tokens, 0);
-                run_step(&mut s);
-                assert_eq!(s.request(handle).unwrap().prefix_tokens, end as u64);
-            }
-            finish(&mut s, handle);
-            assert_eq!(s.stats().message_checkpoints, endpoints.len() as u64);
-            for &end in &endpoints {
-                let mut branch = spec(None, end + 10, 1);
-                branch.prompt[end..].fill(1000);
-                branch.message_boundaries = endpoints.iter().copied().filter(|&n| n <= end).collect();
-                let branch = s.admit(branch).unwrap();
-                assert_eq!(s.request(branch).unwrap().reused_prompt_tokens(), end as u64);
-                assert_eq!(run_step(&mut s).kind, GraphKind::Restore);
-                let step = s.plan().unwrap();
-                assert_eq!(step.kind, GraphKind::Prefill);
-                assert_eq!(step.lanes[0].start_position, end as u64);
-                assert_eq!(step.lanes[0].tokens, 10);
-                s.cancel(branch).unwrap();
+            for endpoints in endpoints.chunks(6) {
+                let mut s = scheduler_with_concurrency(16, mtp, page_tokens, 1);
+                let mut request = spec(None, 901, 3);
+                request.message_boundaries = endpoints.to_vec();
+                let handle = s.admit(request).unwrap();
+                for &end in endpoints {
+                    stop_at_capture(&mut s, handle, end as u64);
+                    assert_eq!(s.request(handle).unwrap().generated_tokens, 0);
+                    run_step(&mut s);
+                    assert_eq!(s.request(handle).unwrap().prefix_tokens, end as u64);
+                }
+                finish(&mut s, handle);
+                assert_eq!(s.stats().message_checkpoints, endpoints.len() as u64);
+                for &end in endpoints {
+                    let mut branch = spec(None, end + 10, 1);
+                    branch.prompt[end..].fill(1000);
+                    branch.message_boundaries = endpoints.iter().copied().filter(|&n| n <= end).collect();
+                    let branch = s.admit(branch).unwrap();
+                    assert_eq!(s.request(branch).unwrap().reused_prompt_tokens(), end as u64);
+                    assert_eq!(run_step(&mut s).kind, GraphKind::Restore);
+                    let step = s.plan().unwrap();
+                    assert_eq!(step.kind, GraphKind::Prefill);
+                    assert_eq!(step.lanes[0].start_position, end as u64);
+                    assert_eq!(step.lanes[0].tokens, 10);
+                    s.cancel(branch).unwrap();
+                }
             }
         }
     }

helios/src/scheduler/sharing/tests.rs

@@ -46,6 +46,28 @@ fn publish(
         .id()
 }
 
+#[test]
+fn cached_rows_beyond_eight_per_request_yield_backing_to_admission() {
+    let mut s = scheduler(1, 64, 64);
+    let aliases: Vec<_> = (0..64).map(|i| 63 - i..64 - i).collect();
+    s.share_snapshot_pages(&aliases).unwrap();
+    let h = history(1);
+    for end in 1..=48 {
+        publish(&mut s, &h, None, end, &[], 0);
+    }
+    assert_eq!(s.stats().snapshot_backing_pages, 48);
+    assert_eq!(s.free_page_count(), 16);
+    let mut request = spec(1984, 1);
+    request.prompt.fill(999);
+    let handle = s.admit(request).unwrap();
+    assert_eq!(s.request(handle).unwrap().reused_prompt_tokens(), 0);
+    assert_eq!(s.request(handle).unwrap().reserved_pages.len(), 32);
+    assert_eq!(s.stats().snapshot_backing_pages, 32);
+    assert_eq!(s.stats().snapshot_evictions, 16);
+    s.cancel(handle).unwrap();
+    assert_eq!(s.free_page_count(), 32);
+}
+
 #[test]
 fn compaction_and_leaf_eviction_are_reported_with_the_heir() {
     let mut s = scheduler(2, 16, 4);