use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use serde::Serialize;
#[derive(Debug)]
pub enum RetentionError {
Io(io::Error),
UnreadableChain(PathBuf),
BudgetUnmet {
budget: u64,
remaining: u64,
protected_chain_heads: usize,
notes_kept: bool,
protected_bytes: u64,
nothing_deleted: bool,
},
}
impl std::fmt::Display for RetentionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RetentionError::Io(e) => write!(f, "retention io error: {e}"),
RetentionError::UnreadableChain(path) => write!(
f,
"unreadable named-chain file {} — refusing every destructive session \
operation while chain protection is unknown",
path.display()
),
RetentionError::BudgetUnmet {
budget,
remaining,
protected_chain_heads,
notes_kept,
protected_bytes,
nothing_deleted,
} => {
if *nothing_deleted {
write!(
f,
"disk budget unmet: nothing was deleted, because the budget of \
{budget} bytes is below the {protected_bytes} bytes this sweep \
may not delete ({protected_chain_heads} protected chain head(s)"
)?;
if *notes_kept {
write!(f, " and every saved note")?;
}
write!(f, "); {remaining} bytes are in use")?;
} else {
write!(
f,
"disk budget unmet: {remaining} bytes remain against a budget of \
{budget} after evicting every unprotected artifact \
({protected_chain_heads} protected chain head(s) retained)"
)?;
}
if *notes_kept {
write!(
f,
"; no note was deleted: the budget evicts notes only with \
--include-notes"
)?;
}
Ok(())
}
}
}
}
impl std::error::Error for RetentionError {}
impl From<io::Error> for RetentionError {
fn from(e: io::Error) -> Self {
RetentionError::Io(e)
}
}
#[derive(Debug, Clone)]
pub struct RetentionRoots {
pub config_dir: PathBuf,
pub base_dir: PathBuf,
pub cache_dir: PathBuf,
}
impl RetentionRoots {
pub fn resolve() -> Self {
let cache_dir = std::env::var("HOME")
.map(|home| PathBuf::from(home).join(".cache/synaps"))
.unwrap_or_else(|_| crate::config::base_dir().join("cache"));
Self {
config_dir: crate::config::get_active_config_dir(),
base_dir: crate::config::base_dir(),
cache_dir,
}
}
fn sessions_dir(&self) -> PathBuf {
self.config_dir.join("sessions")
}
fn chains_dir(&self) -> PathBuf {
self.config_dir.join("chains")
}
fn memory_dir(&self) -> PathBuf {
self.base_dir.join("memory")
}
fn spill_dir(&self) -> PathBuf {
crate::core::tool_spill::spill_root(&self.base_dir)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RetentionDomain {
Sessions,
Memory,
Traces,
Logs,
}
#[derive(Debug, Clone, Copy, Default)]
pub struct RetentionPolicy {
pub max_age_days: Option<u32>,
pub max_disk_bytes: Option<u64>,
pub include_notes: bool,
}
const DAY_MS: u64 = 86_400_000;
#[cfg(unix)]
const NOTES_LOCK_SUFFIX: &str = ".jsonl.lock";
#[derive(Debug, Clone, Serialize)]
pub struct DomainReport {
pub domain: RetentionDomain,
pub files: usize,
pub bytes: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct RetentionReport {
pub domains: Vec<DomainReport>,
pub total_bytes: u64,
}
#[derive(Debug, Clone, Copy, Default, Serialize)]
pub struct SweepOutcome {
pub deleted_files: usize,
pub freed_bytes: u64,
pub protected_chain_heads: usize,
pub memory_records_dropped: usize,
}
#[derive(Debug, Clone, Copy, Serialize)]
pub struct ExportSummary {
pub files: usize,
pub bytes: u64,
}
fn list_files(dir: &Path) -> Vec<PathBuf> {
let mut out = Vec::new();
if let Ok(entries) = fs::read_dir(dir) {
for entry in entries.flatten() {
if entry.path().is_file() {
out.push(entry.path());
}
}
}
out.sort();
out
}
fn log_files(config_dir: &Path) -> Vec<PathBuf> {
let mut out: Vec<PathBuf> = list_files(config_dir)
.into_iter()
.filter(|p| {
p.file_name()
.and_then(|n| n.to_str())
.is_some_and(|n| n.starts_with("synaps.log"))
})
.collect();
out.sort();
out
}
fn chain_heads(roots: &RetentionRoots) -> Result<Vec<String>, RetentionError> {
let mut heads = Vec::new();
for path in list_files(&roots.chains_dir()) {
let raw =
fs::read_to_string(&path).map_err(|_| RetentionError::UnreadableChain(path.clone()))?;
let value: serde_json::Value = serde_json::from_str(&raw)
.map_err(|_| RetentionError::UnreadableChain(path.clone()))?;
match value["head"].as_str() {
Some(head) => heads.push(head.to_string()),
None => return Err(RetentionError::UnreadableChain(path.clone())),
}
}
Ok(heads)
}
fn session_age_ms(path: &Path) -> Option<u64> {
if let Ok(raw) = fs::read_to_string(path) {
if let Ok(value) = serde_json::from_str::<serde_json::Value>(&raw) {
if let Some(updated) = value["updated_at"].as_str() {
if let Ok(ts) = chrono::DateTime::parse_from_rfc3339(updated) {
return Some(ts.timestamp_millis().max(0) as u64);
}
}
}
}
mtime_ms(path)
}
fn mtime_ms(path: &Path) -> Option<u64> {
fs::metadata(path)
.and_then(|m| m.modified())
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_millis() as u64)
}
pub fn inspect(roots: &RetentionRoots) -> io::Result<RetentionReport> {
let sessions = list_files(&roots.sessions_dir());
let memory: Vec<PathBuf> = list_files(&roots.memory_dir())
.into_iter()
.filter(|p| p.extension().and_then(|e| e.to_str()) == Some("jsonl"))
.collect();
let traces = cache_files(&roots.cache_dir);
let logs = log_files(&roots.config_dir);
let sum = |files: &[PathBuf]| -> u64 {
files
.iter()
.filter_map(|p| fs::metadata(p).ok().map(|m| m.len()))
.sum()
};
let domains = vec![
DomainReport {
domain: RetentionDomain::Sessions,
files: sessions.len(),
bytes: sum(&sessions),
},
DomainReport {
domain: RetentionDomain::Memory,
files: memory.len(),
bytes: sum(&memory),
},
DomainReport {
domain: RetentionDomain::Traces,
files: traces.len(),
bytes: sum(&traces),
},
DomainReport {
domain: RetentionDomain::Logs,
files: logs.len(),
bytes: sum(&logs),
},
];
let total_bytes = domains.iter().map(|d| d.bytes).sum();
Ok(RetentionReport {
domains,
total_bytes,
})
}
fn protected_bytes(
roots: &RetentionRoots,
protected: &[String],
policy: &RetentionPolicy,
now_ms: u64,
) -> io::Result<u64> {
let mut total = 0u64;
for path in list_files(&roots.sessions_dir()) {
let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or_default();
if protected.iter().any(|id| id == stem) {
total += fs::metadata(&path)?.len();
}
}
if !policy.include_notes {
for path in list_files(&roots.memory_dir()) {
if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
continue;
}
let raw = fs::read(&path)?;
let (dropped, kept) = compacted_lines(&raw, now_ms, None);
total += if dropped == 0 {
raw.len() as u64 } else {
kept.iter().map(|line| line.len() as u64 + 1).sum()
};
}
}
Ok(total)
}
pub fn sweep_at(
roots: &RetentionRoots,
policy: &RetentionPolicy,
now_ms: u64,
) -> Result<SweepOutcome, RetentionError> {
let mut outcome = SweepOutcome::default();
let protected: Vec<String> = chain_heads(roots)?;
let mut may_not_delete = 0;
if let Some(budget) = policy.max_disk_bytes {
may_not_delete = protected_bytes(roots, &protected, policy, now_ms)?;
if budget < may_not_delete {
return Err(RetentionError::BudgetUnmet {
budget,
remaining: inspect(roots)?.total_bytes,
protected_chain_heads: protected.len(),
notes_kept: !policy.include_notes,
protected_bytes: may_not_delete,
nothing_deleted: true,
});
}
}
let age_cutoff = policy
.max_age_days
.map(|days| now_ms.saturating_sub(days as u64 * DAY_MS));
for path in list_files(&roots.sessions_dir()) {
if path.extension().and_then(|e| e.to_str()) != Some("journal")
|| path.with_extension("json").exists()
{
continue;
}
let id = path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or_default()
.to_string();
if protected.contains(&id) {
outcome.protected_chain_heads += 1;
continue;
}
delete_file(&path, &mut outcome)?;
}
if let Some(cutoff) = age_cutoff {
for path in list_files(&roots.sessions_dir()) {
if is_paired_journal(&path) {
continue;
}
let id = path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or_default()
.to_string();
if protected.contains(&id) {
outcome.protected_chain_heads += 1;
continue;
}
let Some(age_ref) = session_age_ms(&path) else {
continue; };
if age_ref < cutoff {
delete_session_artifact(&path, &mut outcome)?;
remove_session_spill(roots, &id, &mut outcome)?;
}
}
for path in cache_files(&roots.cache_dir)
.into_iter()
.chain(log_files(&roots.config_dir))
{
if mtime_ms(&path).is_some_and(|m| m < cutoff) {
delete_file(&path, &mut outcome)?;
}
}
let age = std::time::Duration::from_millis(now_ms.saturating_sub(cutoff));
let now = std::time::UNIX_EPOCH + std::time::Duration::from_millis(now_ms);
let swept = crate::core::tool_spill::sweep_stale_spill(&roots.spill_dir(), age, now);
outcome.deleted_files += swept.files;
outcome.freed_bytes += swept.bytes;
}
let notes_cutoff = age_cutoff.filter(|_| policy.include_notes);
for path in list_files(&roots.memory_dir()) {
if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
continue;
}
let dropped = compact_memory_file(&path, now_ms, notes_cutoff)?;
outcome.memory_records_dropped += dropped;
}
if let Some(budget) = policy.max_disk_bytes {
let total_now = |roots: &RetentionRoots| -> u64 {
inspect(roots).map(|r| r.total_bytes).unwrap_or(u64::MAX)
};
if total_now(roots) > budget {
let mut candidates: Vec<(u64, PathBuf, bool)> = Vec::new();
for path in list_files(&roots.sessions_dir()) {
if is_paired_journal(&path) {
continue;
}
let id = path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or_default()
.to_string();
let age = session_age_ms(&path).unwrap_or(u64::MAX);
candidates.push((age, path, protected.contains(&id)));
}
for path in cache_files(&roots.cache_dir)
.into_iter()
.chain(log_files(&roots.config_dir))
{
let age = mtime_ms(&path).unwrap_or(u64::MAX);
candidates.push((age, path, false));
}
candidates.sort_by_key(|item| item.0);
for (_, path, is_protected) in candidates {
if total_now(roots) <= budget {
break;
}
if is_protected {
outcome.protected_chain_heads += 1;
continue;
}
delete_file(&path, &mut outcome).map_err(RetentionError::Io)?;
}
}
if policy.include_notes && total_now(roots) > budget {
let over = total_now(roots) - budget;
let dropped = evict_oldest_memory_records(roots, over)?;
outcome.memory_records_dropped += dropped;
}
let remaining = total_now(roots);
if remaining > budget {
return Err(RetentionError::BudgetUnmet {
budget,
remaining,
protected_chain_heads: protected.len(),
notes_kept: !policy.include_notes,
protected_bytes: may_not_delete,
nothing_deleted: false,
});
}
}
Ok(outcome)
}
fn evict_oldest_memory_records(
roots: &RetentionRoots,
bytes_to_free: u64,
) -> Result<usize, RetentionError> {
let mut live: Vec<(u64, PathBuf, String, usize)> = Vec::new();
for path in list_files(&roots.memory_dir()) {
if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
continue;
}
let raw = fs::read(&path).map_err(RetentionError::Io)?;
let mut tombstoned: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut records: Vec<(u64, String, usize)> = Vec::new();
for line in notes_lines(&raw).flatten() {
let Ok(value) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
continue;
};
if let Some(id) = value["tombstone"].as_str() {
tombstoned.insert(id.to_string());
continue;
}
if let Some(id) = value["id"].as_str() {
records.push((
value["timestamp_ms"].as_u64().unwrap_or(0),
id.to_string(),
line.len() + 1,
));
}
}
for (ts, id, len) in records {
if !tombstoned.contains(&id) {
live.push((ts, path.clone(), id, len));
}
}
}
live.sort_by_key(|item| item.0);
let mut freed: u64 = 0;
let mut victims: std::collections::HashMap<PathBuf, std::collections::HashSet<String>> =
std::collections::HashMap::new();
let mut dropped = 0usize;
for (_, path, id, len) in live {
if freed >= bytes_to_free {
break;
}
victims.entry(path).or_default().insert(id);
freed += len as u64;
dropped += 1;
}
for (path, ids) in victims {
let _lock = lock_notes(&path)?;
let raw = fs::read(&path).map_err(RetentionError::Io)?;
rewrite_read_done();
let kept: Vec<&str> = notes_lines(&raw)
.flatten()
.filter(|line| {
let Ok(value) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
return false; };
if let Some(id) = value["id"].as_str() {
return !ids.contains(id);
}
true
})
.collect();
let mut body = kept.join("\n");
if !body.is_empty() {
body.push('\n');
}
crate::core::private_fs::write_atomic_private(&path, body.as_bytes())
.map_err(io::Error::from)
.map_err(RetentionError::Io)?;
}
Ok(dropped)
}
fn notes_lines(raw: &[u8]) -> impl Iterator<Item = Option<&str>> {
let empty = raw.is_empty();
let raw = raw.strip_suffix(b"\n").unwrap_or(raw);
raw.split(|byte| *byte == b'\n')
.filter(move |_| !empty)
.map(|line| {
let line = line.strip_suffix(b"\r").unwrap_or(line);
std::str::from_utf8(line).ok()
})
}
fn lock_notes(path: &Path) -> io::Result<crate::core::lock_file::ExclusiveLock> {
crate::memory::store::lock_notes_file(path)
}
#[cfg(test)]
thread_local! {
static AFTER_REWRITE_READ: std::cell::RefCell<Option<Box<dyn FnMut()>>> =
const { std::cell::RefCell::new(None) };
}
#[cfg(test)]
fn rewrite_read_done() {
let hook = AFTER_REWRITE_READ.with(|hook| hook.borrow_mut().take());
if let Some(mut hook) = hook {
hook();
}
}
#[cfg(not(test))]
fn rewrite_read_done() {}
fn walk_regular_files(root: &Path) -> Vec<PathBuf> {
fn recurse(dir: &Path, depth: usize, out: &mut Vec<PathBuf>) {
if depth > 32 {
return;
}
if let Ok(entries) = fs::read_dir(dir) {
for entry in entries.flatten() {
let Ok(file_type) = entry.file_type() else {
continue;
};
if file_type.is_symlink() {
continue; }
let path = entry.path();
if file_type.is_dir() {
recurse(&path, depth + 1, out);
} else if file_type.is_file() {
out.push(path);
}
}
}
}
let mut out = Vec::new();
recurse(root, 0, &mut out);
out.sort();
out
}
fn cache_files(cache_dir: &Path) -> Vec<PathBuf> {
walk_regular_files(cache_dir)
}
pub fn sweep(
roots: &RetentionRoots,
policy: &RetentionPolicy,
) -> Result<SweepOutcome, RetentionError> {
sweep_at(roots, policy, crate::epoch_millis())
}
fn delete_file(path: &Path, outcome: &mut SweepOutcome) -> io::Result<()> {
let len = fs::metadata(path).map(|m| m.len()).unwrap_or(0);
fs::remove_file(path)?;
outcome.deleted_files += 1;
outcome.freed_bytes += len;
Ok(())
}
const PAIRED_SESSION_EXTENSIONS: [&str; 2] = ["journal", "turn"];
fn is_paired_journal(path: &Path) -> bool {
path.extension()
.and_then(|e| e.to_str())
.is_some_and(|e| PAIRED_SESSION_EXTENSIONS.contains(&e))
&& path.with_extension("json").exists()
}
fn remove_session_spill(
roots: &RetentionRoots,
id: &str,
outcome: &mut SweepOutcome,
) -> io::Result<()> {
let freed = crate::core::tool_spill::remove_session_spill(&roots.spill_dir(), id)?;
outcome.deleted_files += freed.files;
outcome.freed_bytes += freed.bytes;
Ok(())
}
fn delete_session_artifact(path: &Path, outcome: &mut SweepOutcome) -> io::Result<()> {
delete_file(path, outcome)?;
if path.extension().and_then(|e| e.to_str()) == Some("json") {
for ext in PAIRED_SESSION_EXTENSIONS {
let paired = path.with_extension(ext);
if paired.exists() {
delete_file(&paired, outcome)?;
}
}
}
Ok(())
}
fn compact_memory_file(
path: &Path,
now_ms: u64,
global_cutoff_ms: Option<u64>,
) -> io::Result<usize> {
let _lock = lock_notes(path)?;
let raw = fs::read(path)?;
rewrite_read_done();
let (dropped, kept) = compacted_lines(&raw, now_ms, global_cutoff_ms);
if dropped > 0 {
let mut body = kept.join("\n");
if !body.is_empty() {
body.push('\n');
}
crate::core::private_fs::write_atomic_private(path, body.as_bytes())
.map_err(io::Error::from)?;
}
Ok(dropped)
}
fn compacted_lines(
raw: &[u8],
now_ms: u64,
global_cutoff_ms: Option<u64>,
) -> (usize, Vec<String>) {
enum Line {
Tombstone(String),
Record(Option<String>, u64, Option<u64>, String),
}
let mut tombstoned: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut lines: Vec<Line> = Vec::new();
let mut dropped = 0usize;
for line in notes_lines(raw) {
let Some(line) = line else {
dropped += 1; continue;
};
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let Ok(value) = serde_json::from_str::<serde_json::Value>(trimmed) else {
dropped += 1; continue;
};
if let Some(id) = value["tombstone"].as_str() {
tombstoned.insert(id.to_string());
lines.push(Line::Tombstone(trimmed.to_string()));
continue;
}
let ts = value["timestamp_ms"].as_u64().unwrap_or(0);
let per_record_span_ms = value["retention"]["max_age_days"]
.as_u64()
.map(|days| days * DAY_MS);
lines.push(Line::Record(
value["id"].as_str().map(String::from),
ts,
per_record_span_ms,
trimmed.to_string(),
));
}
let mut kept: Vec<String> = Vec::new();
for line in lines {
let (id, ts, per_record_span_ms, line) = match line {
Line::Tombstone(line) => {
kept.push(line);
continue;
}
Line::Record(id, ts, span, line) => (id, ts, span, line),
};
if id.as_ref().is_some_and(|id| tombstoned.contains(id)) {
dropped += 1;
continue;
}
if per_record_span_ms.is_some_and(|span| ts.saturating_add(span) < now_ms) {
dropped += 1;
continue;
}
if global_cutoff_ms.is_some_and(|cutoff| ts < cutoff) {
dropped += 1;
continue;
}
kept.push(line);
}
(dropped, kept)
}
#[cfg(unix)]
pub fn export(roots: &RetentionRoots, dest: &Path) -> io::Result<ExportSummary> {
use crate::private_fs::ConfinedDir;
let mut summary = ExportSummary { files: 0, bytes: 0 };
let root = ConfinedDir::create_root(dest)?;
let flat = [
(roots.sessions_dir(), "sessions"),
(roots.memory_dir(), "memory"),
];
for (src, sub) in flat {
let Ok(src_root) = ConfinedDir::open_root(&src) else {
continue; };
let out = root.child_dir(sub)?;
for entry in src_root.entries()? {
if !entry.is_file {
continue; }
if sub == "memory" && entry.name.ends_with(NOTES_LOCK_SUFFIX) {
continue; }
let components = [entry.name.clone()];
match src_root.open_file(&components) {
Ok(source) => {
let bytes = copy_opened_into_confined(source, &out, &entry.name)?;
summary.bytes += bytes;
summary.files += 1;
}
Err(_) => continue,
}
}
}
if let Ok(cache_root) = ConfinedDir::open_root(&roots.cache_dir) {
let traces_root = root.child_dir("traces")?;
export_tree_from_handles(&cache_root, &traces_root, &mut summary, 0)?;
}
if let Ok(logs_root) = ConfinedDir::open_root(&roots.config_dir) {
let logs_out = root.child_dir("logs")?;
for entry in logs_root.entries()? {
if !entry.is_file || !entry.name.starts_with("synaps.log") {
continue;
}
match logs_root.open_file(std::slice::from_ref(&entry.name)) {
Ok(source) => {
let bytes = copy_opened_into_confined(source, &logs_out, &entry.name)?;
summary.bytes += bytes;
summary.files += 1;
}
Err(_) => continue,
}
}
}
Ok(summary)
}
#[cfg(unix)]
fn copy_opened_into_confined(
mut source: fs::File,
dest_dir: &crate::private_fs::ConfinedDir,
name: &str,
) -> io::Result<u64> {
use std::io::{Read, Write};
if !source.metadata()?.is_file() {
return Err(io::Error::other("confined export source is not regular"));
}
let mut out = dest_dir.create_file(name)?;
let mut buf = [0u8; 64 * 1024];
let mut written = 0u64;
loop {
let n = source.read(&mut buf)?;
if n == 0 {
break;
}
out.write_all(&buf[..n])?;
written += n as u64;
}
Ok(written)
}
#[cfg(unix)]
fn export_tree_from_handles(
source: &crate::private_fs::ConfinedDir,
destination: &crate::private_fs::ConfinedDir,
summary: &mut ExportSummary,
depth: usize,
) -> io::Result<()> {
if depth > 32 {
return Err(io::Error::other("confined export source depth exceeds 32"));
}
for entry in source.entries()? {
if entry.is_file {
match source.open_file(std::slice::from_ref(&entry.name)) {
Ok(opened) => {
let bytes = copy_opened_into_confined(opened, destination, &entry.name)?;
summary.bytes += bytes;
summary.files += 1;
}
Err(_) => continue,
}
} else if entry.is_dir {
let source_child = match source.open_dirs(std::slice::from_ref(&entry.name)) {
Ok(child) => child,
Err(_) => continue,
};
let dest_child = destination.child_dir(&entry.name)?;
export_tree_from_handles(&source_child, &dest_child, summary, depth + 1)?;
}
}
Ok(())
}
#[cfg(not(unix))]
pub fn export(_roots: &RetentionRoots, _dest: &Path) -> io::Result<ExportSummary> {
Err(io::Error::other(
"confined export requires a unix platform (directory-handle-relative creation)",
))
}
pub fn forget(roots: &RetentionRoots, domain: RetentionDomain, id: &str) -> io::Result<()> {
match domain {
RetentionDomain::Sessions => {
let file = sanitize_file_id(id)?;
let heads = chain_heads(roots).map_err(|e| io::Error::other(e.to_string()))?;
if heads.contains(&id.to_string()) {
return Err(io::Error::other(format!(
"session {id} is the head of a named chain — delete or repoint the \
chain first (retention never leaves chains dangling)"
)));
}
let path = roots
.sessions_dir()
.join(format!("{}.json", file.display()));
fs::remove_file(path)?;
for ext in PAIRED_SESSION_EXTENSIONS {
let paired = roots
.sessions_dir()
.join(format!("{}.{ext}", file.display()));
match fs::remove_file(paired) {
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
other => other?,
}
}
crate::core::tool_spill::remove_session_spill(&roots.spill_dir(), id)?;
Ok(())
}
RetentionDomain::Traces => {
#[cfg(unix)]
{
let mut components = crate::private_fs::validated_relative_components(id)
.map_err(|_| io::Error::other(format!("invalid artifact id: {id:?}")))?;
let name = components
.pop()
.ok_or_else(|| io::Error::other(format!("invalid artifact id: {id:?}")))?;
let root = crate::private_fs::ConfinedDir::open_root(&roots.cache_dir)?;
let dir = root.open_dirs(&components)?;
dir.remove_file(&name)
}
#[cfg(not(unix))]
{
fs::remove_file(roots.cache_dir.join(sanitize_file_id(id)?))
}
}
RetentionDomain::Logs => {
let file = sanitize_file_id(id)?;
if !id.starts_with("synaps.log") {
return Err(io::Error::other(
"invalid log id: must start with synaps.log",
));
}
fs::remove_file(roots.config_dir.join(file))
}
RetentionDomain::Memory => {
let (namespace, record_id) = id.split_once(':').ok_or_else(|| {
io::Error::other("invalid memory id: use \"<namespace>:<record-id>\"")
})?;
let namespace = sanitize_file_id(namespace)?;
if !crate::memory::store::valid_import_id(record_id) {
return Err(io::Error::other(format!(
"invalid memory record id: {record_id:?}"
)));
}
let path = roots
.memory_dir()
.join(format!("{}.jsonl", namespace.display()));
fs::metadata(&path)?;
let _lock = lock_notes(&path)?;
let raw = fs::read(&path)?;
rewrite_read_done();
let mut live = false;
let mut tombstoned = false;
for line in notes_lines(&raw).flatten() {
let Ok(value) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
continue;
};
if value["tombstone"].as_str() == Some(record_id) {
tombstoned = true;
} else if value["id"].as_str() == Some(record_id) {
live = true;
}
}
if !live || tombstoned {
return Err(io::Error::other(format!(
"record {record_id} not present as a live record in namespace {}",
namespace.display()
)));
}
let line = serde_json::json!({
"tombstone": record_id,
"timestamp_ms": crate::epoch_millis(),
});
let mut body = raw;
if !body.ends_with(b"\n") && !body.is_empty() {
body.push(b'\n');
}
body.extend_from_slice(line.to_string().as_bytes());
body.push(b'\n');
crate::core::private_fs::write_atomic_private(&path, &body)
.map_err(io::Error::from)
}
}
}
fn sanitize_file_id(id: &str) -> io::Result<PathBuf> {
if id.is_empty() || id.contains('/') || id.contains('\\') || id.contains("..") {
return Err(io::Error::other(format!("invalid artifact id: {id:?}")));
}
Ok(PathBuf::from(id))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::memory::store::{self, MemoryRecord};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::Duration;
use tempfile::TempDir;
fn roots(tmp: &TempDir) -> RetentionRoots {
RetentionRoots {
config_dir: tmp.path().join("config"),
base_dir: tmp.path().join("base"),
cache_dir: tmp.path().join("cache"),
}
}
fn note(ns: &str, id: &str, timestamp_ms: u64) -> MemoryRecord {
let mut record = store::new_record(ns, format!("content of {id}"), vec![], None);
record.id = Some(id.to_string());
record.timestamp_ms = timestamp_ms;
record
}
fn save(roots: &RetentionRoots, ns: &str, id: &str, timestamp_ms: u64) {
store::append_to(&roots.base_dir, ¬e(ns, id, timestamp_ms)).unwrap();
}
fn notes_file(roots: &RetentionRoots, ns: &str) -> PathBuf {
roots.memory_dir().join(format!("{ns}.jsonl"))
}
fn ids(path: &Path) -> Vec<String> {
fs::read_to_string(path)
.unwrap()
.lines()
.filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
.filter_map(|value| value["id"].as_str().map(String::from))
.collect()
}
const AGE: RetentionPolicy = RetentionPolicy {
max_age_days: Some(1),
max_disk_bytes: None,
include_notes: true,
};
fn save_during_next_rewrite(
roots: &RetentionRoots,
ns: &'static str,
id: &'static str,
) -> Arc<Mutex<Option<JoinHandle<()>>>> {
let slot = Arc::new(Mutex::new(None));
let (left, roots, path) = (slot.clone(), roots.clone(), notes_file(roots, ns));
AFTER_REWRITE_READ.with(|hook| {
*hook.borrow_mut() = Some(Box::new(move || {
let saver = {
let roots = roots.clone();
std::thread::spawn(move || save(&roots, ns, id, crate::epoch_millis()))
};
std::thread::sleep(Duration::from_millis(150));
assert!(!saver.is_finished(), "the save waits for the rewrite");
assert!(!ids(&path).contains(&id.to_string()));
*left.lock().unwrap() = Some(saver);
}));
});
slot
}
fn join_save(slot: &Arc<Mutex<Option<JoinHandle<()>>>>) {
let saver = slot.lock().unwrap().take();
saver.expect("the rewriter read the file").join().unwrap();
}
#[test]
fn a_save_waits_for_a_rewrite_and_its_note_is_in_the_file_after_it() {
let now = crate::epoch_millis();
let tmp = TempDir::new().unwrap();
let roots = roots(&tmp);
save(&roots, "ns", "mem-old", 1);
save(&roots, "ns", "mem-new", now);
let late = save_during_next_rewrite(&roots, "ns", "mem-late");
let outcome = sweep_at(&roots, &AGE, now).unwrap();
join_save(&late);
assert_eq!(outcome.memory_records_dropped, 1);
assert_eq!(ids(¬es_file(&roots, "ns")), ["mem-new", "mem-late"]);
let tmp = TempDir::new().unwrap();
let roots = self::roots(&tmp);
save(&roots, "ns", "mem-old", 1);
save(&roots, "ns", "mem-new", 2);
let late = save_during_next_rewrite(&roots, "ns", "mem-late");
assert_eq!(evict_oldest_memory_records(&roots, 1).unwrap(), 1);
join_save(&late);
assert_eq!(ids(¬es_file(&roots, "ns")), ["mem-new", "mem-late"]);
let tmp = TempDir::new().unwrap();
let roots = self::roots(&tmp);
save(&roots, "ns", "mem-a", now);
let late = save_during_next_rewrite(&roots, "ns", "mem-late");
forget(&roots, RetentionDomain::Memory, "ns:mem-a").unwrap();
join_save(&late);
let path = notes_file(&roots, "ns");
assert_eq!(ids(&path), ["mem-a", "mem-late"]);
assert!(fs::read_to_string(&path)
.unwrap()
.contains(r#""tombstone":"mem-a""#));
}
#[test]
fn saves_beside_repeated_rewrites_lose_no_note() {
const SAVERS: usize = 4;
const SAVES: usize = 40;
let tmp = TempDir::new().unwrap();
let roots = roots(&tmp);
save(&roots, "ns", "mem-first", crate::epoch_millis());
let done = Arc::new(AtomicBool::new(false));
let rewriter = {
let (roots, done) = (roots.clone(), done.clone());
std::thread::spawn(move || {
let mut rewrites = 0usize;
let mut round = 0usize;
while !done.load(Ordering::Acquire) || round < 3 {
let now = crate::epoch_millis();
let victim = format!("mem-victim-{round}");
save(&roots, "ns", &victim, now);
save(&roots, "ns", &format!("mem-aged-{round}"), 1);
forget(&roots, RetentionDomain::Memory, &format!("ns:{victim}")).unwrap();
let outcome = sweep_at(&roots, &AGE, now).unwrap();
assert_eq!(outcome.memory_records_dropped, 2, "round {round}");
rewrites += 2;
round += 1;
}
rewrites
})
};
let savers: Vec<_> = (0..SAVERS)
.map(|saver| {
let roots = roots.clone();
std::thread::spawn(move || {
for n in 0..SAVES {
save(&roots, "ns", &format!("mem-{saver}-{n}"), crate::epoch_millis());
std::thread::sleep(Duration::from_millis(1));
}
})
})
.collect();
for saver in savers {
saver.join().unwrap();
}
done.store(true, Ordering::Release);
assert!(rewriter.join().unwrap() >= 6);
let mut found = ids(¬es_file(&roots, "ns"));
found.sort();
let mut expected: Vec<String> = (0..SAVERS)
.flat_map(|saver| (0..SAVES).map(move |n| format!("mem-{saver}-{n}")))
.collect();
expected.push("mem-first".into());
expected.sort();
assert_eq!(found, expected);
}
#[test]
fn a_rewrite_that_cannot_get_the_lock_fails_and_rewrites_nothing() {
let tmp = TempDir::new().unwrap();
let roots = roots(&tmp);
let now = crate::epoch_millis();
save(&roots, "ns", "mem-old", 1);
save(&roots, "ns", "mem-new", now);
let path = notes_file(&roots, "ns");
let before = fs::read(&path).unwrap();
let held = store::lock_notes_file(&path).unwrap();
store::set_notes_lock_wait(Some(Duration::from_millis(60)));
let swept = sweep_at(&roots, &AGE, now).unwrap_err();
let evicted = evict_oldest_memory_records(&roots, 1).unwrap_err();
let forgotten = forget(&roots, RetentionDomain::Memory, "ns:mem-new").unwrap_err();
store::set_notes_lock_wait(None);
for message in [swept.to_string(), evicted.to_string(), forgotten.to_string()] {
assert!(message.contains("ns.jsonl is locked by another writer"), "{message}");
}
assert_eq!(fs::read(&path).unwrap(), before);
assert!(!path.with_extension("jsonl.tmp").exists());
drop(held);
assert_eq!(sweep_at(&roots, &AGE, now).unwrap().memory_records_dropped, 1);
forget(&roots, RetentionDomain::Memory, "ns:mem-new").unwrap();
}
#[test]
fn a_tombstone_outlives_every_rewrite() {
use crate::memory::store::{
import_line_in, ImportOutcome, ImportRun, ImportTarget, MemoryProvenance,
MemoryRetention, MemorySensitivity, NewMemoryRecord, ProjectScope,
};
let tmp = TempDir::new().unwrap();
let roots = roots(&tmp);
let project = tmp.path().join("project");
fs::create_dir_all(&project).unwrap();
let scope = ProjectScope::for_root(&project).unwrap();
let ns = scope.namespace();
let path = notes_file(&roots, &ns);
let stored = |content: &str| {
let new = NewMemoryRecord {
content: content.to_string(),
tags: vec![],
provenance: MemoryProvenance {
source: "model".into(),
session: None,
},
sensitivity: MemorySensitivity::Normal,
retention: MemoryRetention::Standard,
};
let record = store::store_record_in(&roots.base_dir, &scope, new).unwrap();
std::thread::sleep(Duration::from_millis(2)); record.id.unwrap()
};
let inode = || {
#[cfg(unix)]
{
std::os::unix::fs::MetadataExt::ino(&fs::metadata(&path).unwrap())
}
#[cfg(not(unix))]
{
0u64
}
};
let forgotten = stored("forgotten");
let exported = fs::read_to_string(&path).unwrap().trim().to_string();
let (second, third, fourth) = (stored("second"), stored("third"), stored("fourth"));
store::forget_in(&roots.base_dir, &scope, &forgotten).unwrap();
let tombstone = fs::read_to_string(&path)
.unwrap()
.lines()
.last()
.unwrap()
.to_string();
assert!(tombstone.contains(&format!(r#""tombstone":"{forgotten}""#)));
let check = |live: &[&String]| {
let text = fs::read_to_string(&path).unwrap();
assert_eq!(text.lines().filter(|line| *line == tombstone).count(), 1);
let live: Vec<String> = live.iter().map(|id| id.to_string()).collect();
assert_eq!(ids(&path), live);
let imported = import_line_in(
&roots.base_dir,
&exported,
ImportTarget::AsRecorded,
&mut ImportRun::new(false),
)
.unwrap();
assert_eq!(imported.outcome.name(), ImportOutcome::KeptForgotten.name());
assert_eq!(fs::read_to_string(&path).unwrap(), text, "nothing imported");
};
#[cfg(unix)]
let before = inode();
let now = crate::epoch_millis();
let none = RetentionPolicy::default();
assert_eq!(sweep_at(&roots, &none, now).unwrap().memory_records_dropped, 1);
#[cfg(unix)]
assert_ne!(inode(), before, "the sweep rewrote the file");
check(&[&second, &third, &fourth]);
let before = inode();
assert_eq!(sweep_at(&roots, &none, now).unwrap().memory_records_dropped, 0);
assert_eq!(inode(), before, "a sweep with nothing to drop does not rewrite");
check(&[&second, &third, &fourth]);
assert_eq!(evict_oldest_memory_records(&roots, 1).unwrap(), 1);
#[cfg(unix)]
assert_ne!(inode(), before, "the eviction rewrote the file");
check(&[&third, &fourth]);
store::forget_in(&roots.base_dir, &scope, &third).unwrap();
check(&[&third, &fourth]);
#[cfg(unix)]
let before = inode();
forget(&roots, RetentionDomain::Memory, &format!("{ns}:{fourth}")).unwrap();
#[cfg(unix)]
assert_ne!(inode(), before, "the forget rewrote the file");
check(&[&third, &fourth]);
let text = fs::read_to_string(&path).unwrap();
for id in [&forgotten, &third, &fourth] {
assert!(text.contains(&format!(r#""tombstone":"{id}""#)), "{id}");
}
assert_eq!(sweep_at(&roots, &none, now).unwrap().memory_records_dropped, 2);
check(&[]);
let text = fs::read_to_string(&path).unwrap();
assert_eq!(text.lines().count(), 3);
assert_eq!(text.matches(r#""tombstone":"#).count(), 3);
}
#[test]
fn a_line_that_is_not_utf8_does_not_stop_a_rewriter() {
let torn = b"{\"namespace\":\"ns\",\"content\":\"caf\xc3";
let file_with_fragment = |roots: &RetentionRoots| {
save(roots, "ns", "mem-a", 10);
let path = notes_file(roots, "ns");
let mut bytes = fs::read(&path).unwrap();
bytes.extend_from_slice(torn);
fs::write(&path, bytes).unwrap();
save(roots, "ns", "mem-b", 20);
save(roots, "ns", "mem-c", crate::epoch_millis());
assert!(fs::read_to_string(&path).is_err(), "the file is not UTF-8");
path
};
let has_fragment = |path: &Path| {
let bytes = fs::read(path).unwrap();
bytes.windows(torn.len()).any(|window| window == torn)
};
let tmp = TempDir::new().unwrap();
let roots = roots(&tmp);
let path = file_with_fragment(&roots);
let before = fs::read(&path).unwrap();
forget(&roots, RetentionDomain::Memory, "ns:mem-b").unwrap();
let after = fs::read(&path).unwrap();
assert!(after.starts_with(&before));
assert!(has_fragment(&path));
let added = std::str::from_utf8(&after[before.len()..]).unwrap();
assert!(added.contains(r#""tombstone":"mem-b""#) && added.ends_with('\n'));
assert_eq!(added.lines().count(), 1);
let outcome = sweep_at(&roots, &RetentionPolicy::default(), 30).unwrap();
assert_eq!(outcome.memory_records_dropped, 2);
assert!(!has_fragment(&path));
assert_eq!(ids(&path), ["mem-a", "mem-c"]);
assert!(fs::read_to_string(&path).unwrap().contains(r#""tombstone":"mem-b""#));
let tmp = TempDir::new().unwrap();
let roots = self::roots(&tmp);
let path = file_with_fragment(&roots);
assert_eq!(evict_oldest_memory_records(&roots, 1).unwrap(), 1);
assert!(!has_fragment(&path));
assert_eq!(ids(&path), ["mem-b", "mem-c"]);
let lines: Vec<_> = notes_lines(b"a\r\n\n\xff\nb").collect();
assert_eq!(lines, [Some("a"), Some(""), None, Some("b")]);
assert_eq!(notes_lines(b"").count(), 0);
assert_eq!(notes_lines(b"a\n").collect::<Vec<_>>(), [Some("a")]);
assert_eq!(notes_lines(b"\n").collect::<Vec<_>>(), [Some("")]);
}
#[test]
fn an_age_sweep_keeps_saved_notes_unless_it_is_told_to_include_them() {
use crate::memory::store::MemoryRetention;
let tmp = TempDir::new().unwrap();
let roots = roots(&tmp);
let now = 1_000 * DAY_MS;
let with_class = |id: &str, timestamp_ms: u64, class: MemoryRetention| {
let mut record = note("ns", id, timestamp_ms);
record.retention = Some(class);
store::append_to(&roots.base_dir, &record).unwrap();
};
with_class("mem-standard-old", DAY_MS, MemoryRetention::Standard);
save(&roots, "ns", "mem-classless-old", 2 * DAY_MS);
with_class("mem-expired", 900 * DAY_MS, MemoryRetention::MaxAgeDays(30));
with_class("mem-not-expired", 990 * DAY_MS, MemoryRetention::MaxAgeDays(30));
with_class("mem-standard-new", 999 * DAY_MS, MemoryRetention::Standard);
let path = notes_file(&roots, "ns");
let age = |include_notes| RetentionPolicy {
max_age_days: Some(5),
max_disk_bytes: None,
include_notes,
};
let outcome = sweep_at(&roots, &age(false), now).unwrap();
assert_eq!(outcome.memory_records_dropped, 1, "the note past its own expiry");
assert_eq!(
ids(&path),
["mem-standard-old", "mem-classless-old", "mem-not-expired", "mem-standard-new"]
);
assert_eq!(sweep_at(&roots, &age(false), now).unwrap().memory_records_dropped, 0);
let outcome = sweep_at(&roots, &age(true), now).unwrap();
assert_eq!(outcome.memory_records_dropped, 3);
assert_eq!(ids(&path), ["mem-standard-new"]);
let none = RetentionPolicy {
include_notes: true,
..Default::default()
};
assert_eq!(sweep_at(&roots, &none, now).unwrap().memory_records_dropped, 0);
with_class("mem-standard-newer", now, MemoryRetention::Standard);
let total = inspect(&roots).unwrap().total_bytes;
let budget = |include_notes| RetentionPolicy {
max_disk_bytes: Some(total - 1),
include_notes,
..Default::default()
};
assert!(matches!(
sweep_at(&roots, &budget(false), now),
Err(RetentionError::BudgetUnmet { notes_kept: true, .. })
));
assert_eq!(ids(&path), ["mem-standard-new", "mem-standard-newer"]);
assert_eq!(sweep_at(&roots, &budget(true), now).unwrap().memory_records_dropped, 1);
assert_eq!(ids(&path), ["mem-standard-newer"]);
}
#[test]
fn a_note_with_an_imported_id_can_be_forgotten_from_the_command_line() {
let tmp = TempDir::new().unwrap();
let roots = roots(&tmp);
for id in ["note:2024/a", "note:2024/ab", "mem-x"] {
save(&roots, "ns", id, 5);
}
let path = notes_file(&roots, "ns");
forget(&roots, RetentionDomain::Memory, "ns:note:2024/a").unwrap();
let text = fs::read_to_string(&path).unwrap();
assert_eq!(text.matches("tombstone").count(), 1);
assert!(text.contains(r#""tombstone":"note:2024/a""#));
let again = forget(&roots, RetentionDomain::Memory, "ns:note:2024/a").unwrap_err();
assert!(again.to_string().contains("not present as a live record"), "{again}");
let part = forget(&roots, RetentionDomain::Memory, "ns:note:2024").unwrap_err();
assert!(part.to_string().contains("not present as a live record"), "{part}");
let before = fs::read(&path).unwrap();
let long = format!("ns:{}", "i".repeat(129));
for bad in ["ns:", "ns:has space", "ns:tab\there", long.as_str()] {
let error = forget(&roots, RetentionDomain::Memory, bad).unwrap_err();
assert!(error.to_string().starts_with("invalid memory record id"), "{error}");
}
assert_eq!(fs::read(&path).unwrap(), before);
}
#[test]
fn the_lock_file_is_not_a_notes_file() {
let tmp = TempDir::new().unwrap();
let roots = roots(&tmp);
let now = crate::epoch_millis();
save(&roots, "ns", "mem-old", 1);
save(&roots, "ns", "mem-new", now);
let path = notes_file(&roots, "ns");
let lock = roots.memory_dir().join("ns.jsonl.lock");
assert_eq!(store::notes_lock_path(&path), lock);
assert!(lock.is_file());
assert_eq!(store::list_namespaces_in(&roots.base_dir).unwrap(), ["ns"]);
let counted = |roots: &RetentionRoots| {
let report = inspect(roots).unwrap();
let memory = report
.domains
.iter()
.find(|d| d.domain == RetentionDomain::Memory)
.unwrap();
(memory.files, memory.bytes)
};
assert_eq!(counted(&roots), (1, fs::metadata(&path).unwrap().len()));
assert_eq!(sweep_at(&roots, &AGE, now).unwrap().memory_records_dropped, 1);
assert_eq!(evict_oldest_memory_records(&roots, 1).unwrap(), 1);
assert!(lock.is_file(), "the lock file is never removed or renamed");
assert_eq!(fs::metadata(&lock).unwrap().len(), 0);
assert_eq!(counted(&roots), (1, 0));
assert_eq!(store::list_namespaces_in(&roots.base_dir).unwrap(), ["ns"]);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
assert_eq!(fs::metadata(&lock).unwrap().permissions().mode() & 0o777, 0o600);
let dest = tmp.path().join("export");
let summary = export(&roots, &dest).unwrap();
assert_eq!(summary.files, 1);
let exported: Vec<_> = fs::read_dir(dest.join("memory"))
.unwrap()
.map(|entry| entry.unwrap().file_name())
.collect();
assert_eq!(exported, ["ns.jsonl"]);
}
}
}