use rusqlite::{Connection, OptionalExtension, params};
use serde::{Deserialize, Serialize};
use crate::store::StoreError;
pub const MEMORY_SCHEMA: &str = "roteiro.memory/v1";
pub const DEFAULT_MEMORY_SCOPE: &str = "repo";
pub const MAX_MEMORY_BODY: usize = 64 * 1024;
pub const MAX_MEMORY_SCOPE: usize = 128;
#[derive(Debug, thiserror::Error)]
pub enum MemoryError {
#[error(transparent)]
Store(#[from] StoreError),
#[error(
"invalid scope {0:?} (expected 1 to {MAX_MEMORY_SCOPE} bytes, no control characters, no surrounding whitespace)"
)]
InvalidScope(String),
#[error("invalid body: {0}")]
InvalidBody(String),
#[error("invalid confidence {0}: expected a finite number in [0.0, 1.0]")]
InvalidConfidence(f64),
#[error("no memory record with id {0}")]
NotFound(i64),
#[error("memory record {id} is already superseded by {by}")]
AlreadySuperseded {
id: i64,
by: i64,
},
#[error("corrupt memory record: {0}")]
Corrupt(String),
#[error(
"invalid {CACHE_BUDGET_ENV}={0:?}: expected a whole number of megabytes (the default is \
{default} MB)",
default = DEFAULT_CACHE_BUDGET_BYTES / (1024 * 1024)
)]
InvalidBudget(String),
}
impl From<rusqlite::Error> for MemoryError {
fn from(err: rusqlite::Error) -> Self {
Self::Store(StoreError::Sqlite(err))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum MemoryKind {
Lesson,
Attempt,
Decision,
Pattern,
Outcome,
}
impl MemoryKind {
pub const ALL: [Self; 5] = [
Self::Lesson,
Self::Attempt,
Self::Decision,
Self::Pattern,
Self::Outcome,
];
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Self::Lesson => "lesson",
Self::Attempt => "attempt",
Self::Decision => "decision",
Self::Pattern => "pattern",
Self::Outcome => "outcome",
}
}
#[must_use]
pub fn from_token(s: &str) -> Option<Self> {
Self::ALL.into_iter().find(|k| k.as_str() == s)
}
}
impl std::fmt::Display for MemoryKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
impl std::str::FromStr for MemoryKind {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
Self::from_token(s).ok_or_else(|| {
let known = Self::ALL.map(Self::as_str).join(", ");
format!("unknown memory kind {s:?} (expected one of: {known})")
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum AnchorState {
Unanchored,
Valid,
Drifted,
Vanished,
Unverifiable,
}
impl AnchorState {
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Self::Unanchored => "unanchored",
Self::Valid => "valid",
Self::Drifted => "drifted",
Self::Vanished => "vanished",
Self::Unverifiable => "unverifiable",
}
}
#[must_use]
pub fn applies(self) -> bool {
matches!(self, Self::Unanchored | Self::Valid)
}
#[must_use]
pub fn is_stale(self) -> bool {
matches!(self, Self::Drifted | Self::Vanished)
}
}
impl std::fmt::Display for AnchorState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct MemoryAnchor {
pub key: String,
pub blob: Option<String>,
pub path: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct MemoryRecord {
pub id: i64,
pub scope: String,
pub kind: MemoryKind,
#[serde(skip_serializing_if = "Option::is_none")]
pub anchor: Option<MemoryAnchor>,
pub anchor_state: AnchorState,
pub applies: bool,
pub body: String,
pub confidence: Option<f64>,
pub tree: Option<String>,
pub created_at: String,
pub superseded_by: Option<i64>,
pub superseded_at: Option<String>,
}
impl MemoryRecord {
#[must_use]
pub fn is_live(&self) -> bool {
self.superseded_by.is_none()
}
}
#[derive(Debug, Clone, Copy)]
pub struct MemoryWrite<'a> {
pub scope: &'a str,
pub kind: MemoryKind,
pub anchor: Option<&'a str>,
pub body: &'a str,
pub confidence: Option<f64>,
pub supersedes: Option<i64>,
}
impl MemoryWrite<'_> {
pub fn validate(&self) -> Result<(), MemoryError> {
if self.scope.is_empty()
|| self.scope.len() > MAX_MEMORY_SCOPE
|| self.scope.trim() != self.scope
|| self.scope.chars().any(char::is_control)
{
return Err(MemoryError::InvalidScope(self.scope.to_owned()));
}
if self.body.trim().is_empty() {
return Err(MemoryError::InvalidBody(
"it is empty or only whitespace".to_owned(),
));
}
if self.body.len() > MAX_MEMORY_BODY {
return Err(MemoryError::InvalidBody(format!(
"it is {} bytes, over the {MAX_MEMORY_BODY}-byte limit",
self.body.len()
)));
}
if let Some(confidence) = self.confidence
&& !(confidence.is_finite() && (0.0..=1.0).contains(&confidence))
{
return Err(MemoryError::InvalidConfidence(confidence));
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct MemoryFilter<'a> {
pub scope: Option<&'a str>,
pub kind: Option<MemoryKind>,
pub anchor_key: Option<&'a str>,
pub include_superseded: bool,
pub limit: Option<usize>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct MemoryListing {
pub schema: &'static str,
pub records: Vec<MemoryRecord>,
pub live: u64,
pub superseded: u64,
}
pub const RECALL_SCHEMA: &str = "roteiro.recall/v1";
pub const DEFAULT_BASE_CONFIDENCE: f64 = 0.5;
pub const DEFAULT_DECAY_SPAN: u64 = 200;
pub const DEFAULT_HALF_LIFE: u64 = 50;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Decay {
None,
Linear {
span: u64,
},
Exponential {
half_life: u64,
},
}
impl Default for Decay {
fn default() -> Self {
Self::None
}
}
impl Decay {
#[must_use]
pub fn factor(self, age: u64) -> f64 {
match self {
Self::None => 1.0,
Self::Linear { span } => {
let span = span.max(1);
#[expect(
clippy::cast_precision_loss,
reason = "generation counts are small; the ratio is a ranking weight"
)]
let ratio = age as f64 / span as f64;
(1.0 - ratio).max(0.0)
}
Self::Exponential { half_life } => {
let half_life = half_life.max(1);
#[expect(
clippy::cast_precision_loss,
reason = "generation counts are small; the ratio is a ranking weight"
)]
let ratio = age as f64 / half_life as f64;
0.5_f64.powf(ratio)
}
}
}
#[must_use]
pub fn as_str(self) -> &'static str {
match self {
Self::None => "none",
Self::Linear { .. } => "linear",
Self::Exponential { .. } => "exponential",
}
}
#[must_use]
pub fn is_reproducible(self) -> bool {
matches!(self, Self::None)
}
}
impl std::fmt::Display for Decay {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::None => f.write_str("none"),
Self::Linear { span } => write!(f, "linear:{span}"),
Self::Exponential { half_life } => write!(f, "exponential:{half_life}"),
}
}
}
impl std::str::FromStr for Decay {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let (mode, param) = match s.split_once(':') {
Some((mode, param)) => {
let n = param.parse::<u64>().map_err(|_| {
format!("decay parameter {param:?} is not a whole number of generations")
})?;
(mode, Some(n))
}
None => (s, None),
};
match mode {
"none" => {
if param.is_some() {
return Err("decay `none` takes no parameter: it has no age term".to_owned());
}
Ok(Self::None)
}
"linear" => Ok(Self::Linear {
span: param.unwrap_or(DEFAULT_DECAY_SPAN),
}),
"exponential" => Ok(Self::Exponential {
half_life: param.unwrap_or(DEFAULT_HALF_LIFE),
}),
other => Err(format!(
"unknown decay mode {other:?} (expected one of: none, linear[:span], \
exponential[:half-life])"
)),
}
}
}
#[must_use]
pub fn anchor_penalty(state: AnchorState) -> f64 {
match state {
AnchorState::Valid => 1.0,
AnchorState::Unanchored => 0.90,
AnchorState::Unverifiable => 0.50,
AnchorState::Vanished => 0.35,
AnchorState::Drifted => 0.25,
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RecallOptions<'a> {
pub scope: Option<&'a str>,
pub kind: Option<MemoryKind>,
pub anchor_key: Option<&'a str>,
pub query: Option<&'a str>,
pub decay: Decay,
pub applicable_only: bool,
pub limit: Option<usize>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Recalled {
pub score: f64,
pub base_confidence: f64,
pub anchor_penalty: f64,
pub decay_factor: f64,
pub age: u64,
pub record: MemoryRecord,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Recall {
pub schema: &'static str,
pub generation: i64,
pub decay: Decay,
pub reproducible: bool,
pub results: Vec<Recalled>,
pub live: u64,
pub superseded: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct MemoryForgotten {
pub id: i64,
pub restored: Vec<i64>,
}
pub const CACHE_SCHEMA: &str = "roteiro.cache/v1";
pub const DEFAULT_CACHE_BUDGET_BYTES: u64 = 256 * 1024 * 1024;
pub const CACHE_BUDGET_ENV: &str = "ROTEIRO_CACHE_BUDGET_MB";
pub fn cache_budget_bytes() -> Result<u64, MemoryError> {
let Some(raw) = std::env::var_os(CACHE_BUDGET_ENV) else {
return Ok(DEFAULT_CACHE_BUDGET_BYTES);
};
let raw = raw.to_string_lossy().into_owned();
let megabytes: u64 = raw
.trim()
.parse()
.map_err(|_| MemoryError::InvalidBudget(raw.clone()))?;
megabytes
.checked_mul(1024 * 1024)
.ok_or(MemoryError::InvalidBudget(raw))
}
#[derive(Debug, Clone, Copy)]
pub struct CacheWrite<'a> {
pub key: &'a str,
pub fingerprint: &'a str,
pub json: &'a str,
pub anchor: Option<&'a str>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CacheEntry {
pub key: String,
pub fingerprint: String,
pub json: String,
pub bytes: u64,
pub generation: i64,
pub last_used: i64,
pub hits: u64,
#[serde(skip_serializing_if = "Option::is_none")]
pub anchor: Option<String>,
pub anchor_state: AnchorState,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CacheStats {
pub schema: &'static str,
pub entries: u64,
pub bytes: u64,
pub budget_bytes: u64,
pub generation: i64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CacheSweep {
pub schema: &'static str,
pub budget_bytes: u64,
pub scanned: u64,
pub pinned: u64,
pub evicted: u64,
pub freed_bytes: u64,
pub retained_bytes: u64,
pub over_budget: bool,
pub generation: i64,
}
const RECORD_COLS: &str = "m.id, m.scope, m.kind, m.anchor_key, m.anchor_blob, m.anchor_path, \
m.body, m.confidence, m.tree, m.created_at, m.superseded_by, m.superseded_at, \
n.key, n.blob_hash";
const RECORD_FROM: &str = " FROM agent_memory m LEFT JOIN nodes n ON n.key = m.anchor_key";
pub(crate) fn record(conn: &Connection, write: &MemoryWrite<'_>) -> Result<i64, MemoryError> {
write.validate()?;
if let Some(target) = write.supersedes {
let existing: Option<Option<i64>> = conn
.query_row(
"SELECT superseded_by FROM agent_memory WHERE id = ?1",
[target],
|r| r.get(0),
)
.optional()?;
match existing {
None => return Err(MemoryError::NotFound(target)),
Some(Some(by)) => return Err(MemoryError::AlreadySuperseded { id: target, by }),
Some(None) => {}
}
}
let anchor: Option<(Option<String>, Option<String>)> = match write.anchor {
Some(key) => Some(
conn.query_row(
"SELECT blob_hash, path FROM nodes WHERE key = ?1",
[key],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.optional()?
.unwrap_or((None, None)),
),
None => None,
};
let tree: Option<String> = conn
.query_row("SELECT tree FROM sync_state WHERE id = 0", [], |r| r.get(0))
.optional()?;
conn.execute(
"INSERT INTO agent_memory (
scope, kind, anchor_key, anchor_blob, anchor_path, body, confidence, tree
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
params![
write.scope,
write.kind.as_str(),
write.anchor,
anchor.as_ref().and_then(|(blob, _)| blob.as_deref()),
anchor.as_ref().and_then(|(_, path)| path.as_deref()),
write.body,
write.confidence,
tree,
],
)?;
let id = conn.last_insert_rowid();
if let Some(target) = write.supersedes {
conn.execute(
"UPDATE agent_memory
SET superseded_by = ?1, superseded_at = datetime('now')
WHERE id = ?2",
params![id, target],
)?;
}
Ok(id)
}
pub(crate) fn records(
conn: &Connection,
filter: &MemoryFilter<'_>,
) -> Result<Vec<MemoryRecord>, StoreError> {
let mut where_parts: Vec<&str> = Vec::new();
let mut bound: Vec<String> = Vec::new();
if let Some(scope) = filter.scope {
where_parts.push("m.scope = ?");
bound.push(scope.to_owned());
}
if let Some(kind) = filter.kind {
where_parts.push("m.kind = ?");
bound.push(kind.as_str().to_owned());
}
if let Some(key) = filter.anchor_key {
where_parts.push("m.anchor_key = ?");
bound.push(key.to_owned());
}
if !filter.include_superseded {
where_parts.push("m.superseded_by IS NULL");
}
let clause = if where_parts.is_empty() {
String::new()
} else {
format!(" WHERE {}", where_parts.join(" AND "))
};
let limit = match filter.limit {
Some(n) => format!(" LIMIT {n}"),
None => String::new(),
};
let sql = format!("SELECT {RECORD_COLS}{RECORD_FROM}{clause} ORDER BY m.id DESC{limit}");
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query(rusqlite::params_from_iter(bound))?;
let mut out = Vec::new();
while let Some(row) = rows.next()? {
out.push(record_from_row(row)?);
}
Ok(out)
}
pub(crate) fn generation(conn: &Connection) -> Result<i64, StoreError> {
Ok(
conn.query_row("SELECT COALESCE(MAX(id), 0) FROM agent_memory", [], |r| {
r.get(0)
})?,
)
}
pub(crate) fn recall(
conn: &Connection,
opts: &RecallOptions<'_>,
) -> Result<Vec<Recalled>, StoreError> {
let generation = generation(conn)?;
let rows = records(
conn,
&MemoryFilter {
scope: opts.scope,
kind: opts.kind,
anchor_key: opts.anchor_key,
include_superseded: false,
limit: None,
},
)?;
let query = opts.query.map(|q| q.trim().to_lowercase());
let tokens: Vec<&str> = query
.as_deref()
.map(|q| q.split("::").flat_map(str::split_whitespace).collect())
.unwrap_or_default();
let mut out: Vec<Recalled> = Vec::new();
for record in rows {
if opts.applicable_only && !record.applies {
continue;
}
if !tokens.is_empty() && !matches_tokens(&record, &tokens) {
continue;
}
let base_confidence = record.confidence.unwrap_or(DEFAULT_BASE_CONFIDENCE);
let anchor_penalty = anchor_penalty(record.anchor_state);
let age = u64::try_from(generation.saturating_sub(record.id)).unwrap_or(0);
let decay_factor = opts.decay.factor(age);
out.push(Recalled {
score: base_confidence * anchor_penalty * decay_factor,
base_confidence,
anchor_penalty,
decay_factor,
age,
record,
});
}
out.sort_by(|a, b| {
b.score
.total_cmp(&a.score)
.then_with(|| b.record.id.cmp(&a.record.id))
});
if let Some(limit) = opts.limit {
out.truncate(limit);
}
Ok(out)
}
fn matches_tokens(record: &MemoryRecord, tokens: &[&str]) -> bool {
let body = record.body.to_lowercase();
let anchor_key = record
.anchor
.as_ref()
.map(|a| a.key.to_lowercase())
.unwrap_or_default();
let anchor_path = record
.anchor
.as_ref()
.and_then(|a| a.path.as_deref())
.unwrap_or_default()
.to_lowercase();
tokens
.iter()
.all(|t| body.contains(t) || anchor_key.contains(t) || anchor_path.contains(t))
}
pub(crate) fn get(conn: &Connection, id: i64) -> Result<Option<MemoryRecord>, StoreError> {
let sql = format!("SELECT {RECORD_COLS}{RECORD_FROM} WHERE m.id = ?1");
conn.query_row(&sql, [id], |row| Ok(record_from_row(row)))
.optional()?
.transpose()
}
pub(crate) fn forget(conn: &Connection, id: i64) -> Result<Option<MemoryForgotten>, StoreError> {
let present: Option<i64> = conn
.query_row("SELECT id FROM agent_memory WHERE id = ?1", [id], |r| {
r.get(0)
})
.optional()?;
if present.is_none() {
return Ok(None);
}
let restored: Vec<i64> = {
let mut stmt =
conn.prepare("SELECT id FROM agent_memory WHERE superseded_by = ?1 ORDER BY id")?;
let mut rows = stmt.query([id])?;
let mut out = Vec::new();
while let Some(row) = rows.next()? {
out.push(row.get::<_, i64>(0)?);
}
out
};
conn.execute(
"UPDATE agent_memory SET superseded_by = NULL, superseded_at = NULL
WHERE superseded_by = ?1",
[id],
)?;
conn.execute("DELETE FROM agent_memory WHERE id = ?1", [id])?;
Ok(Some(MemoryForgotten { id, restored }))
}
pub(crate) fn counts(conn: &Connection) -> Result<(u64, u64), StoreError> {
let (live, superseded): (i64, i64) = conn.query_row(
"SELECT COALESCE(SUM(superseded_by IS NULL), 0), COALESCE(SUM(superseded_by IS NOT NULL), 0)
FROM agent_memory",
[],
|r| Ok((r.get(0)?, r.get(1)?)),
)?;
Ok((
u64::try_from(live).unwrap_or(0),
u64::try_from(superseded).unwrap_or(0),
))
}
fn next_tick(conn: &Connection) -> Result<i64, StoreError> {
Ok(conn.query_row(
"UPDATE agent_cache_clock SET ticks = ticks + 1 WHERE id = 0 RETURNING ticks",
[],
|r| r.get(0),
)?)
}
fn cache_generation(conn: &Connection) -> Result<i64, StoreError> {
Ok(conn.query_row(
"SELECT generation FROM agent_cache_clock WHERE id = 0",
[],
|r| r.get(0),
)?)
}
pub(crate) fn cache_put(conn: &Connection, write: &CacheWrite<'_>) -> Result<(), StoreError> {
let bytes = u64::try_from(write.key.len() + write.fingerprint.len() + write.json.len())
.unwrap_or(u64::MAX);
let anchor_blob: Option<String> = match write.anchor {
Some(key) => conn
.query_row("SELECT blob_hash FROM nodes WHERE key = ?1", [key], |r| {
r.get(0)
})
.optional()?
.flatten(),
None => None,
};
let tick = next_tick(conn)?;
let generation = cache_generation(conn)?;
conn.execute(
"INSERT INTO agent_cache
(key, fingerprint, json, bytes, generation, last_used, hits, anchor_key, anchor_blob)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, ?7, ?8)
ON CONFLICT(key) DO UPDATE SET
fingerprint = excluded.fingerprint,
json = excluded.json,
bytes = excluded.bytes,
generation = excluded.generation,
last_used = excluded.last_used,
anchor_key = excluded.anchor_key,
anchor_blob = excluded.anchor_blob",
params![
write.key,
write.fingerprint,
write.json,
i64::try_from(bytes).unwrap_or(i64::MAX),
generation,
tick,
write.anchor,
anchor_blob,
],
)?;
Ok(())
}
const CACHE_COLS: &str = "c.key, c.fingerprint, c.json, c.bytes, c.generation, c.last_used, \
c.hits, c.anchor_key, c.anchor_blob, n.key, n.blob_hash";
const SWEEP_COLS: &str = "c.key, c.bytes, c.generation, c.last_used, c.anchor_key, \
c.anchor_blob, n.key, n.blob_hash";
const CACHE_FROM: &str = " FROM agent_cache c LEFT JOIN nodes n ON n.key = c.anchor_key";
pub(crate) fn cache_get(conn: &Connection, key: &str) -> Result<Option<CacheEntry>, StoreError> {
let sql = format!("SELECT {CACHE_COLS}{CACHE_FROM} WHERE c.key = ?1");
let entry = conn
.query_row(&sql, [key], |row| Ok(cache_entry_from_row(row)))
.optional()?
.transpose()?;
if entry.is_some() {
let tick = next_tick(conn)?;
conn.execute(
"UPDATE agent_cache SET hits = hits + 1, last_used = ?1 WHERE key = ?2",
params![tick, key],
)?;
}
Ok(entry)
}
pub(crate) fn cache_entries(conn: &Connection) -> Result<Vec<CacheEntry>, StoreError> {
let sql = format!("SELECT {CACHE_COLS}{CACHE_FROM} ORDER BY c.key");
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query([])?;
let mut out = Vec::new();
while let Some(row) = rows.next()? {
out.push(cache_entry_from_row(row)?);
}
Ok(out)
}
#[derive(Debug, Clone)]
struct SweepRow {
key: String,
bytes: u64,
generation: i64,
last_used: i64,
anchor_state: AnchorState,
}
fn sweep_rows(conn: &Connection) -> Result<Vec<SweepRow>, StoreError> {
let sql = format!("SELECT {SWEEP_COLS}{CACHE_FROM} ORDER BY c.key");
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query([])?;
let mut out = Vec::new();
while let Some(row) = rows.next()? {
let bytes: i64 = row.get(1)?;
let anchor_key: Option<String> = row.get(4)?;
let anchor_blob: Option<String> = row.get(5)?;
let node_key: Option<String> = row.get(6)?;
let node_blob: Option<String> = row.get(7)?;
out.push(SweepRow {
key: row.get(0)?,
bytes: u64::try_from(bytes).unwrap_or(0),
generation: row.get(2)?,
last_used: row.get(3)?,
anchor_state: resolve_anchor(
anchor_key.as_deref(),
anchor_blob.as_deref(),
node_key.as_deref(),
node_blob.as_deref(),
),
});
}
Ok(out)
}
pub(crate) fn cache_forget(conn: &Connection, key: &str) -> Result<bool, StoreError> {
Ok(conn.execute("DELETE FROM agent_cache WHERE key = ?1", [key])? > 0)
}
pub(crate) fn cache_stats(conn: &Connection, budget_bytes: u64) -> Result<CacheStats, StoreError> {
let (entries, bytes): (i64, i64) = conn.query_row(
"SELECT COUNT(*), COALESCE(SUM(bytes), 0) FROM agent_cache",
[],
|r| Ok((r.get(0)?, r.get(1)?)),
)?;
Ok(CacheStats {
schema: CACHE_SCHEMA,
entries: u64::try_from(entries).unwrap_or(0),
bytes: u64::try_from(bytes).unwrap_or(0),
budget_bytes,
generation: cache_generation(conn)?,
})
}
fn evict_count(evictable_lru_first: &[u64], pinned_bytes: u64, budget_bytes: u64) -> usize {
let mut total: u64 = pinned_bytes.saturating_add(evictable_lru_first.iter().sum());
let mut evict = 0;
while evict < evictable_lru_first.len() && total > budget_bytes {
total = total.saturating_sub(evictable_lru_first[evict]);
evict += 1;
}
evict
}
pub(crate) fn cache_sweep(conn: &Connection, budget_bytes: u64) -> Result<CacheSweep, StoreError> {
let generation = cache_generation(conn)?;
let entries = sweep_rows(conn)?;
let scanned = u64::try_from(entries.len()).unwrap_or(u64::MAX);
let mru = entries.iter().max_by_key(|e| e.last_used).map(|e| &e.key);
let mut pinned_bytes: u64 = 0;
let mut candidates: Vec<&SweepRow> = Vec::new();
for entry in &entries {
let is_mru = mru.is_some_and(|k| *k == entry.key);
let own_work = entry.generation >= generation && entry.anchor_state.applies();
if is_mru || own_work {
pinned_bytes = pinned_bytes.saturating_add(entry.bytes);
} else {
candidates.push(entry);
}
}
candidates.sort_by(|a, b| {
a.anchor_state
.applies()
.cmp(&b.anchor_state.applies())
.then_with(|| a.last_used.cmp(&b.last_used))
.then_with(|| a.key.cmp(&b.key))
});
let sizes: Vec<u64> = candidates.iter().map(|e| e.bytes).collect();
let evict = evict_count(&sizes, pinned_bytes, budget_bytes);
let mut freed_bytes: u64 = 0;
for entry in candidates.iter().take(evict) {
conn.execute("DELETE FROM agent_cache WHERE key = ?1", [&entry.key])?;
freed_bytes = freed_bytes.saturating_add(entry.bytes);
}
let held: u64 = entries.iter().map(|e| e.bytes).sum();
let retained_bytes = held.saturating_sub(freed_bytes);
let generation: i64 = conn.query_row(
"UPDATE agent_cache_clock SET generation = generation + 1 WHERE id = 0
RETURNING generation",
[],
|r| r.get(0),
)?;
Ok(CacheSweep {
schema: CACHE_SCHEMA,
budget_bytes,
scanned,
pinned: u64::try_from(entries.len().saturating_sub(candidates.len())).unwrap_or(0),
evicted: u64::try_from(evict).unwrap_or(0),
freed_bytes,
retained_bytes,
over_budget: retained_bytes > budget_bytes,
generation,
})
}
fn cache_entry_from_row(row: &rusqlite::Row<'_>) -> Result<CacheEntry, StoreError> {
let anchor_key: Option<String> = row.get(7)?;
let anchor_blob: Option<String> = row.get(8)?;
let node_key: Option<String> = row.get(9)?;
let node_blob: Option<String> = row.get(10)?;
let bytes: i64 = row.get(3)?;
let hits: i64 = row.get(6)?;
Ok(CacheEntry {
key: row.get(0)?,
fingerprint: row.get(1)?,
json: row.get(2)?,
bytes: u64::try_from(bytes).unwrap_or(0),
generation: row.get(4)?,
last_used: row.get(5)?,
hits: u64::try_from(hits).unwrap_or(0),
anchor_state: resolve_anchor(
anchor_key.as_deref(),
anchor_blob.as_deref(),
node_key.as_deref(),
node_blob.as_deref(),
),
anchor: anchor_key,
})
}
fn resolve_anchor(
anchor_key: Option<&str>,
anchor_blob: Option<&str>,
node_key: Option<&str>,
node_blob: Option<&str>,
) -> AnchorState {
match (anchor_key, node_key) {
(None, _) => AnchorState::Unanchored,
(Some(_), None) => AnchorState::Vanished,
(Some(_), Some(_)) => match (anchor_blob, node_blob) {
(Some(captured), Some(current)) if captured == current => AnchorState::Valid,
(Some(_), Some(_)) => AnchorState::Drifted,
_ => AnchorState::Unverifiable,
},
}
}
fn record_from_row(row: &rusqlite::Row<'_>) -> Result<MemoryRecord, StoreError> {
let kind_token: String = row.get(2)?;
let kind = MemoryKind::from_token(&kind_token)
.ok_or_else(|| StoreError::Corrupt(format!("unknown memory kind: {kind_token}")))?;
let anchor_key: Option<String> = row.get(3)?;
let anchor_blob: Option<String> = row.get(4)?;
let node_key: Option<String> = row.get(12)?;
let node_blob: Option<String> = row.get(13)?;
let anchor_state = resolve_anchor(
anchor_key.as_deref(),
anchor_blob.as_deref(),
node_key.as_deref(),
node_blob.as_deref(),
);
let anchor = anchor_key.map(|key| MemoryAnchor {
key,
blob: anchor_blob,
path: row.get(5).unwrap_or(None),
});
Ok(MemoryRecord {
id: row.get(0)?,
scope: row.get(1)?,
kind,
anchor,
anchor_state,
applies: anchor_state.applies(),
body: row.get(6)?,
confidence: row.get(7)?,
tree: row.get(8)?,
created_at: row.get(9)?,
superseded_by: row.get(10)?,
superseded_at: row.get(11)?,
})
}
#[cfg(test)]
mod tests {
use super::{
AnchorState, CACHE_COLS, DEFAULT_DECAY_SPAN, DEFAULT_HALF_LIFE, DEFAULT_MEMORY_SCOPE,
Decay, MAX_MEMORY_BODY, MAX_MEMORY_SCOPE, MemoryKind, MemoryWrite, SWEEP_COLS,
anchor_penalty, cache_entries, cache_sweep, evict_count, sweep_rows,
};
fn write(body: &str) -> MemoryWrite<'_> {
MemoryWrite {
scope: DEFAULT_MEMORY_SCOPE,
kind: MemoryKind::Lesson,
anchor: None,
body,
confidence: None,
supersedes: None,
}
}
#[test]
fn kind_tokens_round_trip_and_reject_the_unknown() {
for kind in MemoryKind::ALL {
assert_eq!(MemoryKind::from_token(kind.as_str()), Some(kind));
assert_eq!(kind.to_string(), kind.as_str());
}
assert_eq!(MemoryKind::from_token("note"), None);
assert_eq!(MemoryKind::from_token("Lesson"), None);
let err = "note".parse::<MemoryKind>().expect_err("unknown kind");
assert!(
err.contains("lesson"),
"the error lists the vocabulary: {err}"
);
}
#[test]
fn anchor_state_tokens_and_staleness() {
assert!(AnchorState::Drifted.is_stale());
assert!(AnchorState::Vanished.is_stale());
for state in [
AnchorState::Unanchored,
AnchorState::Valid,
AnchorState::Unverifiable,
] {
assert!(!state.is_stale(), "{state} must not read as stale");
}
assert_eq!(AnchorState::Vanished.to_string(), "vanished");
}
#[test]
fn validation_names_what_was_actually_wrong() {
write("a real lesson").validate().expect("the good case");
let over_long = "x".repeat(MAX_MEMORY_BODY + 1);
for (case, w) in [
("empty body", write("")),
("whitespace body", write(" \n\t ")),
("over-long body", write(&over_long)),
] {
assert!(w.validate().is_err(), "{case} must be refused");
}
let long_scope = "s".repeat(MAX_MEMORY_SCOPE + 1);
for scope in ["", " repo", "repo ", "re\npo", &long_scope] {
let w = MemoryWrite {
scope,
..write("body")
};
assert!(w.validate().is_err(), "scope {scope:?} must be refused");
}
for confidence in [Some(0.0), Some(1.0), Some(0.5), None] {
let w = MemoryWrite {
confidence,
..write("body")
};
w.validate().expect("a probability is fine");
}
for confidence in [-0.1, 1.1, f64::NAN, f64::INFINITY] {
let w = MemoryWrite {
confidence: Some(confidence),
..write("body")
};
assert!(
w.validate().is_err(),
"{confidence} is not a probability and must be refused"
);
}
}
#[test]
fn decay_none_is_exactly_one_at_every_age() {
for age in [0, 1, 7, 1_000, u64::from(u32::MAX)] {
assert!(
(Decay::None.factor(age) - 1.0).abs() < f64::EPSILON,
"none must not price age at all, but age {age} moved it",
);
}
assert!(Decay::None.is_reproducible());
assert!(!Decay::Linear { span: 10 }.is_reproducible());
assert!(!Decay::Exponential { half_life: 10 }.is_reproducible());
}
#[test]
fn decay_modes_start_at_one_and_never_rise() {
for decay in [
Decay::Linear { span: 8 },
Decay::Linear {
span: DEFAULT_DECAY_SPAN,
},
Decay::Exponential { half_life: 4 },
Decay::Exponential {
half_life: DEFAULT_HALF_LIFE,
},
] {
assert!(
(decay.factor(0) - 1.0).abs() < f64::EPSILON,
"{decay} must not discount the newest record",
);
let mut previous = f64::INFINITY;
for age in 0..64_u64 {
let f = decay.factor(age);
assert!((0.0..=1.0).contains(&f), "{decay} at age {age} gave {f}");
assert!(f <= previous, "{decay} rose at age {age}");
previous = f;
}
}
assert!((Decay::Linear { span: 10 }.factor(5) - 0.5).abs() < 1e-12);
assert!((Decay::Exponential { half_life: 10 }.factor(10) - 0.5).abs() < 1e-12);
assert!(
(Decay::Exponential { half_life: 10 }.factor(20) - 0.25).abs() < 1e-12,
"two half-lives is a quarter",
);
}
#[test]
fn linear_bottoms_out_and_exponential_does_not() {
let linear = Decay::Linear { span: 10 };
assert!(linear.factor(10).abs() < f64::EPSILON);
assert!(
linear.factor(10_000).abs() < f64::EPSILON,
"and stays there"
);
let exponential = Decay::Exponential { half_life: 10 };
assert!(
exponential.factor(10_000) > 0.0,
"an exponential is never quite zero",
);
assert!((Decay::Linear { span: 0 }.factor(0) - 1.0).abs() < f64::EPSILON);
assert!(Decay::Linear { span: 0 }.factor(1).abs() < f64::EPSILON);
assert!((Decay::Exponential { half_life: 0 }.factor(0) - 1.0).abs() < f64::EPSILON);
}
#[test]
fn decay_tokens_round_trip_and_reject_the_unknown() {
for decay in [
Decay::None,
Decay::Linear { span: 7 },
Decay::Exponential { half_life: 9 },
] {
assert_eq!(
decay.to_string().parse::<Decay>(),
Ok(decay),
"{decay} must round-trip through its token",
);
}
assert_eq!(
"linear".parse::<Decay>(),
Ok(Decay::Linear {
span: DEFAULT_DECAY_SPAN
}),
"a bare mode takes its documented default span",
);
assert_eq!(
"exponential".parse::<Decay>(),
Ok(Decay::Exponential {
half_life: DEFAULT_HALF_LIFE
})
);
assert_eq!(Decay::default(), Decay::None, "reproducible by default");
for bad in ["clock", "none:5", "linear:soon", ""] {
assert!(bad.parse::<Decay>().is_err(), "{bad:?} must be refused");
}
}
#[test]
fn anchor_penalty_demotes_without_ever_silencing() {
let states = [
AnchorState::Unanchored,
AnchorState::Valid,
AnchorState::Drifted,
AnchorState::Vanished,
AnchorState::Unverifiable,
];
for state in states {
let p = anchor_penalty(state);
assert!(p > 0.0, "{state} was silenced, not demoted");
assert!(p <= 1.0, "{state} scored above the maximum");
}
let worst_applying = states
.into_iter()
.filter(|s| s.applies())
.map(anchor_penalty)
.fold(f64::INFINITY, f64::min);
let best_not_applying = states
.into_iter()
.filter(|s| !s.applies())
.map(anchor_penalty)
.fold(0.0, f64::max);
assert!(
worst_applying > best_not_applying,
"a record that applies here must outrank every record that does not \
({worst_applying} vs {best_not_applying})",
);
assert!(
anchor_penalty(AnchorState::Vanished) > anchor_penalty(AnchorState::Drifted),
"a record about deleted code must not be the most demoted of all",
);
}
#[test]
fn eviction_matches_the_model_cache_policy_it_ports() {
assert_eq!(evict_count(&[100, 100], 100, 250), 1);
assert_eq!(evict_count(&[100, 100], 100, 0), 2);
assert_eq!(evict_count(&[100, 100], 100, 1000), 0);
assert_eq!(evict_count(&[100, 100], 100, 300), 0);
}
#[test]
fn nothing_evictable_means_nothing_evicted_however_small_the_budget() {
assert_eq!(evict_count(&[], 500, 10), 0, "the sole entry survives");
assert_eq!(evict_count(&[], 0, 0), 0, "an empty tier sweeps to nothing");
assert_eq!(
evict_count(&[100], 500, 10),
1,
"and everything else still goes",
);
}
#[test]
fn eviction_stops_as_soon_as_the_remainder_fits() {
assert_eq!(evict_count(&[10, 20, 30], 40, 70), 2);
assert_eq!(evict_count(&[10, 20, 30], 40, 90), 1);
assert_eq!(evict_count(&[10, 20, 30], 40, 100), 0, "it already fits");
assert_eq!(evict_count(&[10, 20, 30], 40, 40), 3);
}
#[test]
fn pinned_bytes_are_counted_but_never_freed() {
assert_eq!(
evict_count(&[10, 10], 1000, 100),
2,
"everything evictable goes when the pinned set alone exceeds the budget",
);
}
fn cache_store() -> rusqlite::Connection {
let mut conn = rusqlite::Connection::open_in_memory().expect("open");
crate::migrations::apply(&mut conn).expect("apply");
conn.execute("UPDATE agent_cache_clock SET generation = 5", [])
.expect("advance the clock");
conn
}
fn raw_put(conn: &rusqlite::Connection, key: &str, bytes: i64, payload: usize, last_used: i64) {
conn.execute(
"INSERT INTO agent_cache (key, fingerprint, json, bytes, generation, last_used, hits)
VALUES (?1, 'fp', ?2, ?3, 0, ?4, 0)",
rusqlite::params![key, "x".repeat(payload), bytes, last_used],
)
.expect("insert");
}
#[test]
fn the_sweep_query_names_no_payload_column() {
for forbidden in ["json", "fingerprint", "hits"] {
assert!(
!SWEEP_COLS.contains(forbidden),
"the sweep must not read {forbidden}: it decides by the stored size, \
and reading payloads to decide what to evict is what `bytes` exists \
to avoid — on a full tier that is the whole budget in memory",
);
}
for required in [
"c.key",
"c.bytes",
"c.generation",
"c.last_used",
"c.anchor_key",
"c.anchor_blob",
] {
assert!(SWEEP_COLS.contains(required), "the sweep needs {required}");
}
assert!(
CACHE_COLS.contains("c.json"),
"the inspection path returns it"
);
}
#[test]
fn the_sweep_and_the_full_read_agree_row_for_row() {
let conn = cache_store();
conn.execute(
"INSERT INTO nodes (key, kind, name, blob_hash) VALUES ('sym:a', 'fn', 'a', 'blob1')",
[],
)
.expect("node");
raw_put(&conn, "unanchored", 10, 4, 1);
conn.execute(
"INSERT INTO agent_cache
(key, fingerprint, json, bytes, generation, last_used, hits, anchor_key, anchor_blob)
VALUES ('valid', 'fp', '{}', 20, 0, 2, 0, 'sym:a', 'blob1'),
('drifted', 'fp', '{}', 30, 0, 3, 0, 'sym:a', 'blob-old'),
('vanished', 'fp', '{}', 40, 0, 4, 0, 'sym:gone', 'blob1'),
('unverifiable', 'fp', '{}', 50, 0, 5, 0, 'sym:a', NULL)",
[],
)
.expect("anchored entries");
let narrow = sweep_rows(&conn).expect("sweep rows");
let full = cache_entries(&conn).expect("entries");
assert_eq!(narrow.len(), full.len(), "the same rows, or one is blind");
assert_eq!(narrow.len(), 5);
for (n, f) in narrow.iter().zip(full.iter()) {
assert_eq!(n.key, f.key, "same order, same rows");
assert_eq!(n.bytes, f.bytes, "{}: the size must not differ", n.key);
assert_eq!(n.generation, f.generation, "{}", n.key);
assert_eq!(n.last_used, f.last_used, "{}", n.key);
assert_eq!(
n.anchor_state, f.anchor_state,
"{}: both must resolve the anchor identically",
n.key,
);
}
let states: Vec<AnchorState> = narrow.iter().map(|r| r.anchor_state).collect();
for expected in [
AnchorState::Unanchored,
AnchorState::Valid,
AnchorState::Drifted,
AnchorState::Vanished,
AnchorState::Unverifiable,
] {
assert!(states.contains(&expected), "{expected} was not exercised");
}
}
#[test]
fn the_sweep_totals_the_bytes_column_and_never_the_payload() {
let conn = cache_store();
raw_put(&conn, "heavy", 1000, 1, 1);
raw_put(&conn, "light", 10, 1000, 2);
raw_put(&conn, "mru", 0, 0, 3);
let swept = cache_sweep(&conn, 500).expect("sweep");
assert_eq!(
swept.evicted, 1,
"a payload-measuring sweep would have taken `light` as well",
);
assert_eq!(
swept.freed_bytes, 1000,
"the freed total is the stored size, not the 1 byte `heavy` holds",
);
assert_eq!(
swept.retained_bytes, 10,
"and what remains is counted the same way",
);
let survivors: Vec<String> = sweep_rows(&conn)
.expect("rows")
.into_iter()
.map(|r| r.key)
.collect();
assert_eq!(survivors, vec!["light".to_owned(), "mru".to_owned()]);
}
}