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}
93
94#[derive(Debug, Clone, PartialEq, Eq)]
96pub enum MessageStatus {
97 Present { position: u64, total: u64 },
99 Truncated {
103 dropped: u64,
104 oldest_surviving_id: String,
105 },
106 Unknown,
108}
109
110#[derive(Debug, Clone, PartialEq, Eq)]
112pub struct MessageLocation {
113 pub record: String,
114 pub context_id: String,
115 pub path: PathBuf,
116 pub status: MessageStatus,
117}
118
119#[derive(Debug, Clone, PartialEq, Eq)]
121pub struct RecordSummary {
122 pub record: String,
124 pub context_id: String,
126 pub path: PathBuf,
127 pub messages: u64,
129 pub bytes: u64,
131 pub last_touched_ms: Option<u64>,
134 pub capsule: Option<String>,
137 pub truncation: Option<TruncationMarker>,
139}
140
141#[derive(Debug, Clone, PartialEq, Eq)]
143pub struct RemovedRecord {
144 pub record: String,
145 pub context_id: String,
146 pub path: PathBuf,
148 pub messages: u64,
150}
151
152#[derive(Debug, Clone, PartialEq, Eq)]
155pub struct ContextDir {
156 pub record: String,
158 pub context_id: String,
160 pub path: PathBuf,
162}
163
164#[derive(Debug, Clone, PartialEq)]
166pub struct ContextContents {
167 pub header: Option<RecordHeader>,
169 pub messages: Vec<serde_json::Value>,
172 pub harness_session: Option<HarnessSessionEntry>,
174}
175
176pub fn session_id_timestamp_ms(name: &str) -> Option<u64> {
184 uuid_v7_timestamp_ms(name, "ses_")
185}
186
187pub(crate) fn uuid_v7_timestamp_ms(id: &str, prefix: &str) -> Option<u64> {
190 let hex = id.strip_prefix(prefix)?;
191 u64::from_str_radix(hex.get(..12)?, 16).ok()
192}
193
194pub(crate) fn created_workdirs(
198 workdir_flag: Option<&std::path::Path>,
199 manifest_dir: &std::path::Path,
200) -> Option<std::path::PathBuf> {
201 workdir_flag.is_none().then(|| manifest_dir.join("workdir"))
202}
203
204pub fn prune_sessions(
227 records_group: &crate::workdir_fs::WorkdirRoot,
228 created_workdirs: Option<&crate::workdir_fs::WorkdirRoot>,
229 current_session_id: &str,
230 policy: &murmur_artifact::TraceRetainConfig,
231 now_ms: u64,
232) -> Vec<PrunedSession> {
233 if policy.max_sessions.is_none() && policy.max_age_secs.is_none() {
234 return Vec::new();
235 }
236
237 let mut entries: Vec<String> = match records_group.list_dir("") {
238 Ok(Some(listed)) => listed
239 .into_iter()
240 .filter(|(_, kind)| *kind == crate::workdir_fs::EntryKind::Directory)
241 .map(|(name, _)| name)
242 .filter(|name| session_id_timestamp_ms(name).is_some())
243 .collect(),
244 Ok(None) | Err(_) => return Vec::new(),
245 };
246 entries.sort_unstable_by(|a, b| b.cmp(a));
248
249 let age_floor_ms = policy
250 .max_age_secs
251 .map(|secs| now_ms.saturating_sub(secs.saturating_mul(1000)));
252
253 let mut pruned = Vec::new();
254 for (rank, name) in entries.iter().enumerate() {
255 if name.as_str() >= current_session_id {
258 continue;
259 }
260 let over_age = age_floor_ms.is_some_and(|floor| {
261 session_id_timestamp_ms(name).is_some_and(|minted| minted < floor)
262 });
263 let over_count = policy.max_sessions.is_some_and(|max| rank >= max as usize);
264 if !over_age && !over_count {
265 continue;
266 }
267 let reason = if over_age {
268 crate::trace::RETENTION_REASON_MAX_AGE
269 } else {
270 crate::trace::RETENTION_REASON_MAX_SESSIONS
271 };
272 if records_group.remove_dir_all(name).is_ok() {
273 if let Some(workdirs) = created_workdirs {
274 let _ = workdirs.remove_dir_all(name);
275 }
276 pruned.push(PrunedSession {
277 name: name.clone(),
278 reason,
279 });
280 }
281 }
282 pruned
283}
284
285pub fn prune_formations(
311 formations_dir: &Path,
312 current: &FormationId,
313 project_dir: &Path,
314 policy: &murmur_artifact::TraceRetainConfig,
315 now_ms: u64,
316 is_live: impl Fn(&FormationId, &FormationMarker) -> bool,
317) -> Vec<PrunedFormation> {
318 if policy.max_sessions.is_none() && policy.max_age_secs.is_none() {
319 return Vec::new();
320 }
321 let Ok(entries) = std::fs::read_dir(formations_dir) else {
322 return Vec::new();
323 };
324 let project_dir = crate::formation_launch::canonical_project_dir(project_dir);
325 let mut owned: Vec<(FormationId, FormationMarker)> = entries
326 .filter_map(Result::ok)
327 .filter_map(|entry| {
328 let id = FormationId::parse(entry.file_name().to_str()?).ok()?;
329 let path = entry.path();
330 if !std::fs::symlink_metadata(&path).ok()?.is_dir() {
331 return None;
332 }
333 let marker = FormationMarker::read(&path)?;
334 (crate::formation_launch::canonical_project_dir(&marker.project_dir) == project_dir)
335 .then_some((id, marker))
336 })
337 .collect();
338 owned.sort_unstable_by(|(a, _), (b, _)| b.as_str().cmp(a.as_str()));
340
341 let age_floor_ms = policy
342 .max_age_secs
343 .map(|secs| now_ms.saturating_sub(secs.saturating_mul(1000)));
344
345 let mut pruned = Vec::new();
346 for (rank, (id, marker)) in owned.iter().enumerate() {
347 if id.as_str() >= current.as_str() {
348 continue;
349 }
350 let over_age = age_floor_ms.is_some_and(|floor| id.minted_at_ms() < floor);
351 let over_count = policy.max_sessions.is_some_and(|max| rank >= max as usize);
352 if !over_age && !over_count {
353 continue;
354 }
355 if is_live(id, marker) {
356 continue;
357 }
358 let reason = if over_age {
359 crate::trace::RETENTION_REASON_MAX_AGE
360 } else {
361 crate::trace::RETENTION_REASON_MAX_SESSIONS
362 };
363 let path = formations_dir.join(id.as_str());
364 let records = crate::records_dir::records_group(&path).ok();
368 if std::fs::remove_dir_all(&path).is_ok() {
369 if let Some(records) = records {
370 let _ = std::fs::remove_dir_all(records);
371 }
372 pruned.push(PrunedFormation {
373 formation_id: id.clone(),
374 path,
375 reason,
376 });
377 }
378 }
379 pruned
380}
381
382pub fn prune_records(
401 record_root: &Path,
402 capsule_name: &str,
403 current_context_id: Option<&str>,
404 policy: &murmur_artifact::ContextRetainConfig,
405 now_ms: u64,
406) -> Vec<PrunedRecord> {
407 let Some(max_age_secs) = policy.max_age_secs else {
408 return Vec::new();
409 };
410 let floor_ms = now_ms.saturating_sub(max_age_secs.saturating_mul(1000));
411
412 let mut pruned = Vec::new();
413 for (context_id, dir) in context_dirs(record_root) {
414 if Some(context_id.as_str()) == current_context_id {
415 continue;
416 }
417 let file = dir.join(RECORD_FILE_NAME);
418 match read_header(&file) {
419 Some(header) if header.capsule == capsule_name => {}
420 _ => continue,
421 }
422 let Some(touched_ms) = last_touched_ms(&file) else {
423 continue;
424 };
425 if touched_ms >= floor_ms {
426 continue;
427 }
428 if std::fs::remove_dir_all(&dir).is_ok() {
429 pruned.push(PrunedRecord {
430 context_id,
431 path: dir,
432 reason: crate::trace::RETENTION_REASON_MAX_AGE,
433 });
434 }
435 }
436 pruned
437}
438
439pub fn truncate_record(path: &Path, keep: u32, capsule: &str) -> Result<TruncationOutcome, String> {
454 match stage_truncation(path, keep, capsule, now_ms())? {
455 None => Ok(TruncationOutcome {
456 dropped: 0,
457 kept: count_messages(path),
458 oldest_surviving_id: String::new(),
459 last_dropped_id: String::new(),
460 }),
461 Some((staged, outcome)) => {
462 staged.commit()?;
463 Ok(outcome)
464 }
465 }
466}
467
468pub(crate) fn stage_truncation(
474 path: &Path,
475 keep: u32,
476 capsule: &str,
477 at_ms: u64,
478) -> Result<Option<(StagedRewrite, TruncationOutcome)>, String> {
479 let raw = match std::fs::read_to_string(path) {
480 Ok(raw) => raw,
481 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
482 Err(err) => return Err(format!("{}: {err}", path.display())),
483 };
484
485 let existing = read_header(path);
486 let lines: Vec<&str> = raw.lines().collect();
487 let message_indices: Vec<usize> = lines
490 .iter()
491 .enumerate()
492 .filter(|(_, line)| parse_message_line(line).is_some())
493 .map(|(index, _)| index)
494 .collect();
495
496 let total = message_indices.len();
497 let keep = keep as usize;
498 if total <= keep {
499 return Ok(None);
500 }
501 let dropped = total - keep;
502 let first_kept_index = message_indices[dropped];
503
504 let id_of = |index: usize| -> String {
505 parse_message_line(lines[index])
506 .as_ref()
507 .and_then(|message| crate::agent::message_id(message).map(str::to_string))
508 .unwrap_or_default()
509 };
510 let last_dropped_id = id_of(message_indices[dropped - 1]);
511 let oldest_surviving_id = message_indices
512 .get(dropped)
513 .map(|i| id_of(*i))
514 .unwrap_or_default();
515
516 let header = RecordHeader {
517 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
518 capsule: existing
519 .as_ref()
520 .map(|h| h.capsule.clone())
521 .unwrap_or_else(|| capsule.to_string()),
522 created_ms: existing.as_ref().map(|h| h.created_ms).unwrap_or(at_ms),
523 truncated: Some(TruncationMarker {
524 dropped: existing
527 .as_ref()
528 .and_then(|h| h.truncated.as_ref())
529 .map(|marker| marker.dropped)
530 .unwrap_or(0)
531 + dropped as u64,
532 oldest_surviving_id: oldest_surviving_id.clone(),
533 last_dropped_id: last_dropped_id.clone(),
534 at_ms,
535 }),
536 };
537
538 let mut contents = header_line(&header)?;
539 for line in &lines[first_kept_index..] {
540 contents.push_str(line);
541 contents.push('\n');
542 }
543
544 let staged = StagedRewrite::stage_with_mode(
545 path,
546 contents.as_bytes(),
547 Some(crate::conversation::RECORD_FILE_MODE),
548 )?;
549 Ok(Some((
550 staged,
551 TruncationOutcome {
552 dropped: dropped as u64,
553 kept: keep as u64,
554 oldest_surviving_id,
555 last_dropped_id,
556 },
557 )))
558}
559
560pub(crate) struct StagedRewrite {
566 target: PathBuf,
567 temp: PathBuf,
568 committed: bool,
569}
570
571impl StagedRewrite {
572 pub(crate) fn stage_with_mode(
575 target: &Path,
576 contents: &[u8],
577 mode: Option<u32>,
578 ) -> Result<Self, String> {
579 use std::io::Write;
580 use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
581
582 let dir = target
583 .parent()
584 .ok_or_else(|| format!("{} has no parent directory", target.display()))?;
585 let name = target
586 .file_name()
587 .map(|name| name.to_string_lossy().into_owned())
588 .unwrap_or_else(|| "record".to_string());
589 let temp = dir.join(format!(
590 ".{name}.{}.{}.tmp",
591 std::process::id(),
592 uuid::Uuid::now_v7().simple()
593 ));
594 let mut options = std::fs::OpenOptions::new();
595 options.write(true).create(true).truncate(true);
596 if let Some(mode) = mode {
597 options.mode(mode);
598 }
599 let mut file = options
600 .open(&temp)
601 .map_err(|err| format!("{}: {err}", temp.display()))?;
602 file.write_all(contents)
605 .and_then(|()| file.sync_all())
606 .map_err(|err| format!("{}: {err}", temp.display()))?;
607 if let Some(mode) = mode {
611 std::fs::set_permissions(&temp, std::fs::Permissions::from_mode(mode))
612 .map_err(|err| format!("{}: {err}", temp.display()))?;
613 }
614 Ok(Self {
615 target: target.to_path_buf(),
616 temp,
617 committed: false,
618 })
619 }
620
621 #[cfg(test)]
622 pub(crate) fn temp_path(&self) -> &Path {
623 &self.temp
624 }
625
626 pub(crate) fn commit(mut self) -> Result<(), String> {
627 std::fs::rename(&self.temp, &self.target)
628 .map_err(|err| format!("{}: {err}", self.target.display()))?;
629 self.committed = true;
630 Ok(())
631 }
632}
633
634impl Drop for StagedRewrite {
635 fn drop(&mut self) {
636 if !self.committed {
637 let _ = std::fs::remove_file(&self.temp);
638 }
639 }
640}
641
642pub fn list_records() -> Result<Vec<RecordSummary>, String> {
650 let root = conversations_root()?;
651 let mut summaries = Vec::new();
652 for record in child_dirs(&root) {
653 let store = root.join(&record);
654 for (context_id, dir) in context_dirs(&store) {
655 let path = dir.join(RECORD_FILE_NAME);
656 if !path.exists() {
657 continue;
658 }
659 let header = read_header(&path);
660 summaries.push(RecordSummary {
661 record: record.clone(),
662 context_id,
663 messages: count_messages(&path),
664 bytes: std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0),
665 last_touched_ms: last_touched_ms(&path),
666 capsule: header.as_ref().map(|h| h.capsule.clone()),
667 truncation: header.and_then(|h| h.truncated),
668 path,
669 });
670 }
671 }
672 summaries.sort_by(|a, b| {
673 a.record
674 .cmp(&b.record)
675 .then_with(|| a.context_id.cmp(&b.context_id))
676 });
677 Ok(summaries)
678}
679
680pub fn remove_record(record: &str, context_id: &str) -> Result<RemovedRecord, String> {
685 let dir = conversations_root()?.join(record).join(context_id);
686 if !dir.is_dir() {
687 return Err(format!("{} is not a context directory", dir.display()));
688 }
689 let messages = count_messages(&dir.join(RECORD_FILE_NAME));
690 std::fs::remove_dir_all(&dir).map_err(|err| format!("{}: {err}", dir.display()))?;
691 Ok(RemovedRecord {
692 record: record.to_string(),
693 context_id: context_id.to_string(),
694 path: dir,
695 messages,
696 })
697}
698
699pub fn locate_message(message_id: &str) -> Result<Vec<MessageLocation>, String> {
711 let minted_ms = message_id_timestamp_ms(message_id);
712 let mut found = Vec::new();
713 for summary in list_records()? {
714 let raw = std::fs::read_to_string(&summary.path).unwrap_or_default();
715 let mut position = 0u64;
716 let mut at = None;
717 for line in raw.lines() {
718 let Some(message) = parse_message_line(line) else {
719 continue;
720 };
721 position += 1;
722 if crate::agent::message_id(&message) == Some(message_id) {
723 at = Some(position);
724 }
725 }
726 let status = match (at, summary.truncation.as_ref()) {
727 (Some(position), _) => MessageStatus::Present {
728 position,
729 total: position.max(summary.messages),
730 },
731 (None, Some(marker)) => {
732 let dropped_at = message_id_timestamp_ms(&marker.last_dropped_id);
733 match (minted_ms, dropped_at) {
734 (Some(minted), Some(last)) if minted <= last => MessageStatus::Truncated {
735 dropped: marker.dropped,
736 oldest_surviving_id: marker.oldest_surviving_id.clone(),
737 },
738 _ => MessageStatus::Unknown,
739 }
740 }
741 (None, None) => MessageStatus::Unknown,
742 };
743 if status != MessageStatus::Unknown {
744 found.push(MessageLocation {
745 record: summary.record,
746 context_id: summary.context_id,
747 path: summary.path,
748 status,
749 });
750 }
751 }
752 Ok(found)
753}
754
755pub fn find_context_dirs(context_id: &str) -> Result<Vec<ContextDir>, String> {
762 Ok(find_context_dirs_in(&conversations_root()?, context_id))
763}
764
765fn find_context_dirs_in(root: &Path, context_id: &str) -> Vec<ContextDir> {
766 let mut found = Vec::new();
767 for record in child_dirs(root) {
768 for (name, path) in context_dirs(&root.join(&record)) {
769 if name != context_id {
770 continue;
771 }
772 if path.join(RECORD_FILE_NAME).exists() || path.join(HARNESS_SESSION_FILE_NAME).exists()
773 {
774 found.push(ContextDir {
775 record: record.clone(),
776 context_id: name,
777 path,
778 });
779 }
780 }
781 }
782 found
783}
784
785pub fn read_context_dir(dir: &Path) -> Result<ContextContents, String> {
793 let record = dir.join(RECORD_FILE_NAME);
794 let raw = match std::fs::read_to_string(&record) {
795 Ok(raw) => raw,
796 Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
797 Err(err) => return Err(format!("{}: {err}", record.display())),
798 };
799 Ok(ContextContents {
800 header: raw.lines().next().and_then(parse_header),
801 messages: record_messages(&raw, |_| {}),
802 harness_session: read_entry(&dir.join(HARNESS_SESSION_FILE_NAME)),
803 })
804}
805
806pub fn conversations_root() -> Result<PathBuf, String> {
810 Ok(crate::state_store::murmur_home_dir()?.join(CONVERSATION_ROOT_DIR))
811}
812
813pub fn now_ms() -> u64 {
815 std::time::SystemTime::now()
816 .duration_since(std::time::UNIX_EPOCH)
817 .map(|elapsed| elapsed.as_millis() as u64)
818 .unwrap_or(0)
819}
820
821fn child_dirs(dir: &Path) -> Vec<String> {
823 let mut names: Vec<String> = match std::fs::read_dir(dir) {
824 Ok(entries) => entries
825 .filter_map(Result::ok)
826 .filter(|entry| entry.path().is_dir())
827 .map(|entry| entry.file_name().to_string_lossy().into_owned())
828 .filter(|name| !name.starts_with('.'))
829 .collect(),
830 Err(_) => Vec::new(),
831 };
832 names.sort();
833 names
834}
835
836fn context_dirs(record_root: &Path) -> Vec<(String, PathBuf)> {
838 child_dirs(record_root)
839 .into_iter()
840 .map(|name| {
841 let path = record_root.join(&name);
842 (name, path)
843 })
844 .collect()
845}
846
847fn count_messages(path: &Path) -> u64 {
850 std::fs::read_to_string(path)
851 .map(|raw| {
852 raw.lines()
853 .filter(|line| parse_message_line(line).is_some())
854 .count() as u64
855 })
856 .unwrap_or(0)
857}
858
859fn last_touched_ms(path: &Path) -> Option<u64> {
861 std::fs::metadata(path)
862 .and_then(|metadata| metadata.modified())
863 .ok()?
864 .duration_since(std::time::UNIX_EPOCH)
865 .ok()
866 .map(|elapsed| elapsed.as_millis() as u64)
867}
868
869#[cfg(test)]
870mod tests {
871 use super::*;
872 use murmur_artifact::{ContextRetainConfig, TraceRetainConfig};
873 use serde_json::json;
874
875 fn ses(ms: u64, tail: u64) -> String {
878 format!("ses_{ms:012x}{tail:020x}")
879 }
880
881 fn msg(ms: u64, tail: u64) -> String {
883 format!("msg_{ms:012x}{tail:020x}")
884 }
885
886 fn message_line(id: &str, text: &str) -> String {
887 json!({"role": "user", "content": [{"type": "text", "text": text}], "id": id}).to_string()
888 }
889
890 fn make_sessions(root: &Path, names: &[String]) {
891 for name in names {
892 let dir = root.join(name);
893 std::fs::create_dir_all(dir.join("blobs")).unwrap();
894 std::fs::write(dir.join("trace.jsonl"), "{}\n").unwrap();
895 std::fs::write(dir.join("blobs").join("deadbeef"), b"body").unwrap();
896 }
897 }
898
899 fn names(root: &Path) -> Vec<String> {
900 let mut found: Vec<String> = std::fs::read_dir(root)
901 .unwrap()
902 .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
903 .collect();
904 found.sort();
905 found
906 }
907
908 const NOW: u64 = 1_756_400_000_000;
909
910 #[test]
914 fn max_sessions_leaves_exactly_the_newest_n_including_the_current() {
915 let root = tempfile::tempdir().unwrap();
916 let all: Vec<String> = (1..=6).map(|n| ses(NOW - (6 - n) * 1000, n)).collect();
917 make_sessions(root.path(), &all);
918
919 let pruned = prune_sessions(
920 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
921 None,
922 &all[5],
923 &TraceRetainConfig {
924 max_sessions: Some(3),
925 max_age_secs: None,
926 },
927 NOW,
928 );
929
930 assert_eq!(
931 names(root.path()),
932 vec![all[3].clone(), all[4].clone(), all[5].clone()]
933 );
934 let mut removed: Vec<String> = pruned.iter().map(|p| p.name.clone()).collect();
935 removed.sort();
936 assert_eq!(
937 removed,
938 vec![all[0].clone(), all[1].clone(), all[2].clone()]
939 );
940 assert!(pruned
941 .iter()
942 .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_SESSIONS));
943 for name in &all[..3] {
944 assert!(!root.path().join(name).exists(), "{name} went whole");
945 }
946 }
947
948 #[test]
952 fn max_age_reads_the_id_and_never_the_filesystem() {
953 let root = tempfile::tempdir().unwrap();
954 let old_a = ses(NOW - 7_200_000, 1);
955 let old_b = ses(NOW - 7_100_000, 2);
956 let recent = ses(NOW - 60_000, 3);
957 let current = ses(NOW, 4);
958 make_sessions(
959 root.path(),
960 &[
961 old_a.clone(),
962 old_b.clone(),
963 recent.clone(),
964 current.clone(),
965 ],
966 );
967
968 let pruned = prune_sessions(
969 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
970 None,
971 ¤t,
972 &TraceRetainConfig {
973 max_sessions: None,
974 max_age_secs: Some(900),
975 },
976 NOW,
977 );
978
979 assert_eq!(pruned.len(), 2, "{pruned:?}");
980 assert!(pruned
981 .iter()
982 .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_AGE));
983 assert_eq!(names(root.path()), vec![recent, current]);
984 }
985
986 #[test]
989 fn both_trace_keys_keep_only_sessions_inside_both() {
990 let root = tempfile::tempdir().unwrap();
991 let old = [ses(NOW - 7_200_000, 1), ses(NOW - 7_100_000, 2)];
992 let recent = [
993 ses(NOW - 300_000, 3),
994 ses(NOW - 200_000, 4),
995 ses(NOW - 100_000, 5),
996 ];
997 let current = ses(NOW, 6);
998 let mut all = old.to_vec();
999 all.extend(recent.iter().cloned());
1000 all.push(current.clone());
1001 make_sessions(root.path(), &all);
1002
1003 let pruned = prune_sessions(
1004 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1005 None,
1006 ¤t,
1007 &TraceRetainConfig {
1008 max_sessions: Some(2),
1009 max_age_secs: Some(900),
1010 },
1011 NOW,
1012 );
1013
1014 assert_eq!(names(root.path()), vec![recent[2].clone(), current.clone()]);
1015 assert_eq!(pruned.len(), 4);
1016 for session in &pruned {
1017 let expected = if old.contains(&session.name) {
1018 crate::trace::RETENTION_REASON_MAX_AGE
1019 } else {
1020 crate::trace::RETENTION_REASON_MAX_SESSIONS
1021 };
1022 assert_eq!(session.reason, expected, "{session:?}");
1023 }
1024 }
1025
1026 #[test]
1029 fn nothing_at_or_after_the_current_session_id_is_a_candidate() {
1030 let root = tempfile::tempdir().unwrap();
1031 let current = ses(NOW, 5);
1032 let sibling = ses(NOW + 1, 6);
1034 let older = ses(NOW - 1000, 4);
1035 make_sessions(
1036 root.path(),
1037 &[current.clone(), sibling.clone(), older.clone()],
1038 );
1039
1040 let pruned = prune_sessions(
1041 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1042 None,
1043 ¤t,
1044 &TraceRetainConfig {
1045 max_sessions: Some(1),
1046 max_age_secs: Some(1),
1047 },
1048 NOW,
1049 );
1050
1051 assert_eq!(pruned.len(), 1);
1052 assert_eq!(pruned[0].name, older);
1053 assert_eq!(names(root.path()), vec![current.clone(), sibling]);
1054 assert!(root.path().join(¤t).join("trace.jsonl").exists());
1055 }
1056
1057 #[test]
1061 fn a_pruned_session_takes_its_records_and_its_created_work_folder() {
1062 let records = tempfile::tempdir().unwrap();
1063 let workdirs = tempfile::tempdir().unwrap();
1064 let all: Vec<String> = (1..=3).map(|n| ses(NOW - (3 - n) * 1000, n)).collect();
1065 make_sessions(records.path(), &all);
1066 for name in &all {
1067 std::fs::create_dir_all(workdirs.path().join(name).join("out")).unwrap();
1068 }
1069 let workdir_only = ses(NOW - 9_000, 9);
1070 std::fs::create_dir_all(workdirs.path().join(&workdir_only)).unwrap();
1071
1072 let pruned = prune_sessions(
1073 &crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
1074 Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
1075 &all[2],
1076 &TraceRetainConfig {
1077 max_sessions: Some(2),
1078 max_age_secs: None,
1079 },
1080 NOW,
1081 );
1082
1083 assert_eq!(pruned.len(), 1);
1084 assert_eq!(pruned[0].name, all[0]);
1085 assert_eq!(names(records.path()), vec![all[1].clone(), all[2].clone()]);
1086 let mut kept = vec![all[1].clone(), all[2].clone(), workdir_only];
1087 kept.sort();
1088 assert_eq!(names(workdirs.path()), kept);
1089 }
1090
1091 #[test]
1095 fn retention_prunes_a_created_work_folder_and_never_the_operators() {
1096 let records = tempfile::tempdir().unwrap();
1097 let operator = tempfile::tempdir().unwrap();
1098 let manifest_dir = tempfile::tempdir().unwrap();
1099 let old = ses(NOW - 2000, 1);
1100 let current = ses(NOW, 2);
1101 let policy = TraceRetainConfig {
1102 max_sessions: Some(1),
1103 max_age_secs: None,
1104 };
1105 let open = |path: &std::path::Path| crate::workdir_fs::WorkdirRoot::open(path).unwrap();
1106
1107 make_sessions(records.path(), &[old.clone(), current.clone()]);
1108 let planted = [
1109 operator.path().join(".murmur").join(&old),
1110 operator.path().join(&old),
1111 ];
1112 for dir in &planted {
1113 std::fs::create_dir_all(dir).unwrap();
1114 std::fs::write(dir.join("keep"), b"x").unwrap();
1115 }
1116 assert_eq!(
1117 created_workdirs(Some(operator.path()), manifest_dir.path()),
1118 None
1119 );
1120 let pruned = prune_sessions(
1121 &open(records.path()),
1122 created_workdirs(Some(operator.path()), manifest_dir.path())
1123 .map(|dir| open(&dir))
1124 .as_ref(),
1125 ¤t,
1126 &policy,
1127 NOW,
1128 );
1129 assert_eq!(pruned.len(), 1);
1130 assert_eq!(names(records.path()), vec![current.clone()]);
1131 for dir in &planted {
1132 assert!(dir.join("keep").is_file(), "{} survives", dir.display());
1133 }
1134
1135 make_sessions(records.path(), std::slice::from_ref(&old));
1136 let workdirs = created_workdirs(None, manifest_dir.path()).unwrap();
1137 assert_eq!(workdirs, manifest_dir.path().join("workdir"));
1138 std::fs::create_dir_all(workdirs.join(&old)).unwrap();
1139 std::fs::create_dir_all(workdirs.join(¤t)).unwrap();
1140 let pruned = prune_sessions(
1141 &open(records.path()),
1142 Some(&open(&workdirs)),
1143 ¤t,
1144 &policy,
1145 NOW,
1146 );
1147 assert_eq!(pruned.len(), 1);
1148 assert!(!workdirs.join(&old).exists());
1149 assert!(workdirs.join(¤t).exists());
1150 }
1151
1152 #[test]
1155 fn pruning_a_session_removes_its_temp_folder() {
1156 use std::os::unix::fs::PermissionsExt;
1157
1158 let root = tempfile::tempdir().unwrap();
1159 let old = ses(NOW - 2000, 1);
1160 let current = ses(NOW, 2);
1161 make_sessions(root.path(), &[old.clone(), current.clone()]);
1162 let folder = root
1163 .path()
1164 .join(&old)
1165 .join(crate::temp_dir::SESSION_TEMP_DIR_NAME);
1166 let read_only = folder.join("read-only");
1167 let sealed_shut = folder.join("sealed-shut");
1168 std::fs::create_dir_all(&read_only).unwrap();
1169 std::fs::create_dir_all(&sealed_shut).unwrap();
1170 std::fs::write(read_only.join("file"), b"left behind").unwrap();
1171 std::fs::set_permissions(&read_only, std::fs::Permissions::from_mode(0o555)).unwrap();
1172 std::fs::set_permissions(&sealed_shut, std::fs::Permissions::from_mode(0o000)).unwrap();
1173
1174 let pruned = prune_sessions(
1175 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1176 None,
1177 ¤t,
1178 &TraceRetainConfig {
1179 max_sessions: Some(1),
1180 max_age_secs: None,
1181 },
1182 NOW,
1183 );
1184
1185 assert_eq!(pruned.len(), 1, "{pruned:?}");
1186 assert_eq!(pruned[0].name, old);
1187 assert!(!root.path().join(&old).exists(), "the session went whole");
1188 assert_eq!(names(root.path()), vec![current]);
1189 }
1190
1191 #[test]
1194 fn a_link_in_place_of_a_created_work_folder_is_not_followed() {
1195 let records = tempfile::tempdir().unwrap();
1196 let workdirs = tempfile::tempdir().unwrap();
1197 let elsewhere = tempfile::tempdir().unwrap();
1198 std::fs::write(elsewhere.path().join("keep"), b"x").unwrap();
1199 let all: Vec<String> = (1..=2).map(|n| ses(NOW - (2 - n) * 1000, n)).collect();
1200 make_sessions(records.path(), &all);
1201 std::os::unix::fs::symlink(elsewhere.path(), workdirs.path().join(&all[0])).unwrap();
1202
1203 let pruned = prune_sessions(
1204 &crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
1205 Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
1206 &all[1],
1207 &TraceRetainConfig {
1208 max_sessions: Some(1),
1209 max_age_secs: None,
1210 },
1211 NOW,
1212 );
1213
1214 assert_eq!(pruned.len(), 1);
1215 assert!(elsewhere.path().join("keep").exists());
1216 }
1217
1218 #[test]
1221 fn an_empty_policy_removes_nothing() {
1222 let root = tempfile::tempdir().unwrap();
1223 let all = [ses(NOW - 9_000_000, 1), ses(NOW, 2)];
1224 make_sessions(root.path(), &all);
1225 assert!(prune_sessions(
1226 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1227 None,
1228 &all[1],
1229 &TraceRetainConfig {
1230 max_sessions: None,
1231 max_age_secs: None
1232 },
1233 NOW
1234 )
1235 .is_empty());
1236 assert_eq!(names(root.path()), all.to_vec());
1237 }
1238
1239 fn write_record(dir: &Path, capsule: Option<&str>, count: u64) -> PathBuf {
1243 std::fs::create_dir_all(dir).unwrap();
1244 let path = dir.join(RECORD_FILE_NAME);
1245 let mut contents = String::new();
1246 if let Some(capsule) = capsule {
1247 contents.push_str(
1248 &header_line(&RecordHeader {
1249 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
1250 capsule: capsule.to_string(),
1251 created_ms: NOW,
1252 truncated: None,
1253 })
1254 .unwrap(),
1255 );
1256 }
1257 for i in 1..=count {
1258 contents.push_str(&message_line(
1259 &msg(NOW - (count - i) * 1000, i),
1260 &format!("m{i}"),
1261 ));
1262 contents.push('\n');
1263 }
1264 std::fs::write(&path, contents).unwrap();
1265 path
1266 }
1267
1268 #[test]
1271 fn max_messages_leaves_the_newest_n_with_their_ids_intact() {
1272 let dir = tempfile::tempdir().unwrap();
1273 let path = write_record(dir.path(), Some("capsule"), 10);
1274 let before: Vec<String> = std::fs::read_to_string(&path)
1275 .unwrap()
1276 .lines()
1277 .filter_map(parse_message_line)
1278 .map(|m| m["id"].as_str().unwrap().to_string())
1279 .collect();
1280
1281 let outcome = truncate_record(&path, 3, "capsule").unwrap();
1282 assert_eq!(outcome.dropped, 7);
1283 assert_eq!(outcome.kept, 3);
1284 assert_eq!(outcome.last_dropped_id, before[6]);
1285 assert_eq!(outcome.oldest_surviving_id, before[7]);
1286
1287 let raw = std::fs::read_to_string(&path).unwrap();
1288 let lines: Vec<&str> = raw.lines().collect();
1289 assert_eq!(
1290 lines.len(),
1291 4,
1292 "a header plus exactly three messages: {lines:?}"
1293 );
1294 for line in &lines {
1295 serde_json::from_str::<serde_json::Value>(line).expect("every line is JSON");
1296 }
1297 let header = parse_header(lines[0]).expect("the first line is the header");
1298 assert_eq!(header.capsule, "capsule");
1299 let marker = header.truncated.expect("the header records the drop");
1300 assert_eq!(marker.dropped, 7);
1301 assert_eq!(marker.last_dropped_id, before[6]);
1302 assert_eq!(marker.oldest_surviving_id, before[7]);
1303
1304 let after: Vec<String> = lines[1..]
1305 .iter()
1306 .map(|line| {
1307 parse_message_line(line).expect("a message with a string role")["id"]
1308 .as_str()
1309 .unwrap()
1310 .to_string()
1311 })
1312 .collect();
1313 assert_eq!(after, before[7..], "surviving ids are the bytes they were");
1314 }
1315
1316 #[test]
1319 fn a_header_line_is_not_counted_as_a_message() {
1320 let dir = tempfile::tempdir().unwrap();
1321 let headerless = write_record(&dir.path().join("a"), None, 4);
1322 let owned = write_record(&dir.path().join("b"), Some("capsule"), 4);
1323 assert_eq!(count_messages(&headerless), 4);
1324 assert_eq!(count_messages(&owned), 4);
1325 }
1326
1327 #[test]
1330 fn a_second_truncation_accumulates_the_dropped_count() {
1331 let dir = tempfile::tempdir().unwrap();
1332 let path = write_record(dir.path(), Some("capsule"), 10);
1333 truncate_record(&path, 6, "capsule").unwrap();
1334 truncate_record(&path, 2, "capsule").unwrap();
1335 let marker = read_header(&path).unwrap().truncated.unwrap();
1336 assert_eq!(marker.dropped, 8);
1337 assert_eq!(count_messages(&path), 2);
1338 }
1339
1340 #[test]
1341 fn truncation_rewrite_keeps_the_record_owner_only() {
1342 use std::os::unix::fs::PermissionsExt;
1343 let dir = tempfile::tempdir().unwrap();
1344 let path = write_record(dir.path(), Some("capsule"), 5);
1345 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
1346
1347 truncate_record(&path, 2, "capsule").unwrap();
1348
1349 assert_eq!(count_messages(&path), 2);
1350 assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
1351 }
1352
1353 #[test]
1355 fn a_record_inside_the_limit_is_not_rewritten() {
1356 let dir = tempfile::tempdir().unwrap();
1357 let path = write_record(dir.path(), Some("capsule"), 3);
1358 let before = std::fs::read(&path).unwrap();
1359 let outcome = truncate_record(&path, 3, "capsule").unwrap();
1360 assert_eq!(outcome.dropped, 0);
1361 assert_eq!(std::fs::read(&path).unwrap(), before);
1362 }
1363
1364 #[test]
1368 fn a_staged_truncation_leaves_the_original_intact_until_the_rename() {
1369 let dir = tempfile::tempdir().unwrap();
1370 let path = write_record(dir.path(), Some("capsule"), 10);
1371 let before = std::fs::read(&path).unwrap();
1372
1373 let (staged, outcome) = stage_truncation(&path, 3, "capsule", NOW).unwrap().unwrap();
1374 assert_eq!(outcome.dropped, 7);
1375
1376 let temp = staged.temp_path().to_path_buf();
1377 assert_eq!(
1378 temp.parent(),
1379 path.parent(),
1380 "the temp file is in the record's own directory, so the rename is atomic"
1381 );
1382 assert_ne!(temp.file_name().unwrap(), RECORD_FILE_NAME);
1383
1384 assert_eq!(
1385 std::fs::read(&path).unwrap(),
1386 before,
1387 "the original is byte-identical before the rename"
1388 );
1389 assert_eq!(count_messages(&path), 10);
1390 for line in std::fs::read_to_string(&path).unwrap().lines() {
1391 serde_json::from_str::<serde_json::Value>(line).expect("the original parses whole");
1392 }
1393
1394 drop(staged);
1396 assert!(!temp.exists());
1397 assert_eq!(std::fs::read(&path).unwrap(), before);
1398 assert_eq!(count_messages(&path), 10);
1399 }
1400
1401 #[test]
1407 fn record_pruning_skips_another_capsules_records_and_every_unowned_one() {
1408 let root = tempfile::tempdir().unwrap();
1409 write_record(&root.path().join("mine"), Some("mine"), 3);
1410 write_record(&root.path().join("theirs"), Some("theirs"), 3);
1411 write_record(&root.path().join("unowned"), None, 3);
1412 write_record(&root.path().join("current"), Some("mine"), 3);
1413
1414 let far_future = now_ms() + 3_600_000;
1416 let pruned = prune_records(
1417 root.path(),
1418 "mine",
1419 Some("current"),
1420 &ContextRetainConfig {
1421 max_messages: None,
1422 max_age_secs: Some(1),
1423 },
1424 far_future,
1425 );
1426
1427 assert_eq!(pruned.len(), 1, "{pruned:?}");
1428 assert_eq!(pruned[0].context_id, "mine");
1429 assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
1430 assert_eq!(names(root.path()), vec!["current", "theirs", "unowned"]);
1431 }
1432
1433 #[test]
1436 fn a_record_written_inside_the_window_is_kept() {
1437 let root = tempfile::tempdir().unwrap();
1438 write_record(&root.path().join("fresh"), Some("mine"), 3);
1439 let pruned = prune_records(
1440 root.path(),
1441 "mine",
1442 None,
1443 &ContextRetainConfig {
1444 max_messages: None,
1445 max_age_secs: Some(90 * 86_400),
1446 },
1447 now_ms(),
1448 );
1449 assert!(pruned.is_empty());
1450 assert_eq!(names(root.path()), vec!["fresh"]);
1451 }
1452
1453 #[test]
1456 fn max_messages_alone_removes_no_record() {
1457 let root = tempfile::tempdir().unwrap();
1458 write_record(&root.path().join("ctx"), Some("mine"), 100);
1459 assert!(prune_records(
1460 root.path(),
1461 "mine",
1462 None,
1463 &ContextRetainConfig {
1464 max_messages: Some(1),
1465 max_age_secs: None
1466 },
1467 now_ms() + 3_600_000,
1468 )
1469 .is_empty());
1470 assert_eq!(names(root.path()), vec!["ctx"]);
1471 }
1472
1473 fn frm(ms: u64, tail: u64) -> FormationId {
1477 FormationId::parse(&format!("frm_{ms:012x}{tail:020x}")).unwrap()
1478 }
1479
1480 fn retain(max_sessions: Option<u32>, max_age_secs: Option<u64>) -> TraceRetainConfig {
1481 TraceRetainConfig {
1482 max_sessions,
1483 max_age_secs,
1484 }
1485 }
1486
1487 fn make_formation(root: &Path, id: &FormationId, marker: Option<&str>) -> PathBuf {
1490 let dir = root.join(id.as_str());
1491 let session = dir.join("worker/.murmur").join(ses(NOW, 1));
1492 std::fs::create_dir_all(&session).unwrap();
1493 std::fs::write(session.join("trace.jsonl"), "{}\n").unwrap();
1494 if let Some(marker) = marker {
1495 std::fs::write(
1496 dir.join(crate::formation_launch::FORMATION_MARKER_FILE),
1497 marker,
1498 )
1499 .unwrap();
1500 }
1501 dir
1502 }
1503
1504 fn marker_naming(project: &Path) -> String {
1505 serde_json::to_string(&FormationMarker {
1506 project_dir: project.to_path_buf(),
1507 launcher_pid: 1,
1508 launcher_start: String::new(),
1509 })
1510 .unwrap()
1511 }
1512
1513 fn make_owned(root: &Path, project: &Path, ids: &[&FormationId]) {
1515 for id in ids {
1516 make_formation(root, id, Some(&marker_naming(project)));
1517 }
1518 }
1519
1520 fn removed(pruned: &[PrunedFormation]) -> Vec<(FormationId, &'static str)> {
1521 pruned
1522 .iter()
1523 .map(|p| (p.formation_id.clone(), p.reason))
1524 .collect()
1525 }
1526
1527 const DAY_MS: u64 = 86_400_000;
1528
1529 #[test]
1532 fn prune_formations_never_removes_the_current_or_a_later_formation() {
1533 let root = tempfile::tempdir().unwrap();
1534 let project = tempfile::tempdir().unwrap();
1535 let old = frm(NOW - 3 * DAY_MS, 1);
1536 let current = frm(NOW - 2 * DAY_MS, 2);
1537 let later = frm(NOW - 2 * DAY_MS + 1000, 3);
1538 make_owned(root.path(), project.path(), &[&old, ¤t, &later]);
1539
1540 let pruned = prune_formations(
1541 root.path(),
1542 ¤t,
1543 project.path(),
1544 &retain(Some(1), Some(1)),
1545 NOW,
1546 |_, _| false,
1547 );
1548
1549 assert_eq!(
1550 removed(&pruned),
1551 [(old.clone(), crate::trace::RETENTION_REASON_MAX_AGE)]
1552 );
1553 assert_eq!(pruned[0].path, root.path().join(old.as_str()));
1554 assert_eq!(
1555 names(root.path()),
1556 vec![current.as_str().to_string(), later.as_str().to_string()]
1557 );
1558 }
1559
1560 #[test]
1563 fn prune_formations_keeps_the_newest_n_with_the_current_counted() {
1564 let root = tempfile::tempdir().unwrap();
1565 let project = tempfile::tempdir().unwrap();
1566 let all: Vec<FormationId> = (1..=5).map(|n| frm(NOW - (5 - n) * 1000, n)).collect();
1567 make_owned(root.path(), project.path(), &all.iter().collect::<Vec<_>>());
1568
1569 let pruned = prune_formations(
1570 root.path(),
1571 &all[3],
1572 project.path(),
1573 &retain(Some(3), None),
1574 NOW,
1575 |_, _| false,
1576 );
1577
1578 let reason = crate::trace::RETENTION_REASON_MAX_SESSIONS;
1579 assert_eq!(
1580 removed(&pruned),
1581 [(all[1].clone(), reason), (all[0].clone(), reason)]
1582 );
1583 assert_eq!(
1584 names(root.path()),
1585 all[2..]
1586 .iter()
1587 .map(|id| id.as_str().to_string())
1588 .collect::<Vec<_>>()
1589 );
1590 }
1591
1592 #[test]
1595 fn prune_formations_reads_age_from_the_id() {
1596 let root = tempfile::tempdir().unwrap();
1597 let project = tempfile::tempdir().unwrap();
1598 let old_a = frm(NOW - 7_200_000, 1);
1599 let old_b = frm(NOW - 7_100_000, 2);
1600 let recent = frm(NOW - 60_000, 3);
1601 let current = frm(NOW, 4);
1602 make_owned(
1603 root.path(),
1604 project.path(),
1605 &[&old_a, &old_b, &recent, ¤t],
1606 );
1607
1608 let pruned = prune_formations(
1609 root.path(),
1610 ¤t,
1611 project.path(),
1612 &retain(None, Some(900)),
1613 NOW,
1614 |_, _| false,
1615 );
1616
1617 let reason = crate::trace::RETENTION_REASON_MAX_AGE;
1618 assert_eq!(removed(&pruned), [(old_b, reason), (old_a, reason)]);
1619 assert_eq!(
1620 names(root.path()),
1621 vec![recent.as_str().to_string(), current.as_str().to_string()]
1622 );
1623 }
1624
1625 #[test]
1628 fn prune_formations_attributes_reasons_age_first() {
1629 let root = tempfile::tempdir().unwrap();
1630 let project = tempfile::tempdir().unwrap();
1631 let old = frm(NOW - 7_200_000, 1);
1632 let recent = frm(NOW - 60_000, 2);
1633 let current = frm(NOW, 3);
1634 make_owned(root.path(), project.path(), &[&old, &recent, ¤t]);
1635
1636 let pruned = prune_formations(
1637 root.path(),
1638 ¤t,
1639 project.path(),
1640 &retain(Some(1), Some(900)),
1641 NOW,
1642 |_, _| false,
1643 );
1644
1645 assert_eq!(
1646 removed(&pruned),
1647 [
1648 (recent, crate::trace::RETENTION_REASON_MAX_SESSIONS),
1649 (old, crate::trace::RETENTION_REASON_MAX_AGE),
1650 ]
1651 );
1652 }
1653
1654 #[test]
1657 fn prune_formations_skips_a_live_directory_and_keeps_its_rank() {
1658 let root = tempfile::tempdir().unwrap();
1659 let project = tempfile::tempdir().unwrap();
1660 let older = frm(NOW - 3000, 1);
1661 let live = frm(NOW - 2000, 2);
1662 let current = frm(NOW, 3);
1663 make_owned(root.path(), project.path(), &[&older, &live, ¤t]);
1664
1665 let asked = std::cell::RefCell::new(Vec::new());
1666 let pruned = prune_formations(
1667 root.path(),
1668 ¤t,
1669 project.path(),
1670 &retain(Some(1), None),
1671 NOW,
1672 |id, marker| {
1673 assert_eq!(marker.launcher_pid, 1);
1674 asked.borrow_mut().push(id.clone());
1675 *id == live
1676 },
1677 );
1678
1679 assert_eq!(
1680 removed(&pruned),
1681 [(older.clone(), crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1682 );
1683 assert_eq!(*asked.borrow(), [live.clone(), older]);
1684 assert_eq!(
1685 names(root.path()),
1686 vec![live.as_str().to_string(), current.as_str().to_string()]
1687 );
1688
1689 let kept = frm(NOW - 2500, 4);
1691 make_owned(root.path(), project.path(), &[&kept]);
1692 let pruned = prune_formations(
1693 root.path(),
1694 ¤t,
1695 project.path(),
1696 &retain(Some(3), None),
1697 NOW,
1698 |id, _| *id == live,
1699 );
1700 assert!(pruned.is_empty(), "{pruned:?}");
1701 let pruned = prune_formations(
1702 root.path(),
1703 ¤t,
1704 project.path(),
1705 &retain(Some(2), None),
1706 NOW,
1707 |id, _| *id == live,
1708 );
1709 assert_eq!(
1710 removed(&pruned),
1711 [(kept, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1712 );
1713 }
1714
1715 #[test]
1718 fn prune_formations_never_ranks_or_removes_unowned_or_foreign_directories() {
1719 let root = tempfile::tempdir().unwrap();
1720 let project = tempfile::tempdir().unwrap();
1721 let other = tempfile::tempdir().unwrap();
1722 let owned_old = frm(NOW - 3 * DAY_MS, 1);
1723 let owned_kept = frm(NOW - 2 * DAY_MS, 2);
1724 let unmarked = frm(NOW - 2 * DAY_MS + 1, 3);
1725 let garbled = frm(NOW - 2 * DAY_MS + 2, 4);
1726 let foreign = frm(NOW - 2 * DAY_MS + 3, 5);
1727 let current = frm(NOW, 6);
1728 make_owned(
1729 root.path(),
1730 project.path(),
1731 &[&owned_old, &owned_kept, ¤t],
1732 );
1733 make_formation(root.path(), &unmarked, None);
1734 make_formation(root.path(), &garbled, Some("not json {"));
1735 make_formation(root.path(), &foreign, Some(&marker_naming(other.path())));
1736
1737 let pruned = prune_formations(
1738 root.path(),
1739 ¤t,
1740 project.path(),
1741 &retain(Some(2), None),
1742 NOW,
1743 |_, _| false,
1744 );
1745 assert_eq!(
1746 removed(&pruned),
1747 [(owned_old, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1748 );
1749
1750 let pruned = prune_formations(
1751 root.path(),
1752 ¤t,
1753 project.path(),
1754 &retain(Some(1), Some(DAY_MS / 1000)),
1755 NOW,
1756 |_, _| false,
1757 );
1758 assert_eq!(
1759 removed(&pruned),
1760 [(owned_kept, crate::trace::RETENTION_REASON_MAX_AGE)]
1761 );
1762 for id in [&unmarked, &garbled, &foreign, ¤t] {
1763 assert!(root.path().join(id.as_str()).is_dir(), "{id}");
1764 }
1765 }
1766
1767 #[test]
1770 fn prune_formations_ignores_a_symlink_a_regular_file_and_a_name_that_is_not_an_id() {
1771 let root = tempfile::tempdir().unwrap();
1772 let elsewhere = tempfile::tempdir().unwrap();
1773 let project = tempfile::tempdir().unwrap();
1774 let marker = marker_naming(project.path());
1775 let target = make_formation(elsewhere.path(), &frm(NOW - 3 * DAY_MS, 1), Some(&marker));
1776 let linked = frm(NOW - 3 * DAY_MS, 2);
1777 std::os::unix::fs::symlink(&target, root.path().join(linked.as_str())).unwrap();
1778 std::fs::write(root.path().join(frm(NOW - 3 * DAY_MS, 3).as_str()), &marker).unwrap();
1779 for name in ["not-a-formation", "frm_ABC", "frm_0123"] {
1780 std::fs::create_dir(root.path().join(name)).unwrap();
1781 std::fs::write(
1782 root.path()
1783 .join(name)
1784 .join(crate::formation_launch::FORMATION_MARKER_FILE),
1785 &marker,
1786 )
1787 .unwrap();
1788 }
1789 let before = names(root.path());
1790
1791 let pruned = prune_formations(
1792 root.path(),
1793 &frm(NOW, 9),
1794 project.path(),
1795 &retain(Some(1), Some(1)),
1796 NOW,
1797 |_, _| false,
1798 );
1799
1800 assert!(pruned.is_empty(), "{pruned:?}");
1801 assert_eq!(names(root.path()), before);
1802 assert!(target.join("worker/.murmur").is_dir());
1803 }
1804
1805 #[test]
1807 fn prune_formations_with_neither_key_or_no_directory_removes_nothing() {
1808 let root = tempfile::tempdir().unwrap();
1809 let project = tempfile::tempdir().unwrap();
1810 let old = frm(NOW - 3 * DAY_MS, 1);
1811 make_owned(root.path(), project.path(), &[&old]);
1812
1813 let pruned = prune_formations(
1814 root.path(),
1815 &frm(NOW, 2),
1816 project.path(),
1817 &retain(None, None),
1818 NOW,
1819 |_, _| panic!("nothing is a candidate without a policy"),
1820 );
1821 assert!(pruned.is_empty());
1822 assert!(root.path().join(old.as_str()).is_dir());
1823
1824 assert!(prune_formations(
1825 &root.path().join("missing"),
1826 &frm(NOW, 2),
1827 project.path(),
1828 &retain(Some(1), Some(1)),
1829 NOW,
1830 |_, _| false,
1831 )
1832 .is_empty());
1833 }
1834
1835 #[test]
1838 fn prune_formations_compares_project_directories_canonically() {
1839 let root = tempfile::tempdir().unwrap();
1840 let project = tempfile::tempdir().unwrap();
1841 let links = tempfile::tempdir().unwrap();
1842 let link = links.path().join("project");
1843 std::os::unix::fs::symlink(project.path(), &link).unwrap();
1844 let old = frm(NOW - 3 * DAY_MS, 1);
1845 make_owned(
1846 root.path(),
1847 &std::fs::canonicalize(project.path()).unwrap(),
1848 &[&old],
1849 );
1850
1851 let pruned = prune_formations(
1852 root.path(),
1853 &frm(NOW, 2),
1854 &link,
1855 &retain(None, Some(1)),
1856 NOW,
1857 |_, _| false,
1858 );
1859 assert_eq!(
1860 removed(&pruned),
1861 [(old, crate::trace::RETENTION_REASON_MAX_AGE)]
1862 );
1863 }
1864
1865 #[test]
1867 fn prune_formations_leaves_a_directory_it_could_not_remove_out_of_the_result() {
1868 use std::os::unix::fs::PermissionsExt;
1869
1870 let root = tempfile::tempdir().unwrap();
1871 let project = tempfile::tempdir().unwrap();
1872 let stuck = frm(NOW - 3 * DAY_MS, 1);
1873 let gone = frm(NOW - 3 * DAY_MS + 1, 2);
1874 make_owned(root.path(), project.path(), &[&stuck, &gone]);
1875 let stuck_dir = root.path().join(stuck.as_str());
1876 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o500)).unwrap();
1877 if std::fs::write(stuck_dir.join("probe"), b"").is_ok() {
1879 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
1880 return;
1881 }
1882
1883 let pruned = prune_formations(
1884 root.path(),
1885 &frm(NOW, 3),
1886 project.path(),
1887 &retain(None, Some(1)),
1888 NOW,
1889 |_, _| false,
1890 );
1891 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
1892
1893 assert_eq!(
1894 removed(&pruned),
1895 [(gone, crate::trace::RETENTION_REASON_MAX_AGE)]
1896 );
1897 assert!(stuck_dir.is_dir());
1898 }
1899
1900 fn harness_entry_json(kind: &str, session_id: &str) -> String {
1903 json!({"type": kind, "session_id": session_id, "harness": "h", "driver": "d",
1904 "created_ms": 1, "updated_ms": 2})
1905 .to_string()
1906 }
1907
1908 #[test]
1911 fn the_context_finder_sees_either_file_in_every_store() {
1912 let root = tempfile::tempdir().unwrap();
1913 let dir = |store: &str, context: &str| {
1914 let path = root.path().join(store).join(context);
1915 std::fs::create_dir_all(&path).unwrap();
1916 path
1917 };
1918 std::fs::write(dir("b", "ctx").join(RECORD_FILE_NAME), "").unwrap();
1919 std::fs::write(
1920 dir("a", "ctx").join(HARNESS_SESSION_FILE_NAME),
1921 harness_entry_json(crate::harness_session::HARNESS_SESSION_TYPE, "s"),
1922 )
1923 .unwrap();
1924 dir("c", "ctx");
1925 std::fs::write(dir(".hidden", "ctx").join(RECORD_FILE_NAME), "").unwrap();
1926 std::fs::write(dir("a", ".ctx").join(RECORD_FILE_NAME), "").unwrap();
1927
1928 let found = find_context_dirs_in(root.path(), "ctx");
1929 let stores: Vec<&str> = found.iter().map(|dir| dir.record.as_str()).collect();
1930 assert_eq!(stores, ["a", "b"]);
1931 assert!(found.iter().all(|dir| dir.context_id == "ctx"));
1932 assert_eq!(found[1].path, root.path().join("b").join("ctx"));
1933 assert!(find_context_dirs_in(root.path(), ".ctx").is_empty());
1934 assert!(find_context_dirs_in(root.path(), "other").is_empty());
1935 assert!(find_context_dirs_in(&root.path().join("absent"), "ctx").is_empty());
1936 }
1937
1938 #[test]
1941 fn the_context_reader_reads_as_the_runtime_reads() {
1942 let dir = tempfile::tempdir().unwrap();
1943 let header = RecordHeader {
1944 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
1945 capsule: "owner".to_string(),
1946 created_ms: 7,
1947 truncated: None,
1948 };
1949 let first = json!({"role": "user", "content": "hi", "id": "msg_1", "kept": {"x": 1}});
1950 let second = json!({"role": "assistant", "content": [{"type": "text", "text": "yo"}]});
1951 std::fs::write(
1952 dir.path().join(RECORD_FILE_NAME),
1953 format!(
1954 "{}{first}\n\n{second}\n{}\n{{\"role\": \"us",
1955 header_line(&header).unwrap(),
1956 header_line(&header).unwrap().trim_end()
1957 ),
1958 )
1959 .unwrap();
1960
1961 let contents = read_context_dir(dir.path()).unwrap();
1962 assert_eq!(contents.header, Some(header.clone()));
1963 assert_eq!(contents.messages, vec![first.clone(), second]);
1964 assert_eq!(contents.harness_session, None);
1965
1966 std::fs::write(
1967 dir.path().join(RECORD_FILE_NAME),
1968 format!("{first}\n{}", header_line(&header).unwrap()),
1969 )
1970 .unwrap();
1971 let contents = read_context_dir(dir.path()).unwrap();
1972 assert_eq!(contents.header, None, "a header below line 1 is no header");
1973 assert_eq!(contents.messages, vec![first]);
1974
1975 let empty = tempfile::tempdir().unwrap();
1976 let contents = read_context_dir(empty.path()).unwrap();
1977 assert_eq!(contents.header, None);
1978 assert!(contents.messages.is_empty());
1979
1980 std::fs::create_dir(empty.path().join(RECORD_FILE_NAME)).unwrap();
1981 assert!(
1982 read_context_dir(empty.path()).is_err(),
1983 "an unreadable record"
1984 );
1985 }
1986
1987 #[test]
1990 fn the_context_reader_reads_no_session_where_the_runtime_does() {
1991 let dir = tempfile::tempdir().unwrap();
1992 let file = dir.path().join(HARNESS_SESSION_FILE_NAME);
1993 assert_eq!(read_context_dir(dir.path()).unwrap().harness_session, None);
1994
1995 let session_type = crate::harness_session::HARNESS_SESSION_TYPE;
1996 for (body, expected) in [
1997 ("{not json".to_string(), false),
1998 (harness_entry_json("another-tool", "s"), false),
1999 (harness_entry_json(session_type, ""), false),
2000 (harness_entry_json(session_type, "s"), true),
2001 ] {
2002 std::fs::write(&file, &body).unwrap();
2003 let session = read_context_dir(dir.path()).unwrap().harness_session;
2004 assert_eq!(session.is_some(), expected, "{body}");
2005 if let Some(session) = session {
2006 assert_eq!(
2007 (session.session_id.as_str(), session.harness.as_str()),
2008 ("s", "h")
2009 );
2010 assert_eq!((session.created_ms, session.updated_ms), (1, 2));
2011 }
2012 }
2013 }
2014}