use std::path::{Path, PathBuf};
use crate::conversation::{
header_line, message_id_timestamp_ms, parse_header, parse_message_line, read_header,
record_messages, CONVERSATION_ROOT_DIR, RECORD_FILE_NAME,
};
use crate::formation::FormationId;
use crate::formation_launch::FormationMarker;
use crate::harness_session::{read_entry, HARNESS_SESSION_FILE_NAME};
pub use crate::conversation::{RecordHeader, TruncationMarker};
pub use crate::harness_session::HarnessSessionEntry;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PrunedSession {
pub name: String,
pub reason: &'static str,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PrunedFormation {
pub formation_id: FormationId,
pub path: PathBuf,
pub reason: &'static str,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PrunedRecord {
pub context_id: String,
pub path: PathBuf,
pub reason: &'static str,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TruncationOutcome {
pub dropped: u64,
pub kept: u64,
pub oldest_surviving_id: String,
pub last_dropped_id: String,
pub blobs_removed: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MessageStatus {
Present { position: u64, total: u64 },
Truncated {
dropped: u64,
oldest_surviving_id: String,
},
Unknown,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MessageLocation {
pub record: String,
pub context_id: String,
pub path: PathBuf,
pub status: MessageStatus,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecordSummary {
pub record: String,
pub context_id: String,
pub path: PathBuf,
pub messages: u64,
pub bytes: u64,
pub last_touched_ms: Option<u64>,
pub capsule: Option<String>,
pub truncation: Option<TruncationMarker>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemovedRecord {
pub record: String,
pub context_id: String,
pub path: PathBuf,
pub messages: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ContextDir {
pub record: String,
pub context_id: String,
pub path: PathBuf,
}
#[derive(Debug, Clone, PartialEq)]
pub struct ContextContents {
pub header: Option<RecordHeader>,
pub messages: Vec<serde_json::Value>,
pub harness_session: Option<HarnessSessionEntry>,
}
pub fn session_id_timestamp_ms(name: &str) -> Option<u64> {
uuid_v7_timestamp_ms(name, "ses_")
}
pub(crate) fn uuid_v7_timestamp_ms(id: &str, prefix: &str) -> Option<u64> {
let hex = id.strip_prefix(prefix)?;
u64::from_str_radix(hex.get(..12)?, 16).ok()
}
pub(crate) fn created_workdirs(
workdir_flag: Option<&std::path::Path>,
manifest_dir: &std::path::Path,
) -> Option<std::path::PathBuf> {
workdir_flag.is_none().then(|| manifest_dir.join("workdir"))
}
pub fn prune_sessions(
records_group: &crate::workdir_fs::WorkdirRoot,
created_workdirs: Option<&crate::workdir_fs::WorkdirRoot>,
current_session_id: &str,
policy: &murmur_artifact::TraceRetainConfig,
now_ms: u64,
) -> Vec<PrunedSession> {
if policy.max_sessions.is_none() && policy.max_age_secs.is_none() {
return Vec::new();
}
let mut entries: Vec<String> = match records_group.list_dir("") {
Ok(Some(listed)) => listed
.into_iter()
.filter(|(_, kind)| *kind == crate::workdir_fs::EntryKind::Directory)
.map(|(name, _)| name)
.filter(|name| session_id_timestamp_ms(name).is_some())
.collect(),
Ok(None) | Err(_) => return Vec::new(),
};
entries.sort_unstable_by(|a, b| b.cmp(a));
let age_floor_ms = policy
.max_age_secs
.map(|secs| now_ms.saturating_sub(secs.saturating_mul(1000)));
let mut pruned = Vec::new();
for (rank, name) in entries.iter().enumerate() {
if name.as_str() >= current_session_id {
continue;
}
let over_age = age_floor_ms.is_some_and(|floor| {
session_id_timestamp_ms(name).is_some_and(|minted| minted < floor)
});
let over_count = policy.max_sessions.is_some_and(|max| rank >= max as usize);
if !over_age && !over_count {
continue;
}
let reason = if over_age {
crate::trace::RETENTION_REASON_MAX_AGE
} else {
crate::trace::RETENTION_REASON_MAX_SESSIONS
};
if records_group.remove_dir_all(name).is_ok() {
if let Some(workdirs) = created_workdirs {
let _ = workdirs.remove_dir_all(name);
}
pruned.push(PrunedSession {
name: name.clone(),
reason,
});
}
}
pruned
}
pub fn prune_formations(
formations_dir: &Path,
current: &FormationId,
project_dir: &Path,
policy: &murmur_artifact::TraceRetainConfig,
now_ms: u64,
is_live: impl Fn(&FormationId, &FormationMarker) -> bool,
) -> Vec<PrunedFormation> {
if policy.max_sessions.is_none() && policy.max_age_secs.is_none() {
return Vec::new();
}
let Ok(entries) = std::fs::read_dir(formations_dir) else {
return Vec::new();
};
let project_dir = crate::formation_launch::canonical_project_dir(project_dir);
let mut owned: Vec<(FormationId, FormationMarker)> = entries
.filter_map(Result::ok)
.filter_map(|entry| {
let id = FormationId::parse(entry.file_name().to_str()?).ok()?;
let path = entry.path();
if !std::fs::symlink_metadata(&path).ok()?.is_dir() {
return None;
}
let marker = FormationMarker::read(&path)?;
(crate::formation_launch::canonical_project_dir(&marker.project_dir) == project_dir)
.then_some((id, marker))
})
.collect();
owned.sort_unstable_by(|(a, _), (b, _)| b.as_str().cmp(a.as_str()));
let age_floor_ms = policy
.max_age_secs
.map(|secs| now_ms.saturating_sub(secs.saturating_mul(1000)));
let mut pruned = Vec::new();
for (rank, (id, marker)) in owned.iter().enumerate() {
if id.as_str() >= current.as_str() {
continue;
}
let over_age = age_floor_ms.is_some_and(|floor| id.minted_at_ms() < floor);
let over_count = policy.max_sessions.is_some_and(|max| rank >= max as usize);
if !over_age && !over_count {
continue;
}
if is_live(id, marker) {
continue;
}
let reason = if over_age {
crate::trace::RETENTION_REASON_MAX_AGE
} else {
crate::trace::RETENTION_REASON_MAX_SESSIONS
};
let path = formations_dir.join(id.as_str());
let records = crate::records_dir::records_group(&path).ok();
if std::fs::remove_dir_all(&path).is_ok() {
if let Some(records) = records {
let _ = std::fs::remove_dir_all(records);
}
pruned.push(PrunedFormation {
formation_id: id.clone(),
path,
reason,
});
}
}
pruned
}
pub fn prune_records(
record_root: &Path,
capsule_name: &str,
current_context_id: Option<&str>,
policy: &murmur_artifact::ContextRetainConfig,
now_ms: u64,
) -> Vec<PrunedRecord> {
let Some(max_age_secs) = policy.max_age_secs else {
return Vec::new();
};
let floor_ms = now_ms.saturating_sub(max_age_secs.saturating_mul(1000));
let mut pruned = Vec::new();
for (context_id, dir) in context_dirs(record_root) {
if Some(context_id.as_str()) == current_context_id {
continue;
}
let file = dir.join(RECORD_FILE_NAME);
match read_header(&file) {
Some(header) if header.capsule == capsule_name => {}
_ => continue,
}
let Some(touched_ms) = last_touched_ms(&file) else {
continue;
};
if touched_ms >= floor_ms {
continue;
}
if std::fs::remove_dir_all(&dir).is_ok() {
pruned.push(PrunedRecord {
context_id,
path: dir,
reason: crate::trace::RETENTION_REASON_MAX_AGE,
});
}
}
pruned
}
pub fn truncate_record(path: &Path, keep: u32, capsule: &str) -> Result<TruncationOutcome, String> {
match stage_truncation(path, keep, capsule, now_ms())? {
None => Ok(TruncationOutcome {
dropped: 0,
kept: count_messages(path),
oldest_surviving_id: String::new(),
last_dropped_id: String::new(),
blobs_removed: 0,
}),
Some((staged, mut outcome)) => {
staged.commit()?;
if let Some(dir) = path.parent() {
crate::record_seal::seal_conversation(dir, crate::record_seal::CLI_SEALER);
}
outcome.blobs_removed = sweep_orphan_blobs(path);
Ok(outcome)
}
}
}
fn sweep_orphan_blobs(path: &Path) -> u64 {
let Some(blobs) = path
.parent()
.map(|dir| dir.join(crate::trace_blobs::BLOB_DIR_NAME))
else {
return 0;
};
let Ok(entries) = std::fs::read_dir(&blobs) else {
return 0;
};
let Ok(raw) = std::fs::read_to_string(path) else {
return 0;
};
let referenced: std::collections::HashSet<String> = raw
.lines()
.filter_map(parse_message_line)
.flat_map(|message| {
crate::images::referenced_blobs(&message)
.map(str::to_string)
.collect::<Vec<_>>()
})
.collect();
let mut removed = 0;
for entry in entries.flatten() {
let name = entry.file_name();
let Some(name) = name.to_str() else {
continue;
};
if !murmur_artifact::is_sha256_hex(name) || referenced.contains(name) {
continue;
}
if std::fs::remove_file(entry.path()).is_ok() {
removed += 1;
}
}
removed
}
pub(crate) fn stage_truncation(
path: &Path,
keep: u32,
capsule: &str,
at_ms: u64,
) -> Result<Option<(StagedRewrite, TruncationOutcome)>, String> {
let raw = match std::fs::read_to_string(path) {
Ok(raw) => raw,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(err) => return Err(format!("{}: {err}", path.display())),
};
let existing = read_header(path);
let lines: Vec<&str> = raw.lines().collect();
let message_indices: Vec<usize> = lines
.iter()
.enumerate()
.filter(|(_, line)| parse_message_line(line).is_some())
.map(|(index, _)| index)
.collect();
let total = message_indices.len();
let keep = keep as usize;
if total <= keep {
return Ok(None);
}
let dropped = total - keep;
let first_kept_index = message_indices[dropped];
let id_of = |index: usize| -> String {
parse_message_line(lines[index])
.as_ref()
.and_then(|message| crate::agent::message_id(message).map(str::to_string))
.unwrap_or_default()
};
let last_dropped_id = id_of(message_indices[dropped - 1]);
let oldest_surviving_id = message_indices
.get(dropped)
.map(|i| id_of(*i))
.unwrap_or_default();
let header = RecordHeader {
kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
capsule: existing
.as_ref()
.map(|h| h.capsule.clone())
.unwrap_or_else(|| capsule.to_string()),
created_ms: existing.as_ref().map(|h| h.created_ms).unwrap_or(at_ms),
truncated: Some(TruncationMarker {
dropped: existing
.as_ref()
.and_then(|h| h.truncated.as_ref())
.map(|marker| marker.dropped)
.unwrap_or(0)
+ dropped as u64,
oldest_surviving_id: oldest_surviving_id.clone(),
last_dropped_id: last_dropped_id.clone(),
at_ms,
}),
};
let mut contents = header_line(&header)?;
for line in &lines[first_kept_index..] {
contents.push_str(line);
contents.push('\n');
}
let staged = StagedRewrite::stage_with_mode(
path,
contents.as_bytes(),
Some(crate::conversation::RECORD_FILE_MODE),
)?;
Ok(Some((
staged,
TruncationOutcome {
dropped: dropped as u64,
kept: keep as u64,
oldest_surviving_id,
last_dropped_id,
blobs_removed: 0,
},
)))
}
pub(crate) struct StagedRewrite {
target: PathBuf,
temp: PathBuf,
committed: bool,
}
impl StagedRewrite {
pub(crate) fn stage_with_mode(
target: &Path,
contents: &[u8],
mode: Option<u32>,
) -> Result<Self, String> {
use std::io::Write;
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
let dir = target
.parent()
.ok_or_else(|| format!("{} has no parent directory", target.display()))?;
let name = target
.file_name()
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_else(|| "record".to_string());
let temp = dir.join(format!(
".{name}.{}.{}.tmp",
std::process::id(),
uuid::Uuid::now_v7().simple()
));
let mut options = std::fs::OpenOptions::new();
options.write(true).create(true).truncate(true);
if let Some(mode) = mode {
options.mode(mode);
}
let mut file = options
.open(&temp)
.map_err(|err| format!("{}: {err}", temp.display()))?;
file.write_all(contents)
.and_then(|()| file.sync_all())
.map_err(|err| format!("{}: {err}", temp.display()))?;
if let Some(mode) = mode {
std::fs::set_permissions(&temp, std::fs::Permissions::from_mode(mode))
.map_err(|err| format!("{}: {err}", temp.display()))?;
}
Ok(Self {
target: target.to_path_buf(),
temp,
committed: false,
})
}
#[cfg(test)]
pub(crate) fn temp_path(&self) -> &Path {
&self.temp
}
pub(crate) fn commit(mut self) -> Result<(), String> {
std::fs::rename(&self.temp, &self.target)
.map_err(|err| format!("{}: {err}", self.target.display()))?;
self.committed = true;
Ok(())
}
}
impl Drop for StagedRewrite {
fn drop(&mut self) {
if !self.committed {
let _ = std::fs::remove_file(&self.temp);
}
}
}
pub fn list_records() -> Result<Vec<RecordSummary>, String> {
let root = conversations_root()?;
let mut summaries = Vec::new();
for record in child_dirs(&root) {
let store = root.join(&record);
for (context_id, dir) in context_dirs(&store) {
let path = dir.join(RECORD_FILE_NAME);
if !path.exists() {
continue;
}
let header = read_header(&path);
summaries.push(RecordSummary {
record: record.clone(),
context_id,
messages: count_messages(&path),
bytes: std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0),
last_touched_ms: last_touched_ms(&path),
capsule: header.as_ref().map(|h| h.capsule.clone()),
truncation: header.and_then(|h| h.truncated),
path,
});
}
}
summaries.sort_by(|a, b| {
a.record
.cmp(&b.record)
.then_with(|| a.context_id.cmp(&b.context_id))
});
Ok(summaries)
}
pub fn remove_record(record: &str, context_id: &str) -> Result<RemovedRecord, String> {
let dir = conversations_root()?.join(record).join(context_id);
if !dir.is_dir() {
return Err(format!("{} is not a context directory", dir.display()));
}
let messages = count_messages(&dir.join(RECORD_FILE_NAME));
std::fs::remove_dir_all(&dir).map_err(|err| format!("{}: {err}", dir.display()))?;
Ok(RemovedRecord {
record: record.to_string(),
context_id: context_id.to_string(),
path: dir,
messages,
})
}
pub fn locate_message(message_id: &str) -> Result<Vec<MessageLocation>, String> {
let minted_ms = message_id_timestamp_ms(message_id);
let mut found = Vec::new();
for summary in list_records()? {
let raw = std::fs::read_to_string(&summary.path).unwrap_or_default();
let mut position = 0u64;
let mut at = None;
for line in raw.lines() {
let Some(message) = parse_message_line(line) else {
continue;
};
position += 1;
if crate::agent::message_id(&message) == Some(message_id) {
at = Some(position);
}
}
let status = match (at, summary.truncation.as_ref()) {
(Some(position), _) => MessageStatus::Present {
position,
total: position.max(summary.messages),
},
(None, Some(marker)) => {
let dropped_at = message_id_timestamp_ms(&marker.last_dropped_id);
match (minted_ms, dropped_at) {
(Some(minted), Some(last)) if minted <= last => MessageStatus::Truncated {
dropped: marker.dropped,
oldest_surviving_id: marker.oldest_surviving_id.clone(),
},
_ => MessageStatus::Unknown,
}
}
(None, None) => MessageStatus::Unknown,
};
if status != MessageStatus::Unknown {
found.push(MessageLocation {
record: summary.record,
context_id: summary.context_id,
path: summary.path,
status,
});
}
}
Ok(found)
}
pub fn find_context_dirs(context_id: &str) -> Result<Vec<ContextDir>, String> {
Ok(find_context_dirs_in(&conversations_root()?, context_id))
}
fn find_context_dirs_in(root: &Path, context_id: &str) -> Vec<ContextDir> {
let mut found = Vec::new();
for record in child_dirs(root) {
for (name, path) in context_dirs(&root.join(&record)) {
if name != context_id {
continue;
}
if path.join(RECORD_FILE_NAME).exists() || path.join(HARNESS_SESSION_FILE_NAME).exists()
{
found.push(ContextDir {
record: record.clone(),
context_id: name,
path,
});
}
}
}
found
}
pub fn read_context_dir(dir: &Path) -> Result<ContextContents, String> {
let record = dir.join(RECORD_FILE_NAME);
let raw = match std::fs::read_to_string(&record) {
Ok(raw) => raw,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
Err(err) => return Err(format!("{}: {err}", record.display())),
};
Ok(ContextContents {
header: raw.lines().next().and_then(parse_header),
messages: record_messages(&raw, |_| {}),
harness_session: read_entry(&dir.join(HARNESS_SESSION_FILE_NAME)),
})
}
pub fn conversations_root() -> Result<PathBuf, String> {
Ok(crate::state_store::murmur_home_dir()?.join(CONVERSATION_ROOT_DIR))
}
pub fn now_ms() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|elapsed| elapsed.as_millis() as u64)
.unwrap_or(0)
}
fn child_dirs(dir: &Path) -> Vec<String> {
let mut names: Vec<String> = match std::fs::read_dir(dir) {
Ok(entries) => entries
.filter_map(Result::ok)
.filter(|entry| entry.path().is_dir())
.map(|entry| entry.file_name().to_string_lossy().into_owned())
.filter(|name| !name.starts_with('.'))
.collect(),
Err(_) => Vec::new(),
};
names.sort();
names
}
fn context_dirs(record_root: &Path) -> Vec<(String, PathBuf)> {
child_dirs(record_root)
.into_iter()
.map(|name| {
let path = record_root.join(&name);
(name, path)
})
.collect()
}
fn count_messages(path: &Path) -> u64 {
std::fs::read_to_string(path)
.map(|raw| {
raw.lines()
.filter(|line| parse_message_line(line).is_some())
.count() as u64
})
.unwrap_or(0)
}
fn last_touched_ms(path: &Path) -> Option<u64> {
std::fs::metadata(path)
.and_then(|metadata| metadata.modified())
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|elapsed| elapsed.as_millis() as u64)
}
#[cfg(test)]
mod tests {
use super::*;
use murmur_artifact::{ContextRetainConfig, TraceRetainConfig};
use serde_json::json;
fn ses(ms: u64, tail: u64) -> String {
format!("ses_{ms:012x}{tail:020x}")
}
fn msg(ms: u64, tail: u64) -> String {
format!("msg_{ms:012x}{tail:020x}")
}
fn message_line(id: &str, text: &str) -> String {
json!({"role": "user", "content": [{"type": "text", "text": text}], "id": id}).to_string()
}
fn make_sessions(root: &Path, names: &[String]) {
for name in names {
let dir = root.join(name);
std::fs::create_dir_all(dir.join("blobs")).unwrap();
std::fs::write(dir.join("trace.jsonl"), "{}\n").unwrap();
std::fs::write(dir.join("blobs").join("deadbeef"), b"body").unwrap();
}
}
fn names(root: &Path) -> Vec<String> {
let mut found: Vec<String> = std::fs::read_dir(root)
.unwrap()
.map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
.collect();
found.sort();
found
}
const NOW: u64 = 1_756_400_000_000;
#[test]
fn max_sessions_leaves_exactly_the_newest_n_including_the_current() {
let root = tempfile::tempdir().unwrap();
let all: Vec<String> = (1..=6).map(|n| ses(NOW - (6 - n) * 1000, n)).collect();
make_sessions(root.path(), &all);
let pruned = prune_sessions(
&crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
None,
&all[5],
&TraceRetainConfig {
max_sessions: Some(3),
max_age_secs: None,
},
NOW,
);
assert_eq!(
names(root.path()),
vec![all[3].clone(), all[4].clone(), all[5].clone()]
);
let mut removed: Vec<String> = pruned.iter().map(|p| p.name.clone()).collect();
removed.sort();
assert_eq!(
removed,
vec![all[0].clone(), all[1].clone(), all[2].clone()]
);
assert!(pruned
.iter()
.all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_SESSIONS));
for name in &all[..3] {
assert!(!root.path().join(name).exists(), "{name} went whole");
}
}
#[test]
fn max_age_reads_the_id_and_never_the_filesystem() {
let root = tempfile::tempdir().unwrap();
let old_a = ses(NOW - 7_200_000, 1);
let old_b = ses(NOW - 7_100_000, 2);
let recent = ses(NOW - 60_000, 3);
let current = ses(NOW, 4);
make_sessions(
root.path(),
&[
old_a.clone(),
old_b.clone(),
recent.clone(),
current.clone(),
],
);
let pruned = prune_sessions(
&crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
None,
¤t,
&TraceRetainConfig {
max_sessions: None,
max_age_secs: Some(900),
},
NOW,
);
assert_eq!(pruned.len(), 2, "{pruned:?}");
assert!(pruned
.iter()
.all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_AGE));
assert_eq!(names(root.path()), vec![recent, current]);
}
#[test]
fn both_trace_keys_keep_only_sessions_inside_both() {
let root = tempfile::tempdir().unwrap();
let old = [ses(NOW - 7_200_000, 1), ses(NOW - 7_100_000, 2)];
let recent = [
ses(NOW - 300_000, 3),
ses(NOW - 200_000, 4),
ses(NOW - 100_000, 5),
];
let current = ses(NOW, 6);
let mut all = old.to_vec();
all.extend(recent.iter().cloned());
all.push(current.clone());
make_sessions(root.path(), &all);
let pruned = prune_sessions(
&crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
None,
¤t,
&TraceRetainConfig {
max_sessions: Some(2),
max_age_secs: Some(900),
},
NOW,
);
assert_eq!(names(root.path()), vec![recent[2].clone(), current.clone()]);
assert_eq!(pruned.len(), 4);
for session in &pruned {
let expected = if old.contains(&session.name) {
crate::trace::RETENTION_REASON_MAX_AGE
} else {
crate::trace::RETENTION_REASON_MAX_SESSIONS
};
assert_eq!(session.reason, expected, "{session:?}");
}
}
#[test]
fn nothing_at_or_after_the_current_session_id_is_a_candidate() {
let root = tempfile::tempdir().unwrap();
let current = ses(NOW, 5);
let sibling = ses(NOW + 1, 6);
let older = ses(NOW - 1000, 4);
make_sessions(
root.path(),
&[current.clone(), sibling.clone(), older.clone()],
);
let pruned = prune_sessions(
&crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
None,
¤t,
&TraceRetainConfig {
max_sessions: Some(1),
max_age_secs: Some(1),
},
NOW,
);
assert_eq!(pruned.len(), 1);
assert_eq!(pruned[0].name, older);
assert_eq!(names(root.path()), vec![current.clone(), sibling]);
assert!(root.path().join(¤t).join("trace.jsonl").exists());
}
#[test]
fn a_pruned_session_takes_its_records_and_its_created_work_folder() {
let records = tempfile::tempdir().unwrap();
let workdirs = tempfile::tempdir().unwrap();
let all: Vec<String> = (1..=3).map(|n| ses(NOW - (3 - n) * 1000, n)).collect();
make_sessions(records.path(), &all);
for name in &all {
std::fs::create_dir_all(workdirs.path().join(name).join("out")).unwrap();
}
let workdir_only = ses(NOW - 9_000, 9);
std::fs::create_dir_all(workdirs.path().join(&workdir_only)).unwrap();
let pruned = prune_sessions(
&crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
&all[2],
&TraceRetainConfig {
max_sessions: Some(2),
max_age_secs: None,
},
NOW,
);
assert_eq!(pruned.len(), 1);
assert_eq!(pruned[0].name, all[0]);
assert_eq!(names(records.path()), vec![all[1].clone(), all[2].clone()]);
let mut kept = vec![all[1].clone(), all[2].clone(), workdir_only];
kept.sort();
assert_eq!(names(workdirs.path()), kept);
}
#[test]
fn retention_prunes_a_created_work_folder_and_never_the_operators() {
let records = tempfile::tempdir().unwrap();
let operator = tempfile::tempdir().unwrap();
let manifest_dir = tempfile::tempdir().unwrap();
let old = ses(NOW - 2000, 1);
let current = ses(NOW, 2);
let policy = TraceRetainConfig {
max_sessions: Some(1),
max_age_secs: None,
};
let open = |path: &std::path::Path| crate::workdir_fs::WorkdirRoot::open(path).unwrap();
make_sessions(records.path(), &[old.clone(), current.clone()]);
let planted = [
operator.path().join(".murmur").join(&old),
operator.path().join(&old),
];
for dir in &planted {
std::fs::create_dir_all(dir).unwrap();
std::fs::write(dir.join("keep"), b"x").unwrap();
}
assert_eq!(
created_workdirs(Some(operator.path()), manifest_dir.path()),
None
);
let pruned = prune_sessions(
&open(records.path()),
created_workdirs(Some(operator.path()), manifest_dir.path())
.map(|dir| open(&dir))
.as_ref(),
¤t,
&policy,
NOW,
);
assert_eq!(pruned.len(), 1);
assert_eq!(names(records.path()), vec![current.clone()]);
for dir in &planted {
assert!(dir.join("keep").is_file(), "{} survives", dir.display());
}
make_sessions(records.path(), std::slice::from_ref(&old));
let workdirs = created_workdirs(None, manifest_dir.path()).unwrap();
assert_eq!(workdirs, manifest_dir.path().join("workdir"));
std::fs::create_dir_all(workdirs.join(&old)).unwrap();
std::fs::create_dir_all(workdirs.join(¤t)).unwrap();
let pruned = prune_sessions(
&open(records.path()),
Some(&open(&workdirs)),
¤t,
&policy,
NOW,
);
assert_eq!(pruned.len(), 1);
assert!(!workdirs.join(&old).exists());
assert!(workdirs.join(¤t).exists());
}
#[test]
fn pruning_a_session_removes_its_temp_folder() {
use std::os::unix::fs::PermissionsExt;
let root = tempfile::tempdir().unwrap();
let old = ses(NOW - 2000, 1);
let current = ses(NOW, 2);
make_sessions(root.path(), &[old.clone(), current.clone()]);
let folder = root
.path()
.join(&old)
.join(crate::temp_dir::SESSION_TEMP_DIR_NAME);
let read_only = folder.join("read-only");
let sealed_shut = folder.join("sealed-shut");
std::fs::create_dir_all(&read_only).unwrap();
std::fs::create_dir_all(&sealed_shut).unwrap();
std::fs::write(read_only.join("file"), b"left behind").unwrap();
std::fs::set_permissions(&read_only, std::fs::Permissions::from_mode(0o555)).unwrap();
std::fs::set_permissions(&sealed_shut, std::fs::Permissions::from_mode(0o000)).unwrap();
let pruned = prune_sessions(
&crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
None,
¤t,
&TraceRetainConfig {
max_sessions: Some(1),
max_age_secs: None,
},
NOW,
);
assert_eq!(pruned.len(), 1, "{pruned:?}");
assert_eq!(pruned[0].name, old);
assert!(!root.path().join(&old).exists(), "the session went whole");
assert_eq!(names(root.path()), vec![current]);
}
#[test]
fn a_link_in_place_of_a_created_work_folder_is_not_followed() {
let records = tempfile::tempdir().unwrap();
let workdirs = tempfile::tempdir().unwrap();
let elsewhere = tempfile::tempdir().unwrap();
std::fs::write(elsewhere.path().join("keep"), b"x").unwrap();
let all: Vec<String> = (1..=2).map(|n| ses(NOW - (2 - n) * 1000, n)).collect();
make_sessions(records.path(), &all);
std::os::unix::fs::symlink(elsewhere.path(), workdirs.path().join(&all[0])).unwrap();
let pruned = prune_sessions(
&crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
&all[1],
&TraceRetainConfig {
max_sessions: Some(1),
max_age_secs: None,
},
NOW,
);
assert_eq!(pruned.len(), 1);
assert!(elsewhere.path().join("keep").exists());
}
#[test]
fn an_empty_policy_removes_nothing() {
let root = tempfile::tempdir().unwrap();
let all = [ses(NOW - 9_000_000, 1), ses(NOW, 2)];
make_sessions(root.path(), &all);
assert!(prune_sessions(
&crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
None,
&all[1],
&TraceRetainConfig {
max_sessions: None,
max_age_secs: None
},
NOW
)
.is_empty());
assert_eq!(names(root.path()), all.to_vec());
}
fn write_record(dir: &Path, capsule: Option<&str>, count: u64) -> PathBuf {
std::fs::create_dir_all(dir).unwrap();
let path = dir.join(RECORD_FILE_NAME);
let mut contents = String::new();
if let Some(capsule) = capsule {
contents.push_str(
&header_line(&RecordHeader {
kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
capsule: capsule.to_string(),
created_ms: NOW,
truncated: None,
})
.unwrap(),
);
}
for i in 1..=count {
contents.push_str(&message_line(
&msg(NOW - (count - i) * 1000, i),
&format!("m{i}"),
));
contents.push('\n');
}
std::fs::write(&path, contents).unwrap();
path
}
fn image_line(id: &str, shas: &[&str]) -> String {
let mut content = vec![serde_json::json!({"type": "text", "text": "shot"})];
for sha in shas {
content.push(serde_json::json!({
"type": "image",
"source": {"type": "file", "sha256": sha, "media_type": "image/png", "bytes": 4},
}));
}
serde_json::json!({"id": id, "role": "tool", "content": content}).to_string()
}
#[test]
fn truncation_removes_blobs_no_surviving_line_names() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join(RECORD_FILE_NAME);
let (old, shared, new) = ("a".repeat(64), "b".repeat(64), "c".repeat(64));
let lines = [
image_line(&msg(NOW - 3000, 1), &[&old, &shared]),
image_line(&msg(NOW - 2000, 2), &[&old]),
image_line(&msg(NOW - 1000, 3), &[&shared]),
image_line(&msg(NOW, 4), &[&new]),
];
std::fs::write(&path, lines.join("\n") + "\n").unwrap();
let blobs = dir.path().join("blobs");
std::fs::create_dir(&blobs).unwrap();
for name in [&old, &shared, &new, &"notes.txt".to_string()] {
std::fs::write(blobs.join(name), b"\x89PNG").unwrap();
}
let outcome = truncate_record(&path, 2, "capsule").unwrap();
assert_eq!(outcome.dropped, 2);
assert_eq!(outcome.blobs_removed, 1);
assert!(!blobs.join(&old).exists());
assert!(blobs.join(&shared).exists(), "a kept line still names it");
assert!(blobs.join(&new).exists());
assert!(blobs.join("notes.txt").exists());
}
#[test]
fn truncation_reseals_the_record() {
use crate::record_seal::{self, PublicKey, Verdict};
let home = tempfile::tempdir().unwrap();
record_seal::set_test_seal_home(Some(home.path().to_path_buf()));
let dir = tempfile::tempdir().unwrap();
let context = dir.path().join("capsule/ctx_1");
std::fs::create_dir_all(&context).unwrap();
let path = write_record(&context, Some("capsule"), 5);
let before = std::fs::read_to_string(&path).unwrap();
truncate_record(&path, 2, "capsule").unwrap();
let after = std::fs::read_to_string(&path).unwrap();
for kept in after
.lines()
.filter(|line| parse_message_line(line).is_some())
{
assert!(before.lines().any(|line| line == kept), "{kept}");
}
let root = crate::workdir_fs::WorkdirRoot::open(&context).unwrap();
let seal = record_seal::read_seal(&root, crate::conversation::RECORD_FILE_NAME)
.unwrap()
.expect("the truncation sealed the record");
assert_eq!(seal.stream, "conversation:capsule/ctx_1");
let key = PublicKey::load(&record_seal::public_key_path(home.path())).unwrap();
let report = record_seal::verify_listed(
&seal.stream,
after.as_bytes(),
false,
Some(&seal),
Some(&key),
None,
);
assert_eq!(report.verdict, Verdict::Verified, "{report:#?}");
}
#[test]
fn truncation_without_blobs_removes_nothing() {
let dir = tempfile::tempdir().unwrap();
let path = write_record(dir.path(), Some("capsule"), 5);
let outcome = truncate_record(&path, 2, "capsule").unwrap();
assert_eq!(outcome.dropped, 3);
assert_eq!(outcome.blobs_removed, 0);
assert!(!dir.path().join("blobs").exists());
}
#[test]
fn max_messages_leaves_the_newest_n_with_their_ids_intact() {
let dir = tempfile::tempdir().unwrap();
let path = write_record(dir.path(), Some("capsule"), 10);
let before: Vec<String> = std::fs::read_to_string(&path)
.unwrap()
.lines()
.filter_map(parse_message_line)
.map(|m| m["id"].as_str().unwrap().to_string())
.collect();
let outcome = truncate_record(&path, 3, "capsule").unwrap();
assert_eq!(outcome.dropped, 7);
assert_eq!(outcome.kept, 3);
assert_eq!(outcome.last_dropped_id, before[6]);
assert_eq!(outcome.oldest_surviving_id, before[7]);
let raw = std::fs::read_to_string(&path).unwrap();
let lines: Vec<&str> = raw.lines().collect();
assert_eq!(
lines.len(),
4,
"a header plus exactly three messages: {lines:?}"
);
for line in &lines {
serde_json::from_str::<serde_json::Value>(line).expect("every line is JSON");
}
let header = parse_header(lines[0]).expect("the first line is the header");
assert_eq!(header.capsule, "capsule");
let marker = header.truncated.expect("the header records the drop");
assert_eq!(marker.dropped, 7);
assert_eq!(marker.last_dropped_id, before[6]);
assert_eq!(marker.oldest_surviving_id, before[7]);
let after: Vec<String> = lines[1..]
.iter()
.map(|line| {
parse_message_line(line).expect("a message with a string role")["id"]
.as_str()
.unwrap()
.to_string()
})
.collect();
assert_eq!(after, before[7..], "surviving ids are the bytes they were");
}
#[test]
fn a_header_line_is_not_counted_as_a_message() {
let dir = tempfile::tempdir().unwrap();
let headerless = write_record(&dir.path().join("a"), None, 4);
let owned = write_record(&dir.path().join("b"), Some("capsule"), 4);
assert_eq!(count_messages(&headerless), 4);
assert_eq!(count_messages(&owned), 4);
}
#[test]
fn a_second_truncation_accumulates_the_dropped_count() {
let dir = tempfile::tempdir().unwrap();
let path = write_record(dir.path(), Some("capsule"), 10);
truncate_record(&path, 6, "capsule").unwrap();
truncate_record(&path, 2, "capsule").unwrap();
let marker = read_header(&path).unwrap().truncated.unwrap();
assert_eq!(marker.dropped, 8);
assert_eq!(count_messages(&path), 2);
}
#[test]
fn truncation_rewrite_keeps_the_record_owner_only() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
let path = write_record(dir.path(), Some("capsule"), 5);
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
truncate_record(&path, 2, "capsule").unwrap();
assert_eq!(count_messages(&path), 2);
assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
}
#[test]
fn a_record_inside_the_limit_is_not_rewritten() {
let dir = tempfile::tempdir().unwrap();
let path = write_record(dir.path(), Some("capsule"), 3);
let before = std::fs::read(&path).unwrap();
let outcome = truncate_record(&path, 3, "capsule").unwrap();
assert_eq!(outcome.dropped, 0);
assert_eq!(std::fs::read(&path).unwrap(), before);
}
#[test]
fn a_staged_truncation_leaves_the_original_intact_until_the_rename() {
let dir = tempfile::tempdir().unwrap();
let path = write_record(dir.path(), Some("capsule"), 10);
let before = std::fs::read(&path).unwrap();
let (staged, outcome) = stage_truncation(&path, 3, "capsule", NOW).unwrap().unwrap();
assert_eq!(outcome.dropped, 7);
let temp = staged.temp_path().to_path_buf();
assert_eq!(
temp.parent(),
path.parent(),
"the temp file is in the record's own directory, so the rename is atomic"
);
assert_ne!(temp.file_name().unwrap(), RECORD_FILE_NAME);
assert_eq!(
std::fs::read(&path).unwrap(),
before,
"the original is byte-identical before the rename"
);
assert_eq!(count_messages(&path), 10);
for line in std::fs::read_to_string(&path).unwrap().lines() {
serde_json::from_str::<serde_json::Value>(line).expect("the original parses whole");
}
drop(staged);
assert!(!temp.exists());
assert_eq!(std::fs::read(&path).unwrap(), before);
assert_eq!(count_messages(&path), 10);
}
#[test]
fn record_pruning_skips_another_capsules_records_and_every_unowned_one() {
let root = tempfile::tempdir().unwrap();
write_record(&root.path().join("mine"), Some("mine"), 3);
write_record(&root.path().join("theirs"), Some("theirs"), 3);
write_record(&root.path().join("unowned"), None, 3);
write_record(&root.path().join("current"), Some("mine"), 3);
let far_future = now_ms() + 3_600_000;
let pruned = prune_records(
root.path(),
"mine",
Some("current"),
&ContextRetainConfig {
max_messages: None,
max_age_secs: Some(1),
},
far_future,
);
assert_eq!(pruned.len(), 1, "{pruned:?}");
assert_eq!(pruned[0].context_id, "mine");
assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
assert_eq!(names(root.path()), vec!["current", "theirs", "unowned"]);
}
#[test]
fn a_record_written_inside_the_window_is_kept() {
let root = tempfile::tempdir().unwrap();
write_record(&root.path().join("fresh"), Some("mine"), 3);
let pruned = prune_records(
root.path(),
"mine",
None,
&ContextRetainConfig {
max_messages: None,
max_age_secs: Some(90 * 86_400),
},
now_ms(),
);
assert!(pruned.is_empty());
assert_eq!(names(root.path()), vec!["fresh"]);
}
#[test]
fn max_messages_alone_removes_no_record() {
let root = tempfile::tempdir().unwrap();
write_record(&root.path().join("ctx"), Some("mine"), 100);
assert!(prune_records(
root.path(),
"mine",
None,
&ContextRetainConfig {
max_messages: Some(1),
max_age_secs: None
},
now_ms() + 3_600_000,
)
.is_empty());
assert_eq!(names(root.path()), vec!["ctx"]);
}
fn frm(ms: u64, tail: u64) -> FormationId {
FormationId::parse(&format!("frm_{ms:012x}{tail:020x}")).unwrap()
}
fn retain(max_sessions: Option<u32>, max_age_secs: Option<u64>) -> TraceRetainConfig {
TraceRetainConfig {
max_sessions,
max_age_secs,
}
}
fn make_formation(root: &Path, id: &FormationId, marker: Option<&str>) -> PathBuf {
let dir = root.join(id.as_str());
let session = dir.join("worker/.murmur").join(ses(NOW, 1));
std::fs::create_dir_all(&session).unwrap();
std::fs::write(session.join("trace.jsonl"), "{}\n").unwrap();
if let Some(marker) = marker {
std::fs::write(
dir.join(crate::formation_launch::FORMATION_MARKER_FILE),
marker,
)
.unwrap();
}
dir
}
fn marker_naming(project: &Path) -> String {
serde_json::to_string(&FormationMarker {
project_dir: project.to_path_buf(),
launcher_pid: 1,
launcher_start: String::new(),
})
.unwrap()
}
fn make_owned(root: &Path, project: &Path, ids: &[&FormationId]) {
for id in ids {
make_formation(root, id, Some(&marker_naming(project)));
}
}
fn removed(pruned: &[PrunedFormation]) -> Vec<(FormationId, &'static str)> {
pruned
.iter()
.map(|p| (p.formation_id.clone(), p.reason))
.collect()
}
const DAY_MS: u64 = 86_400_000;
#[test]
fn prune_formations_never_removes_the_current_or_a_later_formation() {
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let old = frm(NOW - 3 * DAY_MS, 1);
let current = frm(NOW - 2 * DAY_MS, 2);
let later = frm(NOW - 2 * DAY_MS + 1000, 3);
make_owned(root.path(), project.path(), &[&old, ¤t, &later]);
let pruned = prune_formations(
root.path(),
¤t,
project.path(),
&retain(Some(1), Some(1)),
NOW,
|_, _| false,
);
assert_eq!(
removed(&pruned),
[(old.clone(), crate::trace::RETENTION_REASON_MAX_AGE)]
);
assert_eq!(pruned[0].path, root.path().join(old.as_str()));
assert_eq!(
names(root.path()),
vec![current.as_str().to_string(), later.as_str().to_string()]
);
}
#[test]
fn prune_formations_keeps_the_newest_n_with_the_current_counted() {
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let all: Vec<FormationId> = (1..=5).map(|n| frm(NOW - (5 - n) * 1000, n)).collect();
make_owned(root.path(), project.path(), &all.iter().collect::<Vec<_>>());
let pruned = prune_formations(
root.path(),
&all[3],
project.path(),
&retain(Some(3), None),
NOW,
|_, _| false,
);
let reason = crate::trace::RETENTION_REASON_MAX_SESSIONS;
assert_eq!(
removed(&pruned),
[(all[1].clone(), reason), (all[0].clone(), reason)]
);
assert_eq!(
names(root.path()),
all[2..]
.iter()
.map(|id| id.as_str().to_string())
.collect::<Vec<_>>()
);
}
#[test]
fn prune_formations_reads_age_from_the_id() {
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let old_a = frm(NOW - 7_200_000, 1);
let old_b = frm(NOW - 7_100_000, 2);
let recent = frm(NOW - 60_000, 3);
let current = frm(NOW, 4);
make_owned(
root.path(),
project.path(),
&[&old_a, &old_b, &recent, ¤t],
);
let pruned = prune_formations(
root.path(),
¤t,
project.path(),
&retain(None, Some(900)),
NOW,
|_, _| false,
);
let reason = crate::trace::RETENTION_REASON_MAX_AGE;
assert_eq!(removed(&pruned), [(old_b, reason), (old_a, reason)]);
assert_eq!(
names(root.path()),
vec![recent.as_str().to_string(), current.as_str().to_string()]
);
}
#[test]
fn prune_formations_attributes_reasons_age_first() {
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let old = frm(NOW - 7_200_000, 1);
let recent = frm(NOW - 60_000, 2);
let current = frm(NOW, 3);
make_owned(root.path(), project.path(), &[&old, &recent, ¤t]);
let pruned = prune_formations(
root.path(),
¤t,
project.path(),
&retain(Some(1), Some(900)),
NOW,
|_, _| false,
);
assert_eq!(
removed(&pruned),
[
(recent, crate::trace::RETENTION_REASON_MAX_SESSIONS),
(old, crate::trace::RETENTION_REASON_MAX_AGE),
]
);
}
#[test]
fn prune_formations_skips_a_live_directory_and_keeps_its_rank() {
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let older = frm(NOW - 3000, 1);
let live = frm(NOW - 2000, 2);
let current = frm(NOW, 3);
make_owned(root.path(), project.path(), &[&older, &live, ¤t]);
let asked = std::cell::RefCell::new(Vec::new());
let pruned = prune_formations(
root.path(),
¤t,
project.path(),
&retain(Some(1), None),
NOW,
|id, marker| {
assert_eq!(marker.launcher_pid, 1);
asked.borrow_mut().push(id.clone());
*id == live
},
);
assert_eq!(
removed(&pruned),
[(older.clone(), crate::trace::RETENTION_REASON_MAX_SESSIONS)]
);
assert_eq!(*asked.borrow(), [live.clone(), older]);
assert_eq!(
names(root.path()),
vec![live.as_str().to_string(), current.as_str().to_string()]
);
let kept = frm(NOW - 2500, 4);
make_owned(root.path(), project.path(), &[&kept]);
let pruned = prune_formations(
root.path(),
¤t,
project.path(),
&retain(Some(3), None),
NOW,
|id, _| *id == live,
);
assert!(pruned.is_empty(), "{pruned:?}");
let pruned = prune_formations(
root.path(),
¤t,
project.path(),
&retain(Some(2), None),
NOW,
|id, _| *id == live,
);
assert_eq!(
removed(&pruned),
[(kept, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
);
}
#[test]
fn prune_formations_never_ranks_or_removes_unowned_or_foreign_directories() {
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let other = tempfile::tempdir().unwrap();
let owned_old = frm(NOW - 3 * DAY_MS, 1);
let owned_kept = frm(NOW - 2 * DAY_MS, 2);
let unmarked = frm(NOW - 2 * DAY_MS + 1, 3);
let garbled = frm(NOW - 2 * DAY_MS + 2, 4);
let foreign = frm(NOW - 2 * DAY_MS + 3, 5);
let current = frm(NOW, 6);
make_owned(
root.path(),
project.path(),
&[&owned_old, &owned_kept, ¤t],
);
make_formation(root.path(), &unmarked, None);
make_formation(root.path(), &garbled, Some("not json {"));
make_formation(root.path(), &foreign, Some(&marker_naming(other.path())));
let pruned = prune_formations(
root.path(),
¤t,
project.path(),
&retain(Some(2), None),
NOW,
|_, _| false,
);
assert_eq!(
removed(&pruned),
[(owned_old, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
);
let pruned = prune_formations(
root.path(),
¤t,
project.path(),
&retain(Some(1), Some(DAY_MS / 1000)),
NOW,
|_, _| false,
);
assert_eq!(
removed(&pruned),
[(owned_kept, crate::trace::RETENTION_REASON_MAX_AGE)]
);
for id in [&unmarked, &garbled, &foreign, ¤t] {
assert!(root.path().join(id.as_str()).is_dir(), "{id}");
}
}
#[test]
fn prune_formations_ignores_a_symlink_a_regular_file_and_a_name_that_is_not_an_id() {
let root = tempfile::tempdir().unwrap();
let elsewhere = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let marker = marker_naming(project.path());
let target = make_formation(elsewhere.path(), &frm(NOW - 3 * DAY_MS, 1), Some(&marker));
let linked = frm(NOW - 3 * DAY_MS, 2);
std::os::unix::fs::symlink(&target, root.path().join(linked.as_str())).unwrap();
std::fs::write(root.path().join(frm(NOW - 3 * DAY_MS, 3).as_str()), &marker).unwrap();
for name in ["not-a-formation", "frm_ABC", "frm_0123"] {
std::fs::create_dir(root.path().join(name)).unwrap();
std::fs::write(
root.path()
.join(name)
.join(crate::formation_launch::FORMATION_MARKER_FILE),
&marker,
)
.unwrap();
}
let before = names(root.path());
let pruned = prune_formations(
root.path(),
&frm(NOW, 9),
project.path(),
&retain(Some(1), Some(1)),
NOW,
|_, _| false,
);
assert!(pruned.is_empty(), "{pruned:?}");
assert_eq!(names(root.path()), before);
assert!(target.join("worker/.murmur").is_dir());
}
#[test]
fn prune_formations_with_neither_key_or_no_directory_removes_nothing() {
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let old = frm(NOW - 3 * DAY_MS, 1);
make_owned(root.path(), project.path(), &[&old]);
let pruned = prune_formations(
root.path(),
&frm(NOW, 2),
project.path(),
&retain(None, None),
NOW,
|_, _| panic!("nothing is a candidate without a policy"),
);
assert!(pruned.is_empty());
assert!(root.path().join(old.as_str()).is_dir());
assert!(prune_formations(
&root.path().join("missing"),
&frm(NOW, 2),
project.path(),
&retain(Some(1), Some(1)),
NOW,
|_, _| false,
)
.is_empty());
}
#[test]
fn prune_formations_compares_project_directories_canonically() {
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let links = tempfile::tempdir().unwrap();
let link = links.path().join("project");
std::os::unix::fs::symlink(project.path(), &link).unwrap();
let old = frm(NOW - 3 * DAY_MS, 1);
make_owned(
root.path(),
&std::fs::canonicalize(project.path()).unwrap(),
&[&old],
);
let pruned = prune_formations(
root.path(),
&frm(NOW, 2),
&link,
&retain(None, Some(1)),
NOW,
|_, _| false,
);
assert_eq!(
removed(&pruned),
[(old, crate::trace::RETENTION_REASON_MAX_AGE)]
);
}
#[test]
fn prune_formations_leaves_a_directory_it_could_not_remove_out_of_the_result() {
use std::os::unix::fs::PermissionsExt;
let root = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let stuck = frm(NOW - 3 * DAY_MS, 1);
let gone = frm(NOW - 3 * DAY_MS + 1, 2);
make_owned(root.path(), project.path(), &[&stuck, &gone]);
let stuck_dir = root.path().join(stuck.as_str());
std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o500)).unwrap();
if std::fs::write(stuck_dir.join("probe"), b"").is_ok() {
std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
return;
}
let pruned = prune_formations(
root.path(),
&frm(NOW, 3),
project.path(),
&retain(None, Some(1)),
NOW,
|_, _| false,
);
std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
assert_eq!(
removed(&pruned),
[(gone, crate::trace::RETENTION_REASON_MAX_AGE)]
);
assert!(stuck_dir.is_dir());
}
fn harness_entry_json(kind: &str, session_id: &str) -> String {
json!({"type": kind, "session_id": session_id, "harness": "h", "driver": "d",
"created_ms": 1, "updated_ms": 2})
.to_string()
}
#[test]
fn the_context_finder_sees_either_file_in_every_store() {
let root = tempfile::tempdir().unwrap();
let dir = |store: &str, context: &str| {
let path = root.path().join(store).join(context);
std::fs::create_dir_all(&path).unwrap();
path
};
std::fs::write(dir("b", "ctx").join(RECORD_FILE_NAME), "").unwrap();
std::fs::write(
dir("a", "ctx").join(HARNESS_SESSION_FILE_NAME),
harness_entry_json(crate::harness_session::HARNESS_SESSION_TYPE, "s"),
)
.unwrap();
dir("c", "ctx");
std::fs::write(dir(".hidden", "ctx").join(RECORD_FILE_NAME), "").unwrap();
std::fs::write(dir("a", ".ctx").join(RECORD_FILE_NAME), "").unwrap();
let found = find_context_dirs_in(root.path(), "ctx");
let stores: Vec<&str> = found.iter().map(|dir| dir.record.as_str()).collect();
assert_eq!(stores, ["a", "b"]);
assert!(found.iter().all(|dir| dir.context_id == "ctx"));
assert_eq!(found[1].path, root.path().join("b").join("ctx"));
assert!(find_context_dirs_in(root.path(), ".ctx").is_empty());
assert!(find_context_dirs_in(root.path(), "other").is_empty());
assert!(find_context_dirs_in(&root.path().join("absent"), "ctx").is_empty());
}
#[test]
fn the_context_reader_reads_as_the_runtime_reads() {
let dir = tempfile::tempdir().unwrap();
let header = RecordHeader {
kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
capsule: "owner".to_string(),
created_ms: 7,
truncated: None,
};
let first = json!({"role": "user", "content": "hi", "id": "msg_1", "kept": {"x": 1}});
let second = json!({"role": "assistant", "content": [{"type": "text", "text": "yo"}]});
std::fs::write(
dir.path().join(RECORD_FILE_NAME),
format!(
"{}{first}\n\n{second}\n{}\n{{\"role\": \"us",
header_line(&header).unwrap(),
header_line(&header).unwrap().trim_end()
),
)
.unwrap();
let contents = read_context_dir(dir.path()).unwrap();
assert_eq!(contents.header, Some(header.clone()));
assert_eq!(contents.messages, vec![first.clone(), second]);
assert_eq!(contents.harness_session, None);
std::fs::write(
dir.path().join(RECORD_FILE_NAME),
format!("{first}\n{}", header_line(&header).unwrap()),
)
.unwrap();
let contents = read_context_dir(dir.path()).unwrap();
assert_eq!(contents.header, None, "a header below line 1 is no header");
assert_eq!(contents.messages, vec![first]);
let empty = tempfile::tempdir().unwrap();
let contents = read_context_dir(empty.path()).unwrap();
assert_eq!(contents.header, None);
assert!(contents.messages.is_empty());
std::fs::create_dir(empty.path().join(RECORD_FILE_NAME)).unwrap();
assert!(
read_context_dir(empty.path()).is_err(),
"an unreadable record"
);
}
#[test]
fn the_context_reader_reads_no_session_where_the_runtime_does() {
let dir = tempfile::tempdir().unwrap();
let file = dir.path().join(HARNESS_SESSION_FILE_NAME);
assert_eq!(read_context_dir(dir.path()).unwrap().harness_session, None);
let session_type = crate::harness_session::HARNESS_SESSION_TYPE;
for (body, expected) in [
("{not json".to_string(), false),
(harness_entry_json("another-tool", "s"), false),
(harness_entry_json(session_type, ""), false),
(harness_entry_json(session_type, "s"), true),
] {
std::fs::write(&file, &body).unwrap();
let session = read_context_dir(dir.path()).unwrap().harness_session;
assert_eq!(session.is_some(), expected, "{body}");
if let Some(session) = session {
assert_eq!(
(session.session_id.as_str(), session.harness.as_str()),
("s", "h")
);
assert_eq!((session.created_ms, session.updated_ms), (1, 2));
}
}
}
}