1use std::path::{Path, PathBuf};
35
36use crate::conversation::{
37 header_line, message_id_timestamp_ms, parse_header, parse_message_line, read_header,
38 record_messages, CONVERSATION_ROOT_DIR, RECORD_FILE_NAME,
39};
40
41use crate::formation::FormationId;
42use crate::formation_launch::FormationMarker;
43use crate::harness_session::{read_entry, HARNESS_SESSION_FILE_NAME};
44
45pub use crate::conversation::{RecordHeader, TruncationMarker};
46pub use crate::harness_session::HarnessSessionEntry;
47
48#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct PrunedSession {
51 pub name: String,
53 pub reason: &'static str,
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
60pub struct PrunedFormation {
61 pub formation_id: FormationId,
63 pub path: PathBuf,
65 pub reason: &'static str,
68}
69
70#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct PrunedRecord {
73 pub context_id: String,
75 pub path: PathBuf,
77 pub reason: &'static str,
79}
80
81#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct TruncationOutcome {
84 pub dropped: u64,
86 pub kept: u64,
88 pub oldest_surviving_id: String,
90 pub last_dropped_id: String,
92 pub blobs_removed: u64,
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
98pub enum MessageStatus {
99 Present { position: u64, total: u64 },
101 Truncated {
105 dropped: u64,
106 oldest_surviving_id: String,
107 },
108 Unknown,
110}
111
112#[derive(Debug, Clone, PartialEq, Eq)]
114pub struct MessageLocation {
115 pub record: String,
116 pub context_id: String,
117 pub path: PathBuf,
118 pub status: MessageStatus,
119}
120
121#[derive(Debug, Clone, PartialEq, Eq)]
123pub struct RecordSummary {
124 pub record: String,
126 pub context_id: String,
128 pub path: PathBuf,
129 pub messages: u64,
131 pub bytes: u64,
133 pub last_touched_ms: Option<u64>,
136 pub capsule: Option<String>,
139 pub truncation: Option<TruncationMarker>,
141}
142
143#[derive(Debug, Clone, PartialEq, Eq)]
145pub struct RemovedRecord {
146 pub record: String,
147 pub context_id: String,
148 pub path: PathBuf,
150 pub messages: u64,
152}
153
154#[derive(Debug, Clone, PartialEq, Eq)]
157pub struct ContextDir {
158 pub record: String,
160 pub context_id: String,
162 pub path: PathBuf,
164}
165
166#[derive(Debug, Clone, PartialEq)]
168pub struct ContextContents {
169 pub header: Option<RecordHeader>,
171 pub messages: Vec<serde_json::Value>,
174 pub harness_session: Option<HarnessSessionEntry>,
176}
177
178pub fn session_id_timestamp_ms(name: &str) -> Option<u64> {
186 uuid_v7_timestamp_ms(name, "ses_")
187}
188
189pub(crate) fn uuid_v7_timestamp_ms(id: &str, prefix: &str) -> Option<u64> {
192 let hex = id.strip_prefix(prefix)?;
193 u64::from_str_radix(hex.get(..12)?, 16).ok()
194}
195
196pub(crate) fn created_workdirs(
200 workdir_flag: Option<&std::path::Path>,
201 manifest_dir: &std::path::Path,
202) -> Option<std::path::PathBuf> {
203 workdir_flag.is_none().then(|| manifest_dir.join("workdir"))
204}
205
206pub fn prune_sessions(
229 records_group: &crate::workdir_fs::WorkdirRoot,
230 created_workdirs: Option<&crate::workdir_fs::WorkdirRoot>,
231 current_session_id: &str,
232 policy: &murmur_artifact::TraceRetainConfig,
233 now_ms: u64,
234) -> Vec<PrunedSession> {
235 if policy.max_sessions.is_none() && policy.max_age_secs.is_none() {
236 return Vec::new();
237 }
238
239 let mut entries: Vec<String> = match records_group.list_dir("") {
240 Ok(Some(listed)) => listed
241 .into_iter()
242 .filter(|(_, kind)| *kind == crate::workdir_fs::EntryKind::Directory)
243 .map(|(name, _)| name)
244 .filter(|name| session_id_timestamp_ms(name).is_some())
245 .collect(),
246 Ok(None) | Err(_) => return Vec::new(),
247 };
248 entries.sort_unstable_by(|a, b| b.cmp(a));
250
251 let age_floor_ms = policy
252 .max_age_secs
253 .map(|secs| now_ms.saturating_sub(secs.saturating_mul(1000)));
254
255 let mut pruned = Vec::new();
256 for (rank, name) in entries.iter().enumerate() {
257 if name.as_str() >= current_session_id {
260 continue;
261 }
262 let over_age = age_floor_ms.is_some_and(|floor| {
263 session_id_timestamp_ms(name).is_some_and(|minted| minted < floor)
264 });
265 let over_count = policy.max_sessions.is_some_and(|max| rank >= max as usize);
266 if !over_age && !over_count {
267 continue;
268 }
269 let reason = if over_age {
270 crate::trace::RETENTION_REASON_MAX_AGE
271 } else {
272 crate::trace::RETENTION_REASON_MAX_SESSIONS
273 };
274 if records_group.remove_dir_all(name).is_ok() {
275 if let Some(workdirs) = created_workdirs {
276 let _ = workdirs.remove_dir_all(name);
277 }
278 pruned.push(PrunedSession {
279 name: name.clone(),
280 reason,
281 });
282 }
283 }
284 pruned
285}
286
287pub fn prune_formations(
313 formations_dir: &Path,
314 current: &FormationId,
315 project_dir: &Path,
316 policy: &murmur_artifact::TraceRetainConfig,
317 now_ms: u64,
318 is_live: impl Fn(&FormationId, &FormationMarker) -> bool,
319) -> Vec<PrunedFormation> {
320 if policy.max_sessions.is_none() && policy.max_age_secs.is_none() {
321 return Vec::new();
322 }
323 let Ok(entries) = std::fs::read_dir(formations_dir) else {
324 return Vec::new();
325 };
326 let project_dir = crate::formation_launch::canonical_project_dir(project_dir);
327 let mut owned: Vec<(FormationId, FormationMarker)> = entries
328 .filter_map(Result::ok)
329 .filter_map(|entry| {
330 let id = FormationId::parse(entry.file_name().to_str()?).ok()?;
331 let path = entry.path();
332 if !std::fs::symlink_metadata(&path).ok()?.is_dir() {
333 return None;
334 }
335 let marker = FormationMarker::read(&path)?;
336 (crate::formation_launch::canonical_project_dir(&marker.project_dir) == project_dir)
337 .then_some((id, marker))
338 })
339 .collect();
340 owned.sort_unstable_by(|(a, _), (b, _)| b.as_str().cmp(a.as_str()));
342
343 let age_floor_ms = policy
344 .max_age_secs
345 .map(|secs| now_ms.saturating_sub(secs.saturating_mul(1000)));
346
347 let mut pruned = Vec::new();
348 for (rank, (id, marker)) in owned.iter().enumerate() {
349 if id.as_str() >= current.as_str() {
350 continue;
351 }
352 let over_age = age_floor_ms.is_some_and(|floor| id.minted_at_ms() < floor);
353 let over_count = policy.max_sessions.is_some_and(|max| rank >= max as usize);
354 if !over_age && !over_count {
355 continue;
356 }
357 if is_live(id, marker) {
358 continue;
359 }
360 let reason = if over_age {
361 crate::trace::RETENTION_REASON_MAX_AGE
362 } else {
363 crate::trace::RETENTION_REASON_MAX_SESSIONS
364 };
365 let path = formations_dir.join(id.as_str());
366 let records = crate::records_dir::records_group(&path).ok();
370 if std::fs::remove_dir_all(&path).is_ok() {
371 if let Some(records) = records {
372 let _ = std::fs::remove_dir_all(records);
373 }
374 pruned.push(PrunedFormation {
375 formation_id: id.clone(),
376 path,
377 reason,
378 });
379 }
380 }
381 pruned
382}
383
384pub fn prune_records(
403 record_root: &Path,
404 capsule_name: &str,
405 current_context_id: Option<&str>,
406 policy: &murmur_artifact::ContextRetainConfig,
407 now_ms: u64,
408) -> Vec<PrunedRecord> {
409 let Some(max_age_secs) = policy.max_age_secs else {
410 return Vec::new();
411 };
412 let floor_ms = now_ms.saturating_sub(max_age_secs.saturating_mul(1000));
413
414 let mut pruned = Vec::new();
415 for (context_id, dir) in context_dirs(record_root) {
416 if Some(context_id.as_str()) == current_context_id {
417 continue;
418 }
419 let file = dir.join(RECORD_FILE_NAME);
420 match read_header(&file) {
421 Some(header) if header.capsule == capsule_name => {}
422 _ => continue,
423 }
424 let Some(touched_ms) = last_touched_ms(&file) else {
425 continue;
426 };
427 if touched_ms >= floor_ms {
428 continue;
429 }
430 if std::fs::remove_dir_all(&dir).is_ok() {
431 pruned.push(PrunedRecord {
432 context_id,
433 path: dir,
434 reason: crate::trace::RETENTION_REASON_MAX_AGE,
435 });
436 }
437 }
438 pruned
439}
440
441pub fn truncate_record(path: &Path, keep: u32, capsule: &str) -> Result<TruncationOutcome, String> {
459 match stage_truncation(path, keep, capsule, now_ms())? {
460 None => Ok(TruncationOutcome {
461 dropped: 0,
462 kept: count_messages(path),
463 oldest_surviving_id: String::new(),
464 last_dropped_id: String::new(),
465 blobs_removed: 0,
466 }),
467 Some((staged, mut outcome)) => {
468 staged.commit()?;
469 outcome.blobs_removed = sweep_orphan_blobs(path);
470 Ok(outcome)
471 }
472 }
473}
474
475fn sweep_orphan_blobs(path: &Path) -> u64 {
479 let Some(blobs) = path
480 .parent()
481 .map(|dir| dir.join(crate::trace_blobs::BLOB_DIR_NAME))
482 else {
483 return 0;
484 };
485 let Ok(entries) = std::fs::read_dir(&blobs) else {
486 return 0;
487 };
488 let Ok(raw) = std::fs::read_to_string(path) else {
489 return 0;
490 };
491 let referenced: std::collections::HashSet<String> = raw
492 .lines()
493 .filter_map(parse_message_line)
494 .flat_map(|message| {
495 crate::images::referenced_blobs(&message)
496 .map(str::to_string)
497 .collect::<Vec<_>>()
498 })
499 .collect();
500 let mut removed = 0;
501 for entry in entries.flatten() {
502 let name = entry.file_name();
503 let Some(name) = name.to_str() else {
504 continue;
505 };
506 if !crate::images::is_sha256_hex(name) || referenced.contains(name) {
507 continue;
508 }
509 if std::fs::remove_file(entry.path()).is_ok() {
510 removed += 1;
511 }
512 }
513 removed
514}
515
516pub(crate) fn stage_truncation(
522 path: &Path,
523 keep: u32,
524 capsule: &str,
525 at_ms: u64,
526) -> Result<Option<(StagedRewrite, TruncationOutcome)>, String> {
527 let raw = match std::fs::read_to_string(path) {
528 Ok(raw) => raw,
529 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
530 Err(err) => return Err(format!("{}: {err}", path.display())),
531 };
532
533 let existing = read_header(path);
534 let lines: Vec<&str> = raw.lines().collect();
535 let message_indices: Vec<usize> = lines
538 .iter()
539 .enumerate()
540 .filter(|(_, line)| parse_message_line(line).is_some())
541 .map(|(index, _)| index)
542 .collect();
543
544 let total = message_indices.len();
545 let keep = keep as usize;
546 if total <= keep {
547 return Ok(None);
548 }
549 let dropped = total - keep;
550 let first_kept_index = message_indices[dropped];
551
552 let id_of = |index: usize| -> String {
553 parse_message_line(lines[index])
554 .as_ref()
555 .and_then(|message| crate::agent::message_id(message).map(str::to_string))
556 .unwrap_or_default()
557 };
558 let last_dropped_id = id_of(message_indices[dropped - 1]);
559 let oldest_surviving_id = message_indices
560 .get(dropped)
561 .map(|i| id_of(*i))
562 .unwrap_or_default();
563
564 let header = RecordHeader {
565 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
566 capsule: existing
567 .as_ref()
568 .map(|h| h.capsule.clone())
569 .unwrap_or_else(|| capsule.to_string()),
570 created_ms: existing.as_ref().map(|h| h.created_ms).unwrap_or(at_ms),
571 truncated: Some(TruncationMarker {
572 dropped: existing
575 .as_ref()
576 .and_then(|h| h.truncated.as_ref())
577 .map(|marker| marker.dropped)
578 .unwrap_or(0)
579 + dropped as u64,
580 oldest_surviving_id: oldest_surviving_id.clone(),
581 last_dropped_id: last_dropped_id.clone(),
582 at_ms,
583 }),
584 };
585
586 let mut contents = header_line(&header)?;
587 for line in &lines[first_kept_index..] {
588 contents.push_str(line);
589 contents.push('\n');
590 }
591
592 let staged = StagedRewrite::stage_with_mode(
593 path,
594 contents.as_bytes(),
595 Some(crate::conversation::RECORD_FILE_MODE),
596 )?;
597 Ok(Some((
598 staged,
599 TruncationOutcome {
600 dropped: dropped as u64,
601 kept: keep as u64,
602 oldest_surviving_id,
603 last_dropped_id,
604 blobs_removed: 0,
605 },
606 )))
607}
608
609pub(crate) struct StagedRewrite {
615 target: PathBuf,
616 temp: PathBuf,
617 committed: bool,
618}
619
620impl StagedRewrite {
621 pub(crate) fn stage_with_mode(
624 target: &Path,
625 contents: &[u8],
626 mode: Option<u32>,
627 ) -> Result<Self, String> {
628 use std::io::Write;
629 use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
630
631 let dir = target
632 .parent()
633 .ok_or_else(|| format!("{} has no parent directory", target.display()))?;
634 let name = target
635 .file_name()
636 .map(|name| name.to_string_lossy().into_owned())
637 .unwrap_or_else(|| "record".to_string());
638 let temp = dir.join(format!(
639 ".{name}.{}.{}.tmp",
640 std::process::id(),
641 uuid::Uuid::now_v7().simple()
642 ));
643 let mut options = std::fs::OpenOptions::new();
644 options.write(true).create(true).truncate(true);
645 if let Some(mode) = mode {
646 options.mode(mode);
647 }
648 let mut file = options
649 .open(&temp)
650 .map_err(|err| format!("{}: {err}", temp.display()))?;
651 file.write_all(contents)
654 .and_then(|()| file.sync_all())
655 .map_err(|err| format!("{}: {err}", temp.display()))?;
656 if let Some(mode) = mode {
660 std::fs::set_permissions(&temp, std::fs::Permissions::from_mode(mode))
661 .map_err(|err| format!("{}: {err}", temp.display()))?;
662 }
663 Ok(Self {
664 target: target.to_path_buf(),
665 temp,
666 committed: false,
667 })
668 }
669
670 #[cfg(test)]
671 pub(crate) fn temp_path(&self) -> &Path {
672 &self.temp
673 }
674
675 pub(crate) fn commit(mut self) -> Result<(), String> {
676 std::fs::rename(&self.temp, &self.target)
677 .map_err(|err| format!("{}: {err}", self.target.display()))?;
678 self.committed = true;
679 Ok(())
680 }
681}
682
683impl Drop for StagedRewrite {
684 fn drop(&mut self) {
685 if !self.committed {
686 let _ = std::fs::remove_file(&self.temp);
687 }
688 }
689}
690
691pub fn list_records() -> Result<Vec<RecordSummary>, String> {
699 let root = conversations_root()?;
700 let mut summaries = Vec::new();
701 for record in child_dirs(&root) {
702 let store = root.join(&record);
703 for (context_id, dir) in context_dirs(&store) {
704 let path = dir.join(RECORD_FILE_NAME);
705 if !path.exists() {
706 continue;
707 }
708 let header = read_header(&path);
709 summaries.push(RecordSummary {
710 record: record.clone(),
711 context_id,
712 messages: count_messages(&path),
713 bytes: std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0),
714 last_touched_ms: last_touched_ms(&path),
715 capsule: header.as_ref().map(|h| h.capsule.clone()),
716 truncation: header.and_then(|h| h.truncated),
717 path,
718 });
719 }
720 }
721 summaries.sort_by(|a, b| {
722 a.record
723 .cmp(&b.record)
724 .then_with(|| a.context_id.cmp(&b.context_id))
725 });
726 Ok(summaries)
727}
728
729pub fn remove_record(record: &str, context_id: &str) -> Result<RemovedRecord, String> {
734 let dir = conversations_root()?.join(record).join(context_id);
735 if !dir.is_dir() {
736 return Err(format!("{} is not a context directory", dir.display()));
737 }
738 let messages = count_messages(&dir.join(RECORD_FILE_NAME));
739 std::fs::remove_dir_all(&dir).map_err(|err| format!("{}: {err}", dir.display()))?;
740 Ok(RemovedRecord {
741 record: record.to_string(),
742 context_id: context_id.to_string(),
743 path: dir,
744 messages,
745 })
746}
747
748pub fn locate_message(message_id: &str) -> Result<Vec<MessageLocation>, String> {
760 let minted_ms = message_id_timestamp_ms(message_id);
761 let mut found = Vec::new();
762 for summary in list_records()? {
763 let raw = std::fs::read_to_string(&summary.path).unwrap_or_default();
764 let mut position = 0u64;
765 let mut at = None;
766 for line in raw.lines() {
767 let Some(message) = parse_message_line(line) else {
768 continue;
769 };
770 position += 1;
771 if crate::agent::message_id(&message) == Some(message_id) {
772 at = Some(position);
773 }
774 }
775 let status = match (at, summary.truncation.as_ref()) {
776 (Some(position), _) => MessageStatus::Present {
777 position,
778 total: position.max(summary.messages),
779 },
780 (None, Some(marker)) => {
781 let dropped_at = message_id_timestamp_ms(&marker.last_dropped_id);
782 match (minted_ms, dropped_at) {
783 (Some(minted), Some(last)) if minted <= last => MessageStatus::Truncated {
784 dropped: marker.dropped,
785 oldest_surviving_id: marker.oldest_surviving_id.clone(),
786 },
787 _ => MessageStatus::Unknown,
788 }
789 }
790 (None, None) => MessageStatus::Unknown,
791 };
792 if status != MessageStatus::Unknown {
793 found.push(MessageLocation {
794 record: summary.record,
795 context_id: summary.context_id,
796 path: summary.path,
797 status,
798 });
799 }
800 }
801 Ok(found)
802}
803
804pub fn find_context_dirs(context_id: &str) -> Result<Vec<ContextDir>, String> {
811 Ok(find_context_dirs_in(&conversations_root()?, context_id))
812}
813
814fn find_context_dirs_in(root: &Path, context_id: &str) -> Vec<ContextDir> {
815 let mut found = Vec::new();
816 for record in child_dirs(root) {
817 for (name, path) in context_dirs(&root.join(&record)) {
818 if name != context_id {
819 continue;
820 }
821 if path.join(RECORD_FILE_NAME).exists() || path.join(HARNESS_SESSION_FILE_NAME).exists()
822 {
823 found.push(ContextDir {
824 record: record.clone(),
825 context_id: name,
826 path,
827 });
828 }
829 }
830 }
831 found
832}
833
834pub fn read_context_dir(dir: &Path) -> Result<ContextContents, String> {
842 let record = dir.join(RECORD_FILE_NAME);
843 let raw = match std::fs::read_to_string(&record) {
844 Ok(raw) => raw,
845 Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
846 Err(err) => return Err(format!("{}: {err}", record.display())),
847 };
848 Ok(ContextContents {
849 header: raw.lines().next().and_then(parse_header),
850 messages: record_messages(&raw, |_| {}),
851 harness_session: read_entry(&dir.join(HARNESS_SESSION_FILE_NAME)),
852 })
853}
854
855pub fn conversations_root() -> Result<PathBuf, String> {
859 Ok(crate::state_store::murmur_home_dir()?.join(CONVERSATION_ROOT_DIR))
860}
861
862pub fn now_ms() -> u64 {
864 std::time::SystemTime::now()
865 .duration_since(std::time::UNIX_EPOCH)
866 .map(|elapsed| elapsed.as_millis() as u64)
867 .unwrap_or(0)
868}
869
870fn child_dirs(dir: &Path) -> Vec<String> {
872 let mut names: Vec<String> = match std::fs::read_dir(dir) {
873 Ok(entries) => entries
874 .filter_map(Result::ok)
875 .filter(|entry| entry.path().is_dir())
876 .map(|entry| entry.file_name().to_string_lossy().into_owned())
877 .filter(|name| !name.starts_with('.'))
878 .collect(),
879 Err(_) => Vec::new(),
880 };
881 names.sort();
882 names
883}
884
885fn context_dirs(record_root: &Path) -> Vec<(String, PathBuf)> {
887 child_dirs(record_root)
888 .into_iter()
889 .map(|name| {
890 let path = record_root.join(&name);
891 (name, path)
892 })
893 .collect()
894}
895
896fn count_messages(path: &Path) -> u64 {
899 std::fs::read_to_string(path)
900 .map(|raw| {
901 raw.lines()
902 .filter(|line| parse_message_line(line).is_some())
903 .count() as u64
904 })
905 .unwrap_or(0)
906}
907
908fn last_touched_ms(path: &Path) -> Option<u64> {
910 std::fs::metadata(path)
911 .and_then(|metadata| metadata.modified())
912 .ok()?
913 .duration_since(std::time::UNIX_EPOCH)
914 .ok()
915 .map(|elapsed| elapsed.as_millis() as u64)
916}
917
918#[cfg(test)]
919mod tests {
920 use super::*;
921 use murmur_artifact::{ContextRetainConfig, TraceRetainConfig};
922 use serde_json::json;
923
924 fn ses(ms: u64, tail: u64) -> String {
927 format!("ses_{ms:012x}{tail:020x}")
928 }
929
930 fn msg(ms: u64, tail: u64) -> String {
932 format!("msg_{ms:012x}{tail:020x}")
933 }
934
935 fn message_line(id: &str, text: &str) -> String {
936 json!({"role": "user", "content": [{"type": "text", "text": text}], "id": id}).to_string()
937 }
938
939 fn make_sessions(root: &Path, names: &[String]) {
940 for name in names {
941 let dir = root.join(name);
942 std::fs::create_dir_all(dir.join("blobs")).unwrap();
943 std::fs::write(dir.join("trace.jsonl"), "{}\n").unwrap();
944 std::fs::write(dir.join("blobs").join("deadbeef"), b"body").unwrap();
945 }
946 }
947
948 fn names(root: &Path) -> Vec<String> {
949 let mut found: Vec<String> = std::fs::read_dir(root)
950 .unwrap()
951 .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
952 .collect();
953 found.sort();
954 found
955 }
956
957 const NOW: u64 = 1_756_400_000_000;
958
959 #[test]
963 fn max_sessions_leaves_exactly_the_newest_n_including_the_current() {
964 let root = tempfile::tempdir().unwrap();
965 let all: Vec<String> = (1..=6).map(|n| ses(NOW - (6 - n) * 1000, n)).collect();
966 make_sessions(root.path(), &all);
967
968 let pruned = prune_sessions(
969 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
970 None,
971 &all[5],
972 &TraceRetainConfig {
973 max_sessions: Some(3),
974 max_age_secs: None,
975 },
976 NOW,
977 );
978
979 assert_eq!(
980 names(root.path()),
981 vec![all[3].clone(), all[4].clone(), all[5].clone()]
982 );
983 let mut removed: Vec<String> = pruned.iter().map(|p| p.name.clone()).collect();
984 removed.sort();
985 assert_eq!(
986 removed,
987 vec![all[0].clone(), all[1].clone(), all[2].clone()]
988 );
989 assert!(pruned
990 .iter()
991 .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_SESSIONS));
992 for name in &all[..3] {
993 assert!(!root.path().join(name).exists(), "{name} went whole");
994 }
995 }
996
997 #[test]
1001 fn max_age_reads_the_id_and_never_the_filesystem() {
1002 let root = tempfile::tempdir().unwrap();
1003 let old_a = ses(NOW - 7_200_000, 1);
1004 let old_b = ses(NOW - 7_100_000, 2);
1005 let recent = ses(NOW - 60_000, 3);
1006 let current = ses(NOW, 4);
1007 make_sessions(
1008 root.path(),
1009 &[
1010 old_a.clone(),
1011 old_b.clone(),
1012 recent.clone(),
1013 current.clone(),
1014 ],
1015 );
1016
1017 let pruned = prune_sessions(
1018 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1019 None,
1020 ¤t,
1021 &TraceRetainConfig {
1022 max_sessions: None,
1023 max_age_secs: Some(900),
1024 },
1025 NOW,
1026 );
1027
1028 assert_eq!(pruned.len(), 2, "{pruned:?}");
1029 assert!(pruned
1030 .iter()
1031 .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_AGE));
1032 assert_eq!(names(root.path()), vec![recent, current]);
1033 }
1034
1035 #[test]
1038 fn both_trace_keys_keep_only_sessions_inside_both() {
1039 let root = tempfile::tempdir().unwrap();
1040 let old = [ses(NOW - 7_200_000, 1), ses(NOW - 7_100_000, 2)];
1041 let recent = [
1042 ses(NOW - 300_000, 3),
1043 ses(NOW - 200_000, 4),
1044 ses(NOW - 100_000, 5),
1045 ];
1046 let current = ses(NOW, 6);
1047 let mut all = old.to_vec();
1048 all.extend(recent.iter().cloned());
1049 all.push(current.clone());
1050 make_sessions(root.path(), &all);
1051
1052 let pruned = prune_sessions(
1053 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1054 None,
1055 ¤t,
1056 &TraceRetainConfig {
1057 max_sessions: Some(2),
1058 max_age_secs: Some(900),
1059 },
1060 NOW,
1061 );
1062
1063 assert_eq!(names(root.path()), vec![recent[2].clone(), current.clone()]);
1064 assert_eq!(pruned.len(), 4);
1065 for session in &pruned {
1066 let expected = if old.contains(&session.name) {
1067 crate::trace::RETENTION_REASON_MAX_AGE
1068 } else {
1069 crate::trace::RETENTION_REASON_MAX_SESSIONS
1070 };
1071 assert_eq!(session.reason, expected, "{session:?}");
1072 }
1073 }
1074
1075 #[test]
1078 fn nothing_at_or_after_the_current_session_id_is_a_candidate() {
1079 let root = tempfile::tempdir().unwrap();
1080 let current = ses(NOW, 5);
1081 let sibling = ses(NOW + 1, 6);
1083 let older = ses(NOW - 1000, 4);
1084 make_sessions(
1085 root.path(),
1086 &[current.clone(), sibling.clone(), older.clone()],
1087 );
1088
1089 let pruned = prune_sessions(
1090 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1091 None,
1092 ¤t,
1093 &TraceRetainConfig {
1094 max_sessions: Some(1),
1095 max_age_secs: Some(1),
1096 },
1097 NOW,
1098 );
1099
1100 assert_eq!(pruned.len(), 1);
1101 assert_eq!(pruned[0].name, older);
1102 assert_eq!(names(root.path()), vec![current.clone(), sibling]);
1103 assert!(root.path().join(¤t).join("trace.jsonl").exists());
1104 }
1105
1106 #[test]
1110 fn a_pruned_session_takes_its_records_and_its_created_work_folder() {
1111 let records = tempfile::tempdir().unwrap();
1112 let workdirs = tempfile::tempdir().unwrap();
1113 let all: Vec<String> = (1..=3).map(|n| ses(NOW - (3 - n) * 1000, n)).collect();
1114 make_sessions(records.path(), &all);
1115 for name in &all {
1116 std::fs::create_dir_all(workdirs.path().join(name).join("out")).unwrap();
1117 }
1118 let workdir_only = ses(NOW - 9_000, 9);
1119 std::fs::create_dir_all(workdirs.path().join(&workdir_only)).unwrap();
1120
1121 let pruned = prune_sessions(
1122 &crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
1123 Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
1124 &all[2],
1125 &TraceRetainConfig {
1126 max_sessions: Some(2),
1127 max_age_secs: None,
1128 },
1129 NOW,
1130 );
1131
1132 assert_eq!(pruned.len(), 1);
1133 assert_eq!(pruned[0].name, all[0]);
1134 assert_eq!(names(records.path()), vec![all[1].clone(), all[2].clone()]);
1135 let mut kept = vec![all[1].clone(), all[2].clone(), workdir_only];
1136 kept.sort();
1137 assert_eq!(names(workdirs.path()), kept);
1138 }
1139
1140 #[test]
1144 fn retention_prunes_a_created_work_folder_and_never_the_operators() {
1145 let records = tempfile::tempdir().unwrap();
1146 let operator = tempfile::tempdir().unwrap();
1147 let manifest_dir = tempfile::tempdir().unwrap();
1148 let old = ses(NOW - 2000, 1);
1149 let current = ses(NOW, 2);
1150 let policy = TraceRetainConfig {
1151 max_sessions: Some(1),
1152 max_age_secs: None,
1153 };
1154 let open = |path: &std::path::Path| crate::workdir_fs::WorkdirRoot::open(path).unwrap();
1155
1156 make_sessions(records.path(), &[old.clone(), current.clone()]);
1157 let planted = [
1158 operator.path().join(".murmur").join(&old),
1159 operator.path().join(&old),
1160 ];
1161 for dir in &planted {
1162 std::fs::create_dir_all(dir).unwrap();
1163 std::fs::write(dir.join("keep"), b"x").unwrap();
1164 }
1165 assert_eq!(
1166 created_workdirs(Some(operator.path()), manifest_dir.path()),
1167 None
1168 );
1169 let pruned = prune_sessions(
1170 &open(records.path()),
1171 created_workdirs(Some(operator.path()), manifest_dir.path())
1172 .map(|dir| open(&dir))
1173 .as_ref(),
1174 ¤t,
1175 &policy,
1176 NOW,
1177 );
1178 assert_eq!(pruned.len(), 1);
1179 assert_eq!(names(records.path()), vec![current.clone()]);
1180 for dir in &planted {
1181 assert!(dir.join("keep").is_file(), "{} survives", dir.display());
1182 }
1183
1184 make_sessions(records.path(), std::slice::from_ref(&old));
1185 let workdirs = created_workdirs(None, manifest_dir.path()).unwrap();
1186 assert_eq!(workdirs, manifest_dir.path().join("workdir"));
1187 std::fs::create_dir_all(workdirs.join(&old)).unwrap();
1188 std::fs::create_dir_all(workdirs.join(¤t)).unwrap();
1189 let pruned = prune_sessions(
1190 &open(records.path()),
1191 Some(&open(&workdirs)),
1192 ¤t,
1193 &policy,
1194 NOW,
1195 );
1196 assert_eq!(pruned.len(), 1);
1197 assert!(!workdirs.join(&old).exists());
1198 assert!(workdirs.join(¤t).exists());
1199 }
1200
1201 #[test]
1204 fn pruning_a_session_removes_its_temp_folder() {
1205 use std::os::unix::fs::PermissionsExt;
1206
1207 let root = tempfile::tempdir().unwrap();
1208 let old = ses(NOW - 2000, 1);
1209 let current = ses(NOW, 2);
1210 make_sessions(root.path(), &[old.clone(), current.clone()]);
1211 let folder = root
1212 .path()
1213 .join(&old)
1214 .join(crate::temp_dir::SESSION_TEMP_DIR_NAME);
1215 let read_only = folder.join("read-only");
1216 let sealed_shut = folder.join("sealed-shut");
1217 std::fs::create_dir_all(&read_only).unwrap();
1218 std::fs::create_dir_all(&sealed_shut).unwrap();
1219 std::fs::write(read_only.join("file"), b"left behind").unwrap();
1220 std::fs::set_permissions(&read_only, std::fs::Permissions::from_mode(0o555)).unwrap();
1221 std::fs::set_permissions(&sealed_shut, std::fs::Permissions::from_mode(0o000)).unwrap();
1222
1223 let pruned = prune_sessions(
1224 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1225 None,
1226 ¤t,
1227 &TraceRetainConfig {
1228 max_sessions: Some(1),
1229 max_age_secs: None,
1230 },
1231 NOW,
1232 );
1233
1234 assert_eq!(pruned.len(), 1, "{pruned:?}");
1235 assert_eq!(pruned[0].name, old);
1236 assert!(!root.path().join(&old).exists(), "the session went whole");
1237 assert_eq!(names(root.path()), vec![current]);
1238 }
1239
1240 #[test]
1243 fn a_link_in_place_of_a_created_work_folder_is_not_followed() {
1244 let records = tempfile::tempdir().unwrap();
1245 let workdirs = tempfile::tempdir().unwrap();
1246 let elsewhere = tempfile::tempdir().unwrap();
1247 std::fs::write(elsewhere.path().join("keep"), b"x").unwrap();
1248 let all: Vec<String> = (1..=2).map(|n| ses(NOW - (2 - n) * 1000, n)).collect();
1249 make_sessions(records.path(), &all);
1250 std::os::unix::fs::symlink(elsewhere.path(), workdirs.path().join(&all[0])).unwrap();
1251
1252 let pruned = prune_sessions(
1253 &crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
1254 Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
1255 &all[1],
1256 &TraceRetainConfig {
1257 max_sessions: Some(1),
1258 max_age_secs: None,
1259 },
1260 NOW,
1261 );
1262
1263 assert_eq!(pruned.len(), 1);
1264 assert!(elsewhere.path().join("keep").exists());
1265 }
1266
1267 #[test]
1270 fn an_empty_policy_removes_nothing() {
1271 let root = tempfile::tempdir().unwrap();
1272 let all = [ses(NOW - 9_000_000, 1), ses(NOW, 2)];
1273 make_sessions(root.path(), &all);
1274 assert!(prune_sessions(
1275 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1276 None,
1277 &all[1],
1278 &TraceRetainConfig {
1279 max_sessions: None,
1280 max_age_secs: None
1281 },
1282 NOW
1283 )
1284 .is_empty());
1285 assert_eq!(names(root.path()), all.to_vec());
1286 }
1287
1288 fn write_record(dir: &Path, capsule: Option<&str>, count: u64) -> PathBuf {
1292 std::fs::create_dir_all(dir).unwrap();
1293 let path = dir.join(RECORD_FILE_NAME);
1294 let mut contents = String::new();
1295 if let Some(capsule) = capsule {
1296 contents.push_str(
1297 &header_line(&RecordHeader {
1298 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
1299 capsule: capsule.to_string(),
1300 created_ms: NOW,
1301 truncated: None,
1302 })
1303 .unwrap(),
1304 );
1305 }
1306 for i in 1..=count {
1307 contents.push_str(&message_line(
1308 &msg(NOW - (count - i) * 1000, i),
1309 &format!("m{i}"),
1310 ));
1311 contents.push('\n');
1312 }
1313 std::fs::write(&path, contents).unwrap();
1314 path
1315 }
1316
1317 fn image_line(id: &str, shas: &[&str]) -> String {
1319 let mut content = vec![serde_json::json!({"type": "text", "text": "shot"})];
1320 for sha in shas {
1321 content.push(serde_json::json!({
1322 "type": "image",
1323 "source": {"type": "file", "sha256": sha, "media_type": "image/png", "bytes": 4},
1324 }));
1325 }
1326 serde_json::json!({"id": id, "role": "tool", "content": content}).to_string()
1327 }
1328
1329 #[test]
1332 fn truncation_removes_blobs_no_surviving_line_names() {
1333 let dir = tempfile::tempdir().unwrap();
1334 let path = dir.path().join(RECORD_FILE_NAME);
1335 let (old, shared, new) = ("a".repeat(64), "b".repeat(64), "c".repeat(64));
1336 let lines = [
1337 image_line(&msg(NOW - 3000, 1), &[&old, &shared]),
1338 image_line(&msg(NOW - 2000, 2), &[&old]),
1339 image_line(&msg(NOW - 1000, 3), &[&shared]),
1340 image_line(&msg(NOW, 4), &[&new]),
1341 ];
1342 std::fs::write(&path, lines.join("\n") + "\n").unwrap();
1343 let blobs = dir.path().join("blobs");
1344 std::fs::create_dir(&blobs).unwrap();
1345 for name in [&old, &shared, &new, &"notes.txt".to_string()] {
1346 std::fs::write(blobs.join(name), b"\x89PNG").unwrap();
1347 }
1348
1349 let outcome = truncate_record(&path, 2, "capsule").unwrap();
1350 assert_eq!(outcome.dropped, 2);
1351 assert_eq!(outcome.blobs_removed, 1);
1352 assert!(!blobs.join(&old).exists());
1353 assert!(blobs.join(&shared).exists(), "a kept line still names it");
1354 assert!(blobs.join(&new).exists());
1355 assert!(blobs.join("notes.txt").exists());
1356 }
1357
1358 #[test]
1360 fn truncation_without_blobs_removes_nothing() {
1361 let dir = tempfile::tempdir().unwrap();
1362 let path = write_record(dir.path(), Some("capsule"), 5);
1363 let outcome = truncate_record(&path, 2, "capsule").unwrap();
1364 assert_eq!(outcome.dropped, 3);
1365 assert_eq!(outcome.blobs_removed, 0);
1366 assert!(!dir.path().join("blobs").exists());
1367 }
1368
1369 #[test]
1372 fn max_messages_leaves_the_newest_n_with_their_ids_intact() {
1373 let dir = tempfile::tempdir().unwrap();
1374 let path = write_record(dir.path(), Some("capsule"), 10);
1375 let before: Vec<String> = std::fs::read_to_string(&path)
1376 .unwrap()
1377 .lines()
1378 .filter_map(parse_message_line)
1379 .map(|m| m["id"].as_str().unwrap().to_string())
1380 .collect();
1381
1382 let outcome = truncate_record(&path, 3, "capsule").unwrap();
1383 assert_eq!(outcome.dropped, 7);
1384 assert_eq!(outcome.kept, 3);
1385 assert_eq!(outcome.last_dropped_id, before[6]);
1386 assert_eq!(outcome.oldest_surviving_id, before[7]);
1387
1388 let raw = std::fs::read_to_string(&path).unwrap();
1389 let lines: Vec<&str> = raw.lines().collect();
1390 assert_eq!(
1391 lines.len(),
1392 4,
1393 "a header plus exactly three messages: {lines:?}"
1394 );
1395 for line in &lines {
1396 serde_json::from_str::<serde_json::Value>(line).expect("every line is JSON");
1397 }
1398 let header = parse_header(lines[0]).expect("the first line is the header");
1399 assert_eq!(header.capsule, "capsule");
1400 let marker = header.truncated.expect("the header records the drop");
1401 assert_eq!(marker.dropped, 7);
1402 assert_eq!(marker.last_dropped_id, before[6]);
1403 assert_eq!(marker.oldest_surviving_id, before[7]);
1404
1405 let after: Vec<String> = lines[1..]
1406 .iter()
1407 .map(|line| {
1408 parse_message_line(line).expect("a message with a string role")["id"]
1409 .as_str()
1410 .unwrap()
1411 .to_string()
1412 })
1413 .collect();
1414 assert_eq!(after, before[7..], "surviving ids are the bytes they were");
1415 }
1416
1417 #[test]
1420 fn a_header_line_is_not_counted_as_a_message() {
1421 let dir = tempfile::tempdir().unwrap();
1422 let headerless = write_record(&dir.path().join("a"), None, 4);
1423 let owned = write_record(&dir.path().join("b"), Some("capsule"), 4);
1424 assert_eq!(count_messages(&headerless), 4);
1425 assert_eq!(count_messages(&owned), 4);
1426 }
1427
1428 #[test]
1431 fn a_second_truncation_accumulates_the_dropped_count() {
1432 let dir = tempfile::tempdir().unwrap();
1433 let path = write_record(dir.path(), Some("capsule"), 10);
1434 truncate_record(&path, 6, "capsule").unwrap();
1435 truncate_record(&path, 2, "capsule").unwrap();
1436 let marker = read_header(&path).unwrap().truncated.unwrap();
1437 assert_eq!(marker.dropped, 8);
1438 assert_eq!(count_messages(&path), 2);
1439 }
1440
1441 #[test]
1442 fn truncation_rewrite_keeps_the_record_owner_only() {
1443 use std::os::unix::fs::PermissionsExt;
1444 let dir = tempfile::tempdir().unwrap();
1445 let path = write_record(dir.path(), Some("capsule"), 5);
1446 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
1447
1448 truncate_record(&path, 2, "capsule").unwrap();
1449
1450 assert_eq!(count_messages(&path), 2);
1451 assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
1452 }
1453
1454 #[test]
1456 fn a_record_inside_the_limit_is_not_rewritten() {
1457 let dir = tempfile::tempdir().unwrap();
1458 let path = write_record(dir.path(), Some("capsule"), 3);
1459 let before = std::fs::read(&path).unwrap();
1460 let outcome = truncate_record(&path, 3, "capsule").unwrap();
1461 assert_eq!(outcome.dropped, 0);
1462 assert_eq!(std::fs::read(&path).unwrap(), before);
1463 }
1464
1465 #[test]
1469 fn a_staged_truncation_leaves_the_original_intact_until_the_rename() {
1470 let dir = tempfile::tempdir().unwrap();
1471 let path = write_record(dir.path(), Some("capsule"), 10);
1472 let before = std::fs::read(&path).unwrap();
1473
1474 let (staged, outcome) = stage_truncation(&path, 3, "capsule", NOW).unwrap().unwrap();
1475 assert_eq!(outcome.dropped, 7);
1476
1477 let temp = staged.temp_path().to_path_buf();
1478 assert_eq!(
1479 temp.parent(),
1480 path.parent(),
1481 "the temp file is in the record's own directory, so the rename is atomic"
1482 );
1483 assert_ne!(temp.file_name().unwrap(), RECORD_FILE_NAME);
1484
1485 assert_eq!(
1486 std::fs::read(&path).unwrap(),
1487 before,
1488 "the original is byte-identical before the rename"
1489 );
1490 assert_eq!(count_messages(&path), 10);
1491 for line in std::fs::read_to_string(&path).unwrap().lines() {
1492 serde_json::from_str::<serde_json::Value>(line).expect("the original parses whole");
1493 }
1494
1495 drop(staged);
1497 assert!(!temp.exists());
1498 assert_eq!(std::fs::read(&path).unwrap(), before);
1499 assert_eq!(count_messages(&path), 10);
1500 }
1501
1502 #[test]
1508 fn record_pruning_skips_another_capsules_records_and_every_unowned_one() {
1509 let root = tempfile::tempdir().unwrap();
1510 write_record(&root.path().join("mine"), Some("mine"), 3);
1511 write_record(&root.path().join("theirs"), Some("theirs"), 3);
1512 write_record(&root.path().join("unowned"), None, 3);
1513 write_record(&root.path().join("current"), Some("mine"), 3);
1514
1515 let far_future = now_ms() + 3_600_000;
1517 let pruned = prune_records(
1518 root.path(),
1519 "mine",
1520 Some("current"),
1521 &ContextRetainConfig {
1522 max_messages: None,
1523 max_age_secs: Some(1),
1524 },
1525 far_future,
1526 );
1527
1528 assert_eq!(pruned.len(), 1, "{pruned:?}");
1529 assert_eq!(pruned[0].context_id, "mine");
1530 assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
1531 assert_eq!(names(root.path()), vec!["current", "theirs", "unowned"]);
1532 }
1533
1534 #[test]
1537 fn a_record_written_inside_the_window_is_kept() {
1538 let root = tempfile::tempdir().unwrap();
1539 write_record(&root.path().join("fresh"), Some("mine"), 3);
1540 let pruned = prune_records(
1541 root.path(),
1542 "mine",
1543 None,
1544 &ContextRetainConfig {
1545 max_messages: None,
1546 max_age_secs: Some(90 * 86_400),
1547 },
1548 now_ms(),
1549 );
1550 assert!(pruned.is_empty());
1551 assert_eq!(names(root.path()), vec!["fresh"]);
1552 }
1553
1554 #[test]
1557 fn max_messages_alone_removes_no_record() {
1558 let root = tempfile::tempdir().unwrap();
1559 write_record(&root.path().join("ctx"), Some("mine"), 100);
1560 assert!(prune_records(
1561 root.path(),
1562 "mine",
1563 None,
1564 &ContextRetainConfig {
1565 max_messages: Some(1),
1566 max_age_secs: None
1567 },
1568 now_ms() + 3_600_000,
1569 )
1570 .is_empty());
1571 assert_eq!(names(root.path()), vec!["ctx"]);
1572 }
1573
1574 fn frm(ms: u64, tail: u64) -> FormationId {
1578 FormationId::parse(&format!("frm_{ms:012x}{tail:020x}")).unwrap()
1579 }
1580
1581 fn retain(max_sessions: Option<u32>, max_age_secs: Option<u64>) -> TraceRetainConfig {
1582 TraceRetainConfig {
1583 max_sessions,
1584 max_age_secs,
1585 }
1586 }
1587
1588 fn make_formation(root: &Path, id: &FormationId, marker: Option<&str>) -> PathBuf {
1591 let dir = root.join(id.as_str());
1592 let session = dir.join("worker/.murmur").join(ses(NOW, 1));
1593 std::fs::create_dir_all(&session).unwrap();
1594 std::fs::write(session.join("trace.jsonl"), "{}\n").unwrap();
1595 if let Some(marker) = marker {
1596 std::fs::write(
1597 dir.join(crate::formation_launch::FORMATION_MARKER_FILE),
1598 marker,
1599 )
1600 .unwrap();
1601 }
1602 dir
1603 }
1604
1605 fn marker_naming(project: &Path) -> String {
1606 serde_json::to_string(&FormationMarker {
1607 project_dir: project.to_path_buf(),
1608 launcher_pid: 1,
1609 launcher_start: String::new(),
1610 })
1611 .unwrap()
1612 }
1613
1614 fn make_owned(root: &Path, project: &Path, ids: &[&FormationId]) {
1616 for id in ids {
1617 make_formation(root, id, Some(&marker_naming(project)));
1618 }
1619 }
1620
1621 fn removed(pruned: &[PrunedFormation]) -> Vec<(FormationId, &'static str)> {
1622 pruned
1623 .iter()
1624 .map(|p| (p.formation_id.clone(), p.reason))
1625 .collect()
1626 }
1627
1628 const DAY_MS: u64 = 86_400_000;
1629
1630 #[test]
1633 fn prune_formations_never_removes_the_current_or_a_later_formation() {
1634 let root = tempfile::tempdir().unwrap();
1635 let project = tempfile::tempdir().unwrap();
1636 let old = frm(NOW - 3 * DAY_MS, 1);
1637 let current = frm(NOW - 2 * DAY_MS, 2);
1638 let later = frm(NOW - 2 * DAY_MS + 1000, 3);
1639 make_owned(root.path(), project.path(), &[&old, ¤t, &later]);
1640
1641 let pruned = prune_formations(
1642 root.path(),
1643 ¤t,
1644 project.path(),
1645 &retain(Some(1), Some(1)),
1646 NOW,
1647 |_, _| false,
1648 );
1649
1650 assert_eq!(
1651 removed(&pruned),
1652 [(old.clone(), crate::trace::RETENTION_REASON_MAX_AGE)]
1653 );
1654 assert_eq!(pruned[0].path, root.path().join(old.as_str()));
1655 assert_eq!(
1656 names(root.path()),
1657 vec![current.as_str().to_string(), later.as_str().to_string()]
1658 );
1659 }
1660
1661 #[test]
1664 fn prune_formations_keeps_the_newest_n_with_the_current_counted() {
1665 let root = tempfile::tempdir().unwrap();
1666 let project = tempfile::tempdir().unwrap();
1667 let all: Vec<FormationId> = (1..=5).map(|n| frm(NOW - (5 - n) * 1000, n)).collect();
1668 make_owned(root.path(), project.path(), &all.iter().collect::<Vec<_>>());
1669
1670 let pruned = prune_formations(
1671 root.path(),
1672 &all[3],
1673 project.path(),
1674 &retain(Some(3), None),
1675 NOW,
1676 |_, _| false,
1677 );
1678
1679 let reason = crate::trace::RETENTION_REASON_MAX_SESSIONS;
1680 assert_eq!(
1681 removed(&pruned),
1682 [(all[1].clone(), reason), (all[0].clone(), reason)]
1683 );
1684 assert_eq!(
1685 names(root.path()),
1686 all[2..]
1687 .iter()
1688 .map(|id| id.as_str().to_string())
1689 .collect::<Vec<_>>()
1690 );
1691 }
1692
1693 #[test]
1696 fn prune_formations_reads_age_from_the_id() {
1697 let root = tempfile::tempdir().unwrap();
1698 let project = tempfile::tempdir().unwrap();
1699 let old_a = frm(NOW - 7_200_000, 1);
1700 let old_b = frm(NOW - 7_100_000, 2);
1701 let recent = frm(NOW - 60_000, 3);
1702 let current = frm(NOW, 4);
1703 make_owned(
1704 root.path(),
1705 project.path(),
1706 &[&old_a, &old_b, &recent, ¤t],
1707 );
1708
1709 let pruned = prune_formations(
1710 root.path(),
1711 ¤t,
1712 project.path(),
1713 &retain(None, Some(900)),
1714 NOW,
1715 |_, _| false,
1716 );
1717
1718 let reason = crate::trace::RETENTION_REASON_MAX_AGE;
1719 assert_eq!(removed(&pruned), [(old_b, reason), (old_a, reason)]);
1720 assert_eq!(
1721 names(root.path()),
1722 vec![recent.as_str().to_string(), current.as_str().to_string()]
1723 );
1724 }
1725
1726 #[test]
1729 fn prune_formations_attributes_reasons_age_first() {
1730 let root = tempfile::tempdir().unwrap();
1731 let project = tempfile::tempdir().unwrap();
1732 let old = frm(NOW - 7_200_000, 1);
1733 let recent = frm(NOW - 60_000, 2);
1734 let current = frm(NOW, 3);
1735 make_owned(root.path(), project.path(), &[&old, &recent, ¤t]);
1736
1737 let pruned = prune_formations(
1738 root.path(),
1739 ¤t,
1740 project.path(),
1741 &retain(Some(1), Some(900)),
1742 NOW,
1743 |_, _| false,
1744 );
1745
1746 assert_eq!(
1747 removed(&pruned),
1748 [
1749 (recent, crate::trace::RETENTION_REASON_MAX_SESSIONS),
1750 (old, crate::trace::RETENTION_REASON_MAX_AGE),
1751 ]
1752 );
1753 }
1754
1755 #[test]
1758 fn prune_formations_skips_a_live_directory_and_keeps_its_rank() {
1759 let root = tempfile::tempdir().unwrap();
1760 let project = tempfile::tempdir().unwrap();
1761 let older = frm(NOW - 3000, 1);
1762 let live = frm(NOW - 2000, 2);
1763 let current = frm(NOW, 3);
1764 make_owned(root.path(), project.path(), &[&older, &live, ¤t]);
1765
1766 let asked = std::cell::RefCell::new(Vec::new());
1767 let pruned = prune_formations(
1768 root.path(),
1769 ¤t,
1770 project.path(),
1771 &retain(Some(1), None),
1772 NOW,
1773 |id, marker| {
1774 assert_eq!(marker.launcher_pid, 1);
1775 asked.borrow_mut().push(id.clone());
1776 *id == live
1777 },
1778 );
1779
1780 assert_eq!(
1781 removed(&pruned),
1782 [(older.clone(), crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1783 );
1784 assert_eq!(*asked.borrow(), [live.clone(), older]);
1785 assert_eq!(
1786 names(root.path()),
1787 vec![live.as_str().to_string(), current.as_str().to_string()]
1788 );
1789
1790 let kept = frm(NOW - 2500, 4);
1792 make_owned(root.path(), project.path(), &[&kept]);
1793 let pruned = prune_formations(
1794 root.path(),
1795 ¤t,
1796 project.path(),
1797 &retain(Some(3), None),
1798 NOW,
1799 |id, _| *id == live,
1800 );
1801 assert!(pruned.is_empty(), "{pruned:?}");
1802 let pruned = prune_formations(
1803 root.path(),
1804 ¤t,
1805 project.path(),
1806 &retain(Some(2), None),
1807 NOW,
1808 |id, _| *id == live,
1809 );
1810 assert_eq!(
1811 removed(&pruned),
1812 [(kept, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1813 );
1814 }
1815
1816 #[test]
1819 fn prune_formations_never_ranks_or_removes_unowned_or_foreign_directories() {
1820 let root = tempfile::tempdir().unwrap();
1821 let project = tempfile::tempdir().unwrap();
1822 let other = tempfile::tempdir().unwrap();
1823 let owned_old = frm(NOW - 3 * DAY_MS, 1);
1824 let owned_kept = frm(NOW - 2 * DAY_MS, 2);
1825 let unmarked = frm(NOW - 2 * DAY_MS + 1, 3);
1826 let garbled = frm(NOW - 2 * DAY_MS + 2, 4);
1827 let foreign = frm(NOW - 2 * DAY_MS + 3, 5);
1828 let current = frm(NOW, 6);
1829 make_owned(
1830 root.path(),
1831 project.path(),
1832 &[&owned_old, &owned_kept, ¤t],
1833 );
1834 make_formation(root.path(), &unmarked, None);
1835 make_formation(root.path(), &garbled, Some("not json {"));
1836 make_formation(root.path(), &foreign, Some(&marker_naming(other.path())));
1837
1838 let pruned = prune_formations(
1839 root.path(),
1840 ¤t,
1841 project.path(),
1842 &retain(Some(2), None),
1843 NOW,
1844 |_, _| false,
1845 );
1846 assert_eq!(
1847 removed(&pruned),
1848 [(owned_old, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1849 );
1850
1851 let pruned = prune_formations(
1852 root.path(),
1853 ¤t,
1854 project.path(),
1855 &retain(Some(1), Some(DAY_MS / 1000)),
1856 NOW,
1857 |_, _| false,
1858 );
1859 assert_eq!(
1860 removed(&pruned),
1861 [(owned_kept, crate::trace::RETENTION_REASON_MAX_AGE)]
1862 );
1863 for id in [&unmarked, &garbled, &foreign, ¤t] {
1864 assert!(root.path().join(id.as_str()).is_dir(), "{id}");
1865 }
1866 }
1867
1868 #[test]
1871 fn prune_formations_ignores_a_symlink_a_regular_file_and_a_name_that_is_not_an_id() {
1872 let root = tempfile::tempdir().unwrap();
1873 let elsewhere = tempfile::tempdir().unwrap();
1874 let project = tempfile::tempdir().unwrap();
1875 let marker = marker_naming(project.path());
1876 let target = make_formation(elsewhere.path(), &frm(NOW - 3 * DAY_MS, 1), Some(&marker));
1877 let linked = frm(NOW - 3 * DAY_MS, 2);
1878 std::os::unix::fs::symlink(&target, root.path().join(linked.as_str())).unwrap();
1879 std::fs::write(root.path().join(frm(NOW - 3 * DAY_MS, 3).as_str()), &marker).unwrap();
1880 for name in ["not-a-formation", "frm_ABC", "frm_0123"] {
1881 std::fs::create_dir(root.path().join(name)).unwrap();
1882 std::fs::write(
1883 root.path()
1884 .join(name)
1885 .join(crate::formation_launch::FORMATION_MARKER_FILE),
1886 &marker,
1887 )
1888 .unwrap();
1889 }
1890 let before = names(root.path());
1891
1892 let pruned = prune_formations(
1893 root.path(),
1894 &frm(NOW, 9),
1895 project.path(),
1896 &retain(Some(1), Some(1)),
1897 NOW,
1898 |_, _| false,
1899 );
1900
1901 assert!(pruned.is_empty(), "{pruned:?}");
1902 assert_eq!(names(root.path()), before);
1903 assert!(target.join("worker/.murmur").is_dir());
1904 }
1905
1906 #[test]
1908 fn prune_formations_with_neither_key_or_no_directory_removes_nothing() {
1909 let root = tempfile::tempdir().unwrap();
1910 let project = tempfile::tempdir().unwrap();
1911 let old = frm(NOW - 3 * DAY_MS, 1);
1912 make_owned(root.path(), project.path(), &[&old]);
1913
1914 let pruned = prune_formations(
1915 root.path(),
1916 &frm(NOW, 2),
1917 project.path(),
1918 &retain(None, None),
1919 NOW,
1920 |_, _| panic!("nothing is a candidate without a policy"),
1921 );
1922 assert!(pruned.is_empty());
1923 assert!(root.path().join(old.as_str()).is_dir());
1924
1925 assert!(prune_formations(
1926 &root.path().join("missing"),
1927 &frm(NOW, 2),
1928 project.path(),
1929 &retain(Some(1), Some(1)),
1930 NOW,
1931 |_, _| false,
1932 )
1933 .is_empty());
1934 }
1935
1936 #[test]
1939 fn prune_formations_compares_project_directories_canonically() {
1940 let root = tempfile::tempdir().unwrap();
1941 let project = tempfile::tempdir().unwrap();
1942 let links = tempfile::tempdir().unwrap();
1943 let link = links.path().join("project");
1944 std::os::unix::fs::symlink(project.path(), &link).unwrap();
1945 let old = frm(NOW - 3 * DAY_MS, 1);
1946 make_owned(
1947 root.path(),
1948 &std::fs::canonicalize(project.path()).unwrap(),
1949 &[&old],
1950 );
1951
1952 let pruned = prune_formations(
1953 root.path(),
1954 &frm(NOW, 2),
1955 &link,
1956 &retain(None, Some(1)),
1957 NOW,
1958 |_, _| false,
1959 );
1960 assert_eq!(
1961 removed(&pruned),
1962 [(old, crate::trace::RETENTION_REASON_MAX_AGE)]
1963 );
1964 }
1965
1966 #[test]
1968 fn prune_formations_leaves_a_directory_it_could_not_remove_out_of_the_result() {
1969 use std::os::unix::fs::PermissionsExt;
1970
1971 let root = tempfile::tempdir().unwrap();
1972 let project = tempfile::tempdir().unwrap();
1973 let stuck = frm(NOW - 3 * DAY_MS, 1);
1974 let gone = frm(NOW - 3 * DAY_MS + 1, 2);
1975 make_owned(root.path(), project.path(), &[&stuck, &gone]);
1976 let stuck_dir = root.path().join(stuck.as_str());
1977 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o500)).unwrap();
1978 if std::fs::write(stuck_dir.join("probe"), b"").is_ok() {
1980 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
1981 return;
1982 }
1983
1984 let pruned = prune_formations(
1985 root.path(),
1986 &frm(NOW, 3),
1987 project.path(),
1988 &retain(None, Some(1)),
1989 NOW,
1990 |_, _| false,
1991 );
1992 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
1993
1994 assert_eq!(
1995 removed(&pruned),
1996 [(gone, crate::trace::RETENTION_REASON_MAX_AGE)]
1997 );
1998 assert!(stuck_dir.is_dir());
1999 }
2000
2001 fn harness_entry_json(kind: &str, session_id: &str) -> String {
2004 json!({"type": kind, "session_id": session_id, "harness": "h", "driver": "d",
2005 "created_ms": 1, "updated_ms": 2})
2006 .to_string()
2007 }
2008
2009 #[test]
2012 fn the_context_finder_sees_either_file_in_every_store() {
2013 let root = tempfile::tempdir().unwrap();
2014 let dir = |store: &str, context: &str| {
2015 let path = root.path().join(store).join(context);
2016 std::fs::create_dir_all(&path).unwrap();
2017 path
2018 };
2019 std::fs::write(dir("b", "ctx").join(RECORD_FILE_NAME), "").unwrap();
2020 std::fs::write(
2021 dir("a", "ctx").join(HARNESS_SESSION_FILE_NAME),
2022 harness_entry_json(crate::harness_session::HARNESS_SESSION_TYPE, "s"),
2023 )
2024 .unwrap();
2025 dir("c", "ctx");
2026 std::fs::write(dir(".hidden", "ctx").join(RECORD_FILE_NAME), "").unwrap();
2027 std::fs::write(dir("a", ".ctx").join(RECORD_FILE_NAME), "").unwrap();
2028
2029 let found = find_context_dirs_in(root.path(), "ctx");
2030 let stores: Vec<&str> = found.iter().map(|dir| dir.record.as_str()).collect();
2031 assert_eq!(stores, ["a", "b"]);
2032 assert!(found.iter().all(|dir| dir.context_id == "ctx"));
2033 assert_eq!(found[1].path, root.path().join("b").join("ctx"));
2034 assert!(find_context_dirs_in(root.path(), ".ctx").is_empty());
2035 assert!(find_context_dirs_in(root.path(), "other").is_empty());
2036 assert!(find_context_dirs_in(&root.path().join("absent"), "ctx").is_empty());
2037 }
2038
2039 #[test]
2042 fn the_context_reader_reads_as_the_runtime_reads() {
2043 let dir = tempfile::tempdir().unwrap();
2044 let header = RecordHeader {
2045 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
2046 capsule: "owner".to_string(),
2047 created_ms: 7,
2048 truncated: None,
2049 };
2050 let first = json!({"role": "user", "content": "hi", "id": "msg_1", "kept": {"x": 1}});
2051 let second = json!({"role": "assistant", "content": [{"type": "text", "text": "yo"}]});
2052 std::fs::write(
2053 dir.path().join(RECORD_FILE_NAME),
2054 format!(
2055 "{}{first}\n\n{second}\n{}\n{{\"role\": \"us",
2056 header_line(&header).unwrap(),
2057 header_line(&header).unwrap().trim_end()
2058 ),
2059 )
2060 .unwrap();
2061
2062 let contents = read_context_dir(dir.path()).unwrap();
2063 assert_eq!(contents.header, Some(header.clone()));
2064 assert_eq!(contents.messages, vec![first.clone(), second]);
2065 assert_eq!(contents.harness_session, None);
2066
2067 std::fs::write(
2068 dir.path().join(RECORD_FILE_NAME),
2069 format!("{first}\n{}", header_line(&header).unwrap()),
2070 )
2071 .unwrap();
2072 let contents = read_context_dir(dir.path()).unwrap();
2073 assert_eq!(contents.header, None, "a header below line 1 is no header");
2074 assert_eq!(contents.messages, vec![first]);
2075
2076 let empty = tempfile::tempdir().unwrap();
2077 let contents = read_context_dir(empty.path()).unwrap();
2078 assert_eq!(contents.header, None);
2079 assert!(contents.messages.is_empty());
2080
2081 std::fs::create_dir(empty.path().join(RECORD_FILE_NAME)).unwrap();
2082 assert!(
2083 read_context_dir(empty.path()).is_err(),
2084 "an unreadable record"
2085 );
2086 }
2087
2088 #[test]
2091 fn the_context_reader_reads_no_session_where_the_runtime_does() {
2092 let dir = tempfile::tempdir().unwrap();
2093 let file = dir.path().join(HARNESS_SESSION_FILE_NAME);
2094 assert_eq!(read_context_dir(dir.path()).unwrap().harness_session, None);
2095
2096 let session_type = crate::harness_session::HARNESS_SESSION_TYPE;
2097 for (body, expected) in [
2098 ("{not json".to_string(), false),
2099 (harness_entry_json("another-tool", "s"), false),
2100 (harness_entry_json(session_type, ""), false),
2101 (harness_entry_json(session_type, "s"), true),
2102 ] {
2103 std::fs::write(&file, &body).unwrap();
2104 let session = read_context_dir(dir.path()).unwrap().harness_session;
2105 assert_eq!(session.is_some(), expected, "{body}");
2106 if let Some(session) = session {
2107 assert_eq!(
2108 (session.session_id.as_str(), session.harness.as_str()),
2109 ("s", "h")
2110 );
2111 assert_eq!((session.created_ms, session.updated_ms), (1, 2));
2112 }
2113 }
2114 }
2115}