use std::collections::VecDeque;
#[derive(Clone, Debug, PartialEq, Default)]
pub enum KvRetentionPolicy {
#[default]
Full,
Sinks { sinks: usize, window: usize },
HeavyHitter {
sinks: usize,
recent: usize,
budget: usize,
},
Retrieval {
block: usize,
resident_blocks: usize,
sinks: usize,
recent: usize,
},
Auto { max_resident: usize },
}
#[derive(Clone, Debug)]
struct Slot {
abs_pos: usize,
attn_mass: f32,
last_used: u64,
#[allow(dead_code)] block: usize,
}
#[derive(Clone, Debug)]
pub struct StoredBlock {
pub id: usize,
pub start_pos: usize,
pub rows: usize,
pub key: Vec<f32>,
pub k: Vec<Vec<f32>>,
pub v: Vec<Vec<f32>>,
}
#[derive(Clone, Debug, Default)]
pub struct RetentionPlan {
pub keep: Vec<usize>,
pub evict: Vec<usize>,
pub retrieve: Vec<usize>,
pub store_evicted: bool,
}
impl RetentionPlan {
fn keep_all(n: usize) -> Self {
RetentionPlan {
keep: (0..n).collect(),
evict: Vec::new(),
retrieve: Vec::new(),
store_evicted: false,
}
}
pub fn is_noop(&self) -> bool {
self.evict.is_empty() && self.retrieve.is_empty()
}
}
#[derive(Clone, Debug)]
pub struct KvRetentionManager {
policy: KvRetentionPolicy,
kv_dim: usize,
resident: Vec<Slot>,
store: Vec<StoredBlock>,
next_block_id: usize,
step: u64,
concentration: f32,
next_pos: usize,
recorder: Option<crate::kv_metrics::RetentionRecorder>,
}
const MASS_DECAY: f32 = 0.98;
impl KvRetentionManager {
pub fn new(policy: KvRetentionPolicy, kv_dim: usize) -> Self {
KvRetentionManager {
policy,
kv_dim,
resident: Vec::new(),
store: Vec::new(),
next_block_id: 0,
step: 0,
concentration: 0.0,
next_pos: 0,
recorder: None,
}
}
pub fn policy(&self) -> &KvRetentionPolicy {
&self.policy
}
pub fn enable_recording(&mut self) {
if self.recorder.is_none() {
self.recorder = Some(crate::kv_metrics::RetentionRecorder::new());
}
}
pub fn recorder(&self) -> Option<&crate::kv_metrics::RetentionRecorder> {
self.recorder.as_ref()
}
pub fn recorder_mut(&mut self) -> Option<&mut crate::kv_metrics::RetentionRecorder> {
self.recorder.as_mut()
}
pub fn take_recorder(&mut self) -> Option<crate::kv_metrics::RetentionRecorder> {
self.recorder.take()
}
pub fn selection_snapshot(&self) -> (Vec<(usize, f32, u64)>, f32) {
let sel = self
.resident
.iter()
.map(|s| (s.abs_pos, s.attn_mass, s.last_used))
.collect();
(sel, self.concentration)
}
pub fn resident_len(&self) -> usize {
self.resident.len()
}
pub fn needs_attention(&self) -> bool {
matches!(
self.policy,
KvRetentionPolicy::HeavyHitter { .. } | KvRetentionPolicy::Auto { .. }
)
}
pub fn stored_blocks(&self) -> usize {
self.store.len()
}
pub fn stored_tokens(&self) -> usize {
self.store.iter().map(|b| b.rows).sum()
}
pub fn on_prefill(&mut self, n: usize) {
self.resident.clear();
self.store.clear();
self.resident.reserve(n);
let block_sz = self.block_size().max(1);
for pos in 0..n {
self.resident.push(Slot {
abs_pos: pos,
attn_mass: 0.0,
last_used: 0,
block: pos / block_sz,
});
}
self.next_pos = n;
self.next_block_id = self.next_block_id.max(n.div_ceil(block_sz));
}
pub fn append(&mut self) {
let block_sz = self.block_size().max(1);
let pos = self.next_pos;
self.resident.push(Slot {
abs_pos: pos,
attn_mass: 0.0,
last_used: self.step,
block: pos / block_sz,
});
self.next_pos += 1;
}
pub fn observe_attention(&mut self, weights: &[f32]) {
let n = self.resident.len().min(weights.len());
for s in self.resident.iter_mut() {
s.attn_mass *= MASS_DECAY;
}
let mut total = 0.0f32;
let mut peak = 0.0f32;
for i in 0..n {
let w = weights[i].max(0.0);
self.resident[i].attn_mass += w;
if w > 1e-4 {
self.resident[i].last_used = self.step;
}
total += w;
if w > peak {
peak = w;
}
}
if total > 0.0 {
let c = peak / total;
self.concentration = 0.9 * self.concentration + 0.1 * c;
}
self.step += 1;
}
fn block_size(&self) -> usize {
match self.policy {
KvRetentionPolicy::Retrieval { block, .. } => block,
KvRetentionPolicy::Auto { max_resident } => (max_resident / 8).max(16),
_ => 0,
}
}
fn effective_policy(&self) -> KvRetentionPolicy {
match self.policy {
KvRetentionPolicy::Auto { max_resident } => {
let ctx = self.resident.len() + self.stored_tokens();
if ctx <= max_resident {
KvRetentionPolicy::Full
} else if self.concentration >= 0.15 {
let sinks = 4.min(max_resident / 8);
let recent = max_resident / 2;
let budget = max_resident.saturating_sub(sinks + recent);
KvRetentionPolicy::HeavyHitter {
sinks,
recent,
budget,
}
} else {
let block = (max_resident / 8).max(16);
let sinks = 4.min(max_resident / 8);
let recent = max_resident / 2;
let resident_blocks =
max_resident.saturating_sub(sinks + recent) / block.max(1);
KvRetentionPolicy::Retrieval {
block,
resident_blocks: resident_blocks.max(1),
sinks,
recent,
}
}
}
ref p => p.clone(),
}
}
pub fn plan(&mut self, query_key: Option<&[f32]>) -> RetentionPlan {
let n = self.resident.len();
match self.effective_policy() {
KvRetentionPolicy::Full => RetentionPlan::keep_all(n),
KvRetentionPolicy::Sinks { sinks, window } => {
self.plan_keep_set(sinks, window, 0, |_| 0.0, false, None)
}
KvRetentionPolicy::HeavyHitter {
sinks,
recent,
budget,
} => {
let masses: Vec<f32> = self.resident.iter().map(|s| s.attn_mass).collect();
self.plan_keep_set(sinks, recent, budget, |i| masses[i], false, None)
}
KvRetentionPolicy::Retrieval {
block,
resident_blocks,
sinks,
recent,
} => self.plan_retrieval(block, resident_blocks, sinks, recent, query_key),
KvRetentionPolicy::Auto { .. } => unreachable!("resolved by effective_policy"),
}
}
fn plan_keep_set(
&self,
sinks: usize,
recent: usize,
budget: usize,
score: impl Fn(usize) -> f32,
store_evicted: bool,
retrieve: Option<Vec<usize>>,
) -> RetentionPlan {
let n = self.resident.len();
let sinks = sinks.min(n);
let recent = recent.min(n.saturating_sub(sinks));
let mid_lo = sinks;
let mid_hi = n - recent;
let mut mid: Vec<usize> = (mid_lo..mid_hi).collect();
if budget < mid.len() {
mid.sort_by(|&a, &b| {
score(b)
.partial_cmp(&score(a))
.unwrap_or(std::cmp::Ordering::Equal)
});
mid.truncate(budget);
}
let mut keep_flag = vec![false; n];
for i in 0..sinks {
keep_flag[i] = true;
}
for i in mid_hi..n {
keep_flag[i] = true;
}
for &i in &mid {
keep_flag[i] = true;
}
let keep: Vec<usize> = (0..n).filter(|&i| keep_flag[i]).collect();
let evict: Vec<usize> = (0..n).filter(|&i| !keep_flag[i]).collect();
RetentionPlan {
keep,
evict,
retrieve: retrieve.unwrap_or_default(),
store_evicted,
}
}
fn plan_retrieval(
&mut self,
_block: usize, resident_blocks: usize,
sinks: usize,
recent: usize,
query_key: Option<&[f32]>,
) -> RetentionPlan {
let retrieve: Vec<usize> = match query_key {
Some(q) if !self.store.is_empty() => {
let mut scored: Vec<(usize, f32)> =
self.store.iter().map(|b| (b.id, dot(q, &b.key))).collect();
scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
scored
.into_iter()
.take(resident_blocks)
.map(|(id, _)| id)
.collect()
}
_ => Vec::new(),
};
self.plan_keep_set(sinks, recent, 0, |_| 0.0, true, Some(retrieve))
}
pub fn push_evicted_block(
&mut self,
start_pos: usize,
k: Vec<Vec<f32>>,
v: Vec<Vec<f32>>,
) -> usize {
let rows = if self.kv_dim > 0 {
k.first().map(|l| l.len() / self.kv_dim).unwrap_or(0)
} else {
0
};
let key = mean_rows(
k.first().map(|l| l.as_slice()).unwrap_or(&[]),
self.kv_dim,
rows,
);
let id = self.next_block_id;
self.next_block_id += 1;
self.store.push(StoredBlock {
id,
start_pos,
rows,
key,
k,
v,
});
id
}
pub fn store_block_size(&self) -> usize {
self.block_size()
}
pub fn take_block(&mut self, id: usize) -> Option<StoredBlock> {
if let Some(idx) = self.store.iter().position(|b| b.id == id) {
Some(self.store.remove(idx))
} else {
None
}
}
pub fn commit(&mut self, plan: &RetentionPlan, retrieved_positions: &[usize]) {
let mut next: Vec<Slot> = Vec::with_capacity(plan.keep.len() + retrieved_positions.len());
for &i in &plan.keep {
if let Some(s) = self.resident.get(i) {
next.push(s.clone());
}
}
let block_sz = self.block_size().max(1);
for &pos in retrieved_positions {
next.push(Slot {
abs_pos: pos,
attn_mass: 0.0,
last_used: self.step,
block: pos / block_sz,
});
}
next.sort_by_key(|s| s.abs_pos);
next.dedup_by_key(|s| s.abs_pos);
self.resident = next;
if let Some(rec) = &mut self.recorder {
let step = rec.len();
rec.push(crate::kv_metrics::StepRecord {
step,
resident: self.resident.len(),
evicted: plan.evict.len(),
retrieved: retrieved_positions.len(),
store_blocks: self.store.len(),
store_tokens: self.store.iter().map(|b| b.rows).sum(),
decode_ms: None,
});
}
}
pub fn resident_positions(&self) -> Vec<usize> {
self.resident.iter().map(|s| s.abs_pos).collect()
}
}
fn dot(a: &[f32], b: &[f32]) -> f32 {
let n = a.len().min(b.len());
let mut s = 0.0;
for i in 0..n {
s += a[i] * b[i];
}
s
}
fn mean_rows(data: &[f32], inner: usize, rows: usize) -> Vec<f32> {
if inner == 0 || rows == 0 {
return vec![0.0; inner];
}
let mut out = vec![0.0f32; inner];
for r in 0..rows {
let base = r * inner;
for j in 0..inner {
out[j] += data[base + j];
}
}
let inv = 1.0 / rows as f32;
for v in out.iter_mut() {
*v *= inv;
}
out
}
#[allow(dead_code)]
type RecentRing = VecDeque<usize>;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn full_keeps_everything() {
let mut m = KvRetentionManager::new(KvRetentionPolicy::Full, 4);
m.on_prefill(10);
let plan = m.plan(None);
assert_eq!(plan.keep.len(), 10);
assert!(plan.evict.is_empty());
assert!(plan.is_noop());
}
#[test]
fn sinks_keeps_first_and_last() {
let mut m = KvRetentionManager::new(
KvRetentionPolicy::Sinks {
sinks: 2,
window: 3,
},
4,
);
m.on_prefill(10);
let plan = m.plan(None);
assert_eq!(plan.keep, vec![0, 1, 7, 8, 9]);
assert_eq!(plan.evict, vec![2, 3, 4, 5, 6]);
assert!(!plan.store_evicted);
}
#[test]
fn heavy_hitter_keeps_high_attention_middle() {
let mut m = KvRetentionManager::new(
KvRetentionPolicy::HeavyHitter {
sinks: 1,
recent: 1,
budget: 2,
},
4,
);
m.on_prefill(8);
let mut w = vec![0.0f32; 8];
w[3] = 5.0;
w[5] = 4.0;
w[2] = 0.1;
w[4] = 0.1;
w[6] = 0.1;
m.observe_attention(&w);
let plan = m.plan(None);
assert!(plan.keep.contains(&0)); assert!(plan.keep.contains(&7)); assert!(plan.keep.contains(&3)); assert!(plan.keep.contains(&5)); assert!(plan.evict.contains(&2)); assert!(plan.evict.contains(&4));
}
#[test]
fn retrieval_stores_and_retrieves_by_query() {
let kv = 2;
let mut m = KvRetentionManager::new(
KvRetentionPolicy::Retrieval {
block: 2,
resident_blocks: 1,
sinks: 1,
recent: 1,
},
kv,
);
m.on_prefill(2);
let a = m.push_evicted_block(
2,
vec![vec![1.0, 0.0, 1.0, 0.0]],
vec![vec![9.0, 9.0, 9.0, 9.0]],
); let b = m.push_evicted_block(
4,
vec![vec![0.0, 1.0, 0.0, 1.0]],
vec![vec![8.0, 8.0, 8.0, 8.0]],
); assert_eq!(m.stored_blocks(), 2);
assert_eq!(m.stored_tokens(), 4);
let q = vec![0.0, 1.0];
let plan2 = m.plan(Some(&q));
assert_eq!(plan2.retrieve, vec![b]);
assert_ne!(plan2.retrieve, vec![a]);
let blk = m.take_block(b).unwrap();
assert_eq!(blk.start_pos, 4);
assert_eq!(m.stored_blocks(), 1);
}
#[test]
fn auto_full_when_fits_then_selective_when_long() {
let mut m = KvRetentionManager::new(KvRetentionPolicy::Auto { max_resident: 64 }, 4);
m.on_prefill(32);
assert!(m.plan(None).is_noop());
m.on_prefill(100);
let mut w = vec![0.01f32; 100];
w[50] = 10.0; for _ in 0..5 {
m.observe_attention(&w);
}
let plan = m.plan(None);
assert!(
!plan.evict.is_empty(),
"long+concentrated context should evict"
);
}
#[test]
fn commit_reorders_by_absolute_position() {
let mut m = KvRetentionManager::new(
KvRetentionPolicy::Sinks {
sinks: 1,
window: 1,
},
2,
);
m.on_prefill(4);
let plan = m.plan(None); m.commit(&plan, &[1]); assert_eq!(m.resident_positions(), vec![0, 1, 3]);
}
}