1use std::path::{Path, PathBuf};
35
36use crate::conversation::{
37 header_line, message_id_timestamp_ms, parse_message_line, read_header, CONVERSATION_ROOT_DIR,
38 RECORD_FILE_NAME,
39};
40
41use crate::formation::FormationId;
42use crate::formation_launch::FormationMarker;
43
44pub use crate::conversation::{RecordHeader, TruncationMarker};
45
46#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct PrunedSession {
49 pub name: String,
51 pub reason: &'static str,
54}
55
56#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct PrunedFormation {
59 pub formation_id: FormationId,
61 pub path: PathBuf,
63 pub reason: &'static str,
66}
67
68#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct PrunedRecord {
71 pub context_id: String,
73 pub path: PathBuf,
75 pub reason: &'static str,
77}
78
79#[derive(Debug, Clone, PartialEq, Eq)]
81pub struct TruncationOutcome {
82 pub dropped: u64,
84 pub kept: u64,
86 pub oldest_surviving_id: String,
88 pub last_dropped_id: String,
90}
91
92#[derive(Debug, Clone, PartialEq, Eq)]
94pub enum MessageStatus {
95 Present { position: u64, total: u64 },
97 Truncated {
101 dropped: u64,
102 oldest_surviving_id: String,
103 },
104 Unknown,
106}
107
108#[derive(Debug, Clone, PartialEq, Eq)]
110pub struct MessageLocation {
111 pub record: String,
112 pub context_id: String,
113 pub path: PathBuf,
114 pub status: MessageStatus,
115}
116
117#[derive(Debug, Clone, PartialEq, Eq)]
119pub struct RecordSummary {
120 pub record: String,
122 pub context_id: String,
124 pub path: PathBuf,
125 pub messages: u64,
127 pub bytes: u64,
129 pub last_touched_ms: Option<u64>,
132 pub capsule: Option<String>,
135 pub truncation: Option<TruncationMarker>,
137}
138
139#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct RemovedRecord {
142 pub record: String,
143 pub context_id: String,
144 pub path: PathBuf,
146 pub messages: u64,
148}
149
150pub fn session_id_timestamp_ms(name: &str) -> Option<u64> {
158 uuid_v7_timestamp_ms(name, "ses_")
159}
160
161pub(crate) fn uuid_v7_timestamp_ms(id: &str, prefix: &str) -> Option<u64> {
164 let hex = id.strip_prefix(prefix)?;
165 u64::from_str_radix(hex.get(..12)?, 16).ok()
166}
167
168pub fn prune_sessions(
183 sessions_root: &Path,
184 current_session_id: &str,
185 policy: &murmur_artifact::TraceRetainConfig,
186 now_ms: u64,
187) -> Vec<PrunedSession> {
188 if policy.max_sessions.is_none() && policy.max_age_secs.is_none() {
189 return Vec::new();
190 }
191
192 let mut entries: Vec<String> = match std::fs::read_dir(sessions_root) {
193 Ok(dir) => dir
194 .filter_map(Result::ok)
195 .filter(|entry| entry.path().is_dir())
196 .map(|entry| entry.file_name().to_string_lossy().into_owned())
197 .filter(|name| session_id_timestamp_ms(name).is_some())
198 .collect(),
199 Err(_) => return Vec::new(),
200 };
201 entries.sort_unstable_by(|a, b| b.cmp(a));
203
204 let age_floor_ms = policy
205 .max_age_secs
206 .map(|secs| now_ms.saturating_sub(secs.saturating_mul(1000)));
207
208 let mut pruned = Vec::new();
209 for (rank, name) in entries.iter().enumerate() {
210 if name.as_str() >= current_session_id {
213 continue;
214 }
215 let over_age = age_floor_ms.is_some_and(|floor| {
216 session_id_timestamp_ms(name).is_some_and(|minted| minted < floor)
217 });
218 let over_count = policy.max_sessions.is_some_and(|max| rank >= max as usize);
219 if !over_age && !over_count {
220 continue;
221 }
222 let reason = if over_age {
223 crate::trace::RETENTION_REASON_MAX_AGE
224 } else {
225 crate::trace::RETENTION_REASON_MAX_SESSIONS
226 };
227 if std::fs::remove_dir_all(sessions_root.join(name)).is_ok() {
228 pruned.push(PrunedSession {
229 name: name.clone(),
230 reason,
231 });
232 }
233 }
234 pruned
235}
236
237pub fn prune_formations(
262 formations_dir: &Path,
263 current: &FormationId,
264 project_dir: &Path,
265 policy: &murmur_artifact::TraceRetainConfig,
266 now_ms: u64,
267 is_live: impl Fn(&FormationId, &FormationMarker) -> bool,
268) -> Vec<PrunedFormation> {
269 if policy.max_sessions.is_none() && policy.max_age_secs.is_none() {
270 return Vec::new();
271 }
272 let Ok(entries) = std::fs::read_dir(formations_dir) else {
273 return Vec::new();
274 };
275 let project_dir = crate::formation_launch::canonical_project_dir(project_dir);
276 let mut owned: Vec<(FormationId, FormationMarker)> = entries
277 .filter_map(Result::ok)
278 .filter_map(|entry| {
279 let id = FormationId::parse(entry.file_name().to_str()?).ok()?;
280 let path = entry.path();
281 if !std::fs::symlink_metadata(&path).ok()?.is_dir() {
282 return None;
283 }
284 let marker = FormationMarker::read(&path)?;
285 (crate::formation_launch::canonical_project_dir(&marker.project_dir) == project_dir)
286 .then_some((id, marker))
287 })
288 .collect();
289 owned.sort_unstable_by(|(a, _), (b, _)| b.as_str().cmp(a.as_str()));
291
292 let age_floor_ms = policy
293 .max_age_secs
294 .map(|secs| now_ms.saturating_sub(secs.saturating_mul(1000)));
295
296 let mut pruned = Vec::new();
297 for (rank, (id, marker)) in owned.iter().enumerate() {
298 if id.as_str() >= current.as_str() {
299 continue;
300 }
301 let over_age = age_floor_ms.is_some_and(|floor| id.minted_at_ms() < floor);
302 let over_count = policy.max_sessions.is_some_and(|max| rank >= max as usize);
303 if !over_age && !over_count {
304 continue;
305 }
306 if is_live(id, marker) {
307 continue;
308 }
309 let reason = if over_age {
310 crate::trace::RETENTION_REASON_MAX_AGE
311 } else {
312 crate::trace::RETENTION_REASON_MAX_SESSIONS
313 };
314 let path = formations_dir.join(id.as_str());
315 if std::fs::remove_dir_all(&path).is_ok() {
316 pruned.push(PrunedFormation {
317 formation_id: id.clone(),
318 path,
319 reason,
320 });
321 }
322 }
323 pruned
324}
325
326pub fn prune_records(
345 record_root: &Path,
346 capsule_name: &str,
347 current_context_id: Option<&str>,
348 policy: &murmur_artifact::ContextRetainConfig,
349 now_ms: u64,
350) -> Vec<PrunedRecord> {
351 let Some(max_age_secs) = policy.max_age_secs else {
352 return Vec::new();
353 };
354 let floor_ms = now_ms.saturating_sub(max_age_secs.saturating_mul(1000));
355
356 let mut pruned = Vec::new();
357 for (context_id, dir) in context_dirs(record_root) {
358 if Some(context_id.as_str()) == current_context_id {
359 continue;
360 }
361 let file = dir.join(RECORD_FILE_NAME);
362 match read_header(&file) {
363 Some(header) if header.capsule == capsule_name => {}
364 _ => continue,
365 }
366 let Some(touched_ms) = last_touched_ms(&file) else {
367 continue;
368 };
369 if touched_ms >= floor_ms {
370 continue;
371 }
372 if std::fs::remove_dir_all(&dir).is_ok() {
373 pruned.push(PrunedRecord {
374 context_id,
375 path: dir,
376 reason: crate::trace::RETENTION_REASON_MAX_AGE,
377 });
378 }
379 }
380 pruned
381}
382
383pub fn truncate_record(path: &Path, keep: u32, capsule: &str) -> Result<TruncationOutcome, String> {
398 match stage_truncation(path, keep, capsule, now_ms())? {
399 None => Ok(TruncationOutcome {
400 dropped: 0,
401 kept: count_messages(path),
402 oldest_surviving_id: String::new(),
403 last_dropped_id: String::new(),
404 }),
405 Some((staged, outcome)) => {
406 staged.commit()?;
407 Ok(outcome)
408 }
409 }
410}
411
412pub(crate) fn stage_truncation(
418 path: &Path,
419 keep: u32,
420 capsule: &str,
421 at_ms: u64,
422) -> Result<Option<(StagedRewrite, TruncationOutcome)>, String> {
423 let raw = match std::fs::read_to_string(path) {
424 Ok(raw) => raw,
425 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
426 Err(err) => return Err(format!("{}: {err}", path.display())),
427 };
428
429 let existing = read_header(path);
430 let lines: Vec<&str> = raw.lines().collect();
431 let message_indices: Vec<usize> = lines
434 .iter()
435 .enumerate()
436 .filter(|(_, line)| parse_message_line(line).is_some())
437 .map(|(index, _)| index)
438 .collect();
439
440 let total = message_indices.len();
441 let keep = keep as usize;
442 if total <= keep {
443 return Ok(None);
444 }
445 let dropped = total - keep;
446 let first_kept_index = message_indices[dropped];
447
448 let id_of = |index: usize| -> String {
449 parse_message_line(lines[index])
450 .as_ref()
451 .and_then(|message| crate::agent::message_id(message).map(str::to_string))
452 .unwrap_or_default()
453 };
454 let last_dropped_id = id_of(message_indices[dropped - 1]);
455 let oldest_surviving_id = message_indices
456 .get(dropped)
457 .map(|i| id_of(*i))
458 .unwrap_or_default();
459
460 let header = RecordHeader {
461 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
462 capsule: existing
463 .as_ref()
464 .map(|h| h.capsule.clone())
465 .unwrap_or_else(|| capsule.to_string()),
466 created_ms: existing.as_ref().map(|h| h.created_ms).unwrap_or(at_ms),
467 truncated: Some(TruncationMarker {
468 dropped: existing
471 .as_ref()
472 .and_then(|h| h.truncated.as_ref())
473 .map(|marker| marker.dropped)
474 .unwrap_or(0)
475 + dropped as u64,
476 oldest_surviving_id: oldest_surviving_id.clone(),
477 last_dropped_id: last_dropped_id.clone(),
478 at_ms,
479 }),
480 };
481
482 let mut contents = header_line(&header)?;
483 for line in &lines[first_kept_index..] {
484 contents.push_str(line);
485 contents.push('\n');
486 }
487
488 let staged = StagedRewrite::stage_with_mode(
489 path,
490 contents.as_bytes(),
491 Some(crate::conversation::RECORD_FILE_MODE),
492 )?;
493 Ok(Some((
494 staged,
495 TruncationOutcome {
496 dropped: dropped as u64,
497 kept: keep as u64,
498 oldest_surviving_id,
499 last_dropped_id,
500 },
501 )))
502}
503
504pub(crate) struct StagedRewrite {
510 target: PathBuf,
511 temp: PathBuf,
512 committed: bool,
513}
514
515impl StagedRewrite {
516 pub(crate) fn stage_with_mode(
519 target: &Path,
520 contents: &[u8],
521 mode: Option<u32>,
522 ) -> Result<Self, String> {
523 use std::io::Write;
524 use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
525
526 let dir = target
527 .parent()
528 .ok_or_else(|| format!("{} has no parent directory", target.display()))?;
529 let name = target
530 .file_name()
531 .map(|name| name.to_string_lossy().into_owned())
532 .unwrap_or_else(|| "record".to_string());
533 let temp = dir.join(format!(
534 ".{name}.{}.{}.tmp",
535 std::process::id(),
536 uuid::Uuid::now_v7().simple()
537 ));
538 let mut options = std::fs::OpenOptions::new();
539 options.write(true).create(true).truncate(true);
540 if let Some(mode) = mode {
541 options.mode(mode);
542 }
543 let mut file = options
544 .open(&temp)
545 .map_err(|err| format!("{}: {err}", temp.display()))?;
546 file.write_all(contents)
549 .and_then(|()| file.sync_all())
550 .map_err(|err| format!("{}: {err}", temp.display()))?;
551 if let Some(mode) = mode {
555 std::fs::set_permissions(&temp, std::fs::Permissions::from_mode(mode))
556 .map_err(|err| format!("{}: {err}", temp.display()))?;
557 }
558 Ok(Self {
559 target: target.to_path_buf(),
560 temp,
561 committed: false,
562 })
563 }
564
565 #[cfg(test)]
566 pub(crate) fn temp_path(&self) -> &Path {
567 &self.temp
568 }
569
570 pub(crate) fn commit(mut self) -> Result<(), String> {
571 std::fs::rename(&self.temp, &self.target)
572 .map_err(|err| format!("{}: {err}", self.target.display()))?;
573 self.committed = true;
574 Ok(())
575 }
576}
577
578impl Drop for StagedRewrite {
579 fn drop(&mut self) {
580 if !self.committed {
581 let _ = std::fs::remove_file(&self.temp);
582 }
583 }
584}
585
586pub fn list_records() -> Result<Vec<RecordSummary>, String> {
594 let root = conversations_root()?;
595 let mut summaries = Vec::new();
596 for record in child_dirs(&root) {
597 let store = root.join(&record);
598 for (context_id, dir) in context_dirs(&store) {
599 let path = dir.join(RECORD_FILE_NAME);
600 if !path.exists() {
601 continue;
602 }
603 let header = read_header(&path);
604 summaries.push(RecordSummary {
605 record: record.clone(),
606 context_id,
607 messages: count_messages(&path),
608 bytes: std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0),
609 last_touched_ms: last_touched_ms(&path),
610 capsule: header.as_ref().map(|h| h.capsule.clone()),
611 truncation: header.and_then(|h| h.truncated),
612 path,
613 });
614 }
615 }
616 summaries.sort_by(|a, b| {
617 a.record
618 .cmp(&b.record)
619 .then_with(|| a.context_id.cmp(&b.context_id))
620 });
621 Ok(summaries)
622}
623
624pub fn remove_record(record: &str, context_id: &str) -> Result<RemovedRecord, String> {
629 let dir = conversations_root()?.join(record).join(context_id);
630 if !dir.is_dir() {
631 return Err(format!("{} is not a context directory", dir.display()));
632 }
633 let messages = count_messages(&dir.join(RECORD_FILE_NAME));
634 std::fs::remove_dir_all(&dir).map_err(|err| format!("{}: {err}", dir.display()))?;
635 Ok(RemovedRecord {
636 record: record.to_string(),
637 context_id: context_id.to_string(),
638 path: dir,
639 messages,
640 })
641}
642
643pub fn locate_message(message_id: &str) -> Result<Vec<MessageLocation>, String> {
655 let minted_ms = message_id_timestamp_ms(message_id);
656 let mut found = Vec::new();
657 for summary in list_records()? {
658 let raw = std::fs::read_to_string(&summary.path).unwrap_or_default();
659 let mut position = 0u64;
660 let mut at = None;
661 for line in raw.lines() {
662 let Some(message) = parse_message_line(line) else {
663 continue;
664 };
665 position += 1;
666 if crate::agent::message_id(&message) == Some(message_id) {
667 at = Some(position);
668 }
669 }
670 let status = match (at, summary.truncation.as_ref()) {
671 (Some(position), _) => MessageStatus::Present {
672 position,
673 total: position.max(summary.messages),
674 },
675 (None, Some(marker)) => {
676 let dropped_at = message_id_timestamp_ms(&marker.last_dropped_id);
677 match (minted_ms, dropped_at) {
678 (Some(minted), Some(last)) if minted <= last => MessageStatus::Truncated {
679 dropped: marker.dropped,
680 oldest_surviving_id: marker.oldest_surviving_id.clone(),
681 },
682 _ => MessageStatus::Unknown,
683 }
684 }
685 (None, None) => MessageStatus::Unknown,
686 };
687 if status != MessageStatus::Unknown {
688 found.push(MessageLocation {
689 record: summary.record,
690 context_id: summary.context_id,
691 path: summary.path,
692 status,
693 });
694 }
695 }
696 Ok(found)
697}
698
699pub fn conversations_root() -> Result<PathBuf, String> {
703 Ok(crate::state_store::murmur_home_dir()?.join(CONVERSATION_ROOT_DIR))
704}
705
706pub fn now_ms() -> u64 {
708 std::time::SystemTime::now()
709 .duration_since(std::time::UNIX_EPOCH)
710 .map(|elapsed| elapsed.as_millis() as u64)
711 .unwrap_or(0)
712}
713
714fn child_dirs(dir: &Path) -> Vec<String> {
716 let mut names: Vec<String> = match std::fs::read_dir(dir) {
717 Ok(entries) => entries
718 .filter_map(Result::ok)
719 .filter(|entry| entry.path().is_dir())
720 .map(|entry| entry.file_name().to_string_lossy().into_owned())
721 .filter(|name| !name.starts_with('.'))
722 .collect(),
723 Err(_) => Vec::new(),
724 };
725 names.sort();
726 names
727}
728
729fn context_dirs(record_root: &Path) -> Vec<(String, PathBuf)> {
731 child_dirs(record_root)
732 .into_iter()
733 .map(|name| {
734 let path = record_root.join(&name);
735 (name, path)
736 })
737 .collect()
738}
739
740fn count_messages(path: &Path) -> u64 {
743 std::fs::read_to_string(path)
744 .map(|raw| {
745 raw.lines()
746 .filter(|line| parse_message_line(line).is_some())
747 .count() as u64
748 })
749 .unwrap_or(0)
750}
751
752fn last_touched_ms(path: &Path) -> Option<u64> {
754 std::fs::metadata(path)
755 .and_then(|metadata| metadata.modified())
756 .ok()?
757 .duration_since(std::time::UNIX_EPOCH)
758 .ok()
759 .map(|elapsed| elapsed.as_millis() as u64)
760}
761
762#[cfg(test)]
763mod tests {
764 use super::*;
765 use crate::conversation::parse_header;
766 use murmur_artifact::{ContextRetainConfig, TraceRetainConfig};
767 use serde_json::json;
768
769 fn ses(ms: u64, tail: u64) -> String {
772 format!("ses_{ms:012x}{tail:020x}")
773 }
774
775 fn msg(ms: u64, tail: u64) -> String {
777 format!("msg_{ms:012x}{tail:020x}")
778 }
779
780 fn message_line(id: &str, text: &str) -> String {
781 json!({"role": "user", "content": [{"type": "text", "text": text}], "id": id}).to_string()
782 }
783
784 fn make_sessions(root: &Path, names: &[String]) {
785 for name in names {
786 let dir = root.join(name);
787 std::fs::create_dir_all(dir.join("blobs")).unwrap();
788 std::fs::write(dir.join("trace.jsonl"), "{}\n").unwrap();
789 std::fs::write(dir.join("blobs").join("deadbeef"), b"body").unwrap();
790 }
791 }
792
793 fn names(root: &Path) -> Vec<String> {
794 let mut found: Vec<String> = std::fs::read_dir(root)
795 .unwrap()
796 .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
797 .collect();
798 found.sort();
799 found
800 }
801
802 const NOW: u64 = 1_756_400_000_000;
803
804 #[test]
808 fn max_sessions_leaves_exactly_the_newest_n_including_the_current() {
809 let root = tempfile::tempdir().unwrap();
810 let all: Vec<String> = (1..=6).map(|n| ses(NOW - (6 - n) * 1000, n)).collect();
811 make_sessions(root.path(), &all);
812
813 let pruned = prune_sessions(
814 root.path(),
815 &all[5],
816 &TraceRetainConfig {
817 max_sessions: Some(3),
818 max_age_secs: None,
819 },
820 NOW,
821 );
822
823 assert_eq!(
824 names(root.path()),
825 vec![all[3].clone(), all[4].clone(), all[5].clone()]
826 );
827 let mut removed: Vec<String> = pruned.iter().map(|p| p.name.clone()).collect();
828 removed.sort();
829 assert_eq!(
830 removed,
831 vec![all[0].clone(), all[1].clone(), all[2].clone()]
832 );
833 assert!(pruned
834 .iter()
835 .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_SESSIONS));
836 for name in &all[..3] {
837 assert!(!root.path().join(name).exists(), "{name} went whole");
838 }
839 }
840
841 #[test]
845 fn max_age_reads_the_id_and_never_the_filesystem() {
846 let root = tempfile::tempdir().unwrap();
847 let old_a = ses(NOW - 7_200_000, 1);
848 let old_b = ses(NOW - 7_100_000, 2);
849 let recent = ses(NOW - 60_000, 3);
850 let current = ses(NOW, 4);
851 make_sessions(
852 root.path(),
853 &[
854 old_a.clone(),
855 old_b.clone(),
856 recent.clone(),
857 current.clone(),
858 ],
859 );
860
861 let pruned = prune_sessions(
862 root.path(),
863 ¤t,
864 &TraceRetainConfig {
865 max_sessions: None,
866 max_age_secs: Some(900),
867 },
868 NOW,
869 );
870
871 assert_eq!(pruned.len(), 2, "{pruned:?}");
872 assert!(pruned
873 .iter()
874 .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_AGE));
875 assert_eq!(names(root.path()), vec![recent, current]);
876 }
877
878 #[test]
881 fn both_trace_keys_keep_only_sessions_inside_both() {
882 let root = tempfile::tempdir().unwrap();
883 let old = [ses(NOW - 7_200_000, 1), ses(NOW - 7_100_000, 2)];
884 let recent = [
885 ses(NOW - 300_000, 3),
886 ses(NOW - 200_000, 4),
887 ses(NOW - 100_000, 5),
888 ];
889 let current = ses(NOW, 6);
890 let mut all = old.to_vec();
891 all.extend(recent.iter().cloned());
892 all.push(current.clone());
893 make_sessions(root.path(), &all);
894
895 let pruned = prune_sessions(
896 root.path(),
897 ¤t,
898 &TraceRetainConfig {
899 max_sessions: Some(2),
900 max_age_secs: Some(900),
901 },
902 NOW,
903 );
904
905 assert_eq!(names(root.path()), vec![recent[2].clone(), current.clone()]);
906 assert_eq!(pruned.len(), 4);
907 for session in &pruned {
908 let expected = if old.contains(&session.name) {
909 crate::trace::RETENTION_REASON_MAX_AGE
910 } else {
911 crate::trace::RETENTION_REASON_MAX_SESSIONS
912 };
913 assert_eq!(session.reason, expected, "{session:?}");
914 }
915 }
916
917 #[test]
920 fn nothing_at_or_after_the_current_session_id_is_a_candidate() {
921 let root = tempfile::tempdir().unwrap();
922 let current = ses(NOW, 5);
923 let sibling = ses(NOW + 1, 6);
925 let older = ses(NOW - 1000, 4);
926 make_sessions(
927 root.path(),
928 &[current.clone(), sibling.clone(), older.clone()],
929 );
930
931 let pruned = prune_sessions(
932 root.path(),
933 ¤t,
934 &TraceRetainConfig {
935 max_sessions: Some(1),
936 max_age_secs: Some(1),
937 },
938 NOW,
939 );
940
941 assert_eq!(pruned.len(), 1);
942 assert_eq!(pruned[0].name, older);
943 assert_eq!(names(root.path()), vec![current.clone(), sibling]);
944 assert!(root.path().join(¤t).join("trace.jsonl").exists());
945 }
946
947 #[test]
950 fn an_empty_policy_removes_nothing() {
951 let root = tempfile::tempdir().unwrap();
952 let all = [ses(NOW - 9_000_000, 1), ses(NOW, 2)];
953 make_sessions(root.path(), &all);
954 assert!(prune_sessions(
955 root.path(),
956 &all[1],
957 &TraceRetainConfig {
958 max_sessions: None,
959 max_age_secs: None
960 },
961 NOW
962 )
963 .is_empty());
964 assert_eq!(names(root.path()), all.to_vec());
965 }
966
967 fn write_record(dir: &Path, capsule: Option<&str>, count: u64) -> PathBuf {
971 std::fs::create_dir_all(dir).unwrap();
972 let path = dir.join(RECORD_FILE_NAME);
973 let mut contents = String::new();
974 if let Some(capsule) = capsule {
975 contents.push_str(
976 &header_line(&RecordHeader {
977 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
978 capsule: capsule.to_string(),
979 created_ms: NOW,
980 truncated: None,
981 })
982 .unwrap(),
983 );
984 }
985 for i in 1..=count {
986 contents.push_str(&message_line(
987 &msg(NOW - (count - i) * 1000, i),
988 &format!("m{i}"),
989 ));
990 contents.push('\n');
991 }
992 std::fs::write(&path, contents).unwrap();
993 path
994 }
995
996 #[test]
999 fn max_messages_leaves_the_newest_n_with_their_ids_intact() {
1000 let dir = tempfile::tempdir().unwrap();
1001 let path = write_record(dir.path(), Some("capsule"), 10);
1002 let before: Vec<String> = std::fs::read_to_string(&path)
1003 .unwrap()
1004 .lines()
1005 .filter_map(parse_message_line)
1006 .map(|m| m["id"].as_str().unwrap().to_string())
1007 .collect();
1008
1009 let outcome = truncate_record(&path, 3, "capsule").unwrap();
1010 assert_eq!(outcome.dropped, 7);
1011 assert_eq!(outcome.kept, 3);
1012 assert_eq!(outcome.last_dropped_id, before[6]);
1013 assert_eq!(outcome.oldest_surviving_id, before[7]);
1014
1015 let raw = std::fs::read_to_string(&path).unwrap();
1016 let lines: Vec<&str> = raw.lines().collect();
1017 assert_eq!(
1018 lines.len(),
1019 4,
1020 "a header plus exactly three messages: {lines:?}"
1021 );
1022 for line in &lines {
1023 serde_json::from_str::<serde_json::Value>(line).expect("every line is JSON");
1024 }
1025 let header = parse_header(lines[0]).expect("the first line is the header");
1026 assert_eq!(header.capsule, "capsule");
1027 let marker = header.truncated.expect("the header records the drop");
1028 assert_eq!(marker.dropped, 7);
1029 assert_eq!(marker.last_dropped_id, before[6]);
1030 assert_eq!(marker.oldest_surviving_id, before[7]);
1031
1032 let after: Vec<String> = lines[1..]
1033 .iter()
1034 .map(|line| {
1035 parse_message_line(line).expect("a message with a string role")["id"]
1036 .as_str()
1037 .unwrap()
1038 .to_string()
1039 })
1040 .collect();
1041 assert_eq!(after, before[7..], "surviving ids are the bytes they were");
1042 }
1043
1044 #[test]
1047 fn a_header_line_is_not_counted_as_a_message() {
1048 let dir = tempfile::tempdir().unwrap();
1049 let headerless = write_record(&dir.path().join("a"), None, 4);
1050 let owned = write_record(&dir.path().join("b"), Some("capsule"), 4);
1051 assert_eq!(count_messages(&headerless), 4);
1052 assert_eq!(count_messages(&owned), 4);
1053 }
1054
1055 #[test]
1058 fn a_second_truncation_accumulates_the_dropped_count() {
1059 let dir = tempfile::tempdir().unwrap();
1060 let path = write_record(dir.path(), Some("capsule"), 10);
1061 truncate_record(&path, 6, "capsule").unwrap();
1062 truncate_record(&path, 2, "capsule").unwrap();
1063 let marker = read_header(&path).unwrap().truncated.unwrap();
1064 assert_eq!(marker.dropped, 8);
1065 assert_eq!(count_messages(&path), 2);
1066 }
1067
1068 #[test]
1069 fn truncation_rewrite_keeps_the_record_owner_only() {
1070 use std::os::unix::fs::PermissionsExt;
1071 let dir = tempfile::tempdir().unwrap();
1072 let path = write_record(dir.path(), Some("capsule"), 5);
1073 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
1074
1075 truncate_record(&path, 2, "capsule").unwrap();
1076
1077 assert_eq!(count_messages(&path), 2);
1078 assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
1079 }
1080
1081 #[test]
1083 fn a_record_inside_the_limit_is_not_rewritten() {
1084 let dir = tempfile::tempdir().unwrap();
1085 let path = write_record(dir.path(), Some("capsule"), 3);
1086 let before = std::fs::read(&path).unwrap();
1087 let outcome = truncate_record(&path, 3, "capsule").unwrap();
1088 assert_eq!(outcome.dropped, 0);
1089 assert_eq!(std::fs::read(&path).unwrap(), before);
1090 }
1091
1092 #[test]
1096 fn a_staged_truncation_leaves_the_original_intact_until_the_rename() {
1097 let dir = tempfile::tempdir().unwrap();
1098 let path = write_record(dir.path(), Some("capsule"), 10);
1099 let before = std::fs::read(&path).unwrap();
1100
1101 let (staged, outcome) = stage_truncation(&path, 3, "capsule", NOW).unwrap().unwrap();
1102 assert_eq!(outcome.dropped, 7);
1103
1104 let temp = staged.temp_path().to_path_buf();
1105 assert_eq!(
1106 temp.parent(),
1107 path.parent(),
1108 "the temp file is in the record's own directory, so the rename is atomic"
1109 );
1110 assert_ne!(temp.file_name().unwrap(), RECORD_FILE_NAME);
1111
1112 assert_eq!(
1113 std::fs::read(&path).unwrap(),
1114 before,
1115 "the original is byte-identical before the rename"
1116 );
1117 assert_eq!(count_messages(&path), 10);
1118 for line in std::fs::read_to_string(&path).unwrap().lines() {
1119 serde_json::from_str::<serde_json::Value>(line).expect("the original parses whole");
1120 }
1121
1122 drop(staged);
1124 assert!(!temp.exists());
1125 assert_eq!(std::fs::read(&path).unwrap(), before);
1126 assert_eq!(count_messages(&path), 10);
1127 }
1128
1129 #[test]
1135 fn record_pruning_skips_another_capsules_records_and_every_unowned_one() {
1136 let root = tempfile::tempdir().unwrap();
1137 write_record(&root.path().join("mine"), Some("mine"), 3);
1138 write_record(&root.path().join("theirs"), Some("theirs"), 3);
1139 write_record(&root.path().join("unowned"), None, 3);
1140 write_record(&root.path().join("current"), Some("mine"), 3);
1141
1142 let far_future = now_ms() + 3_600_000;
1144 let pruned = prune_records(
1145 root.path(),
1146 "mine",
1147 Some("current"),
1148 &ContextRetainConfig {
1149 max_messages: None,
1150 max_age_secs: Some(1),
1151 },
1152 far_future,
1153 );
1154
1155 assert_eq!(pruned.len(), 1, "{pruned:?}");
1156 assert_eq!(pruned[0].context_id, "mine");
1157 assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
1158 assert_eq!(names(root.path()), vec!["current", "theirs", "unowned"]);
1159 }
1160
1161 #[test]
1164 fn a_record_written_inside_the_window_is_kept() {
1165 let root = tempfile::tempdir().unwrap();
1166 write_record(&root.path().join("fresh"), Some("mine"), 3);
1167 let pruned = prune_records(
1168 root.path(),
1169 "mine",
1170 None,
1171 &ContextRetainConfig {
1172 max_messages: None,
1173 max_age_secs: Some(90 * 86_400),
1174 },
1175 now_ms(),
1176 );
1177 assert!(pruned.is_empty());
1178 assert_eq!(names(root.path()), vec!["fresh"]);
1179 }
1180
1181 #[test]
1184 fn max_messages_alone_removes_no_record() {
1185 let root = tempfile::tempdir().unwrap();
1186 write_record(&root.path().join("ctx"), Some("mine"), 100);
1187 assert!(prune_records(
1188 root.path(),
1189 "mine",
1190 None,
1191 &ContextRetainConfig {
1192 max_messages: Some(1),
1193 max_age_secs: None
1194 },
1195 now_ms() + 3_600_000,
1196 )
1197 .is_empty());
1198 assert_eq!(names(root.path()), vec!["ctx"]);
1199 }
1200
1201 fn frm(ms: u64, tail: u64) -> FormationId {
1205 FormationId::parse(&format!("frm_{ms:012x}{tail:020x}")).unwrap()
1206 }
1207
1208 fn retain(max_sessions: Option<u32>, max_age_secs: Option<u64>) -> TraceRetainConfig {
1209 TraceRetainConfig {
1210 max_sessions,
1211 max_age_secs,
1212 }
1213 }
1214
1215 fn make_formation(root: &Path, id: &FormationId, marker: Option<&str>) -> PathBuf {
1218 let dir = root.join(id.as_str());
1219 let session = dir.join("worker/.murmur").join(ses(NOW, 1));
1220 std::fs::create_dir_all(&session).unwrap();
1221 std::fs::write(session.join("trace.jsonl"), "{}\n").unwrap();
1222 if let Some(marker) = marker {
1223 std::fs::write(
1224 dir.join(crate::formation_launch::FORMATION_MARKER_FILE),
1225 marker,
1226 )
1227 .unwrap();
1228 }
1229 dir
1230 }
1231
1232 fn marker_naming(project: &Path) -> String {
1233 serde_json::to_string(&FormationMarker {
1234 project_dir: project.to_path_buf(),
1235 launcher_pid: 1,
1236 launcher_start: String::new(),
1237 })
1238 .unwrap()
1239 }
1240
1241 fn make_owned(root: &Path, project: &Path, ids: &[&FormationId]) {
1243 for id in ids {
1244 make_formation(root, id, Some(&marker_naming(project)));
1245 }
1246 }
1247
1248 fn removed(pruned: &[PrunedFormation]) -> Vec<(FormationId, &'static str)> {
1249 pruned
1250 .iter()
1251 .map(|p| (p.formation_id.clone(), p.reason))
1252 .collect()
1253 }
1254
1255 const DAY_MS: u64 = 86_400_000;
1256
1257 #[test]
1260 fn prune_formations_never_removes_the_current_or_a_later_formation() {
1261 let root = tempfile::tempdir().unwrap();
1262 let project = tempfile::tempdir().unwrap();
1263 let old = frm(NOW - 3 * DAY_MS, 1);
1264 let current = frm(NOW - 2 * DAY_MS, 2);
1265 let later = frm(NOW - 2 * DAY_MS + 1000, 3);
1266 make_owned(root.path(), project.path(), &[&old, ¤t, &later]);
1267
1268 let pruned = prune_formations(
1269 root.path(),
1270 ¤t,
1271 project.path(),
1272 &retain(Some(1), Some(1)),
1273 NOW,
1274 |_, _| false,
1275 );
1276
1277 assert_eq!(
1278 removed(&pruned),
1279 [(old.clone(), crate::trace::RETENTION_REASON_MAX_AGE)]
1280 );
1281 assert_eq!(pruned[0].path, root.path().join(old.as_str()));
1282 assert_eq!(
1283 names(root.path()),
1284 vec![current.as_str().to_string(), later.as_str().to_string()]
1285 );
1286 }
1287
1288 #[test]
1291 fn prune_formations_keeps_the_newest_n_with_the_current_counted() {
1292 let root = tempfile::tempdir().unwrap();
1293 let project = tempfile::tempdir().unwrap();
1294 let all: Vec<FormationId> = (1..=5).map(|n| frm(NOW - (5 - n) * 1000, n)).collect();
1295 make_owned(root.path(), project.path(), &all.iter().collect::<Vec<_>>());
1296
1297 let pruned = prune_formations(
1298 root.path(),
1299 &all[3],
1300 project.path(),
1301 &retain(Some(3), None),
1302 NOW,
1303 |_, _| false,
1304 );
1305
1306 let reason = crate::trace::RETENTION_REASON_MAX_SESSIONS;
1307 assert_eq!(
1308 removed(&pruned),
1309 [(all[1].clone(), reason), (all[0].clone(), reason)]
1310 );
1311 assert_eq!(
1312 names(root.path()),
1313 all[2..]
1314 .iter()
1315 .map(|id| id.as_str().to_string())
1316 .collect::<Vec<_>>()
1317 );
1318 }
1319
1320 #[test]
1323 fn prune_formations_reads_age_from_the_id() {
1324 let root = tempfile::tempdir().unwrap();
1325 let project = tempfile::tempdir().unwrap();
1326 let old_a = frm(NOW - 7_200_000, 1);
1327 let old_b = frm(NOW - 7_100_000, 2);
1328 let recent = frm(NOW - 60_000, 3);
1329 let current = frm(NOW, 4);
1330 make_owned(
1331 root.path(),
1332 project.path(),
1333 &[&old_a, &old_b, &recent, ¤t],
1334 );
1335
1336 let pruned = prune_formations(
1337 root.path(),
1338 ¤t,
1339 project.path(),
1340 &retain(None, Some(900)),
1341 NOW,
1342 |_, _| false,
1343 );
1344
1345 let reason = crate::trace::RETENTION_REASON_MAX_AGE;
1346 assert_eq!(removed(&pruned), [(old_b, reason), (old_a, reason)]);
1347 assert_eq!(
1348 names(root.path()),
1349 vec![recent.as_str().to_string(), current.as_str().to_string()]
1350 );
1351 }
1352
1353 #[test]
1356 fn prune_formations_attributes_reasons_age_first() {
1357 let root = tempfile::tempdir().unwrap();
1358 let project = tempfile::tempdir().unwrap();
1359 let old = frm(NOW - 7_200_000, 1);
1360 let recent = frm(NOW - 60_000, 2);
1361 let current = frm(NOW, 3);
1362 make_owned(root.path(), project.path(), &[&old, &recent, ¤t]);
1363
1364 let pruned = prune_formations(
1365 root.path(),
1366 ¤t,
1367 project.path(),
1368 &retain(Some(1), Some(900)),
1369 NOW,
1370 |_, _| false,
1371 );
1372
1373 assert_eq!(
1374 removed(&pruned),
1375 [
1376 (recent, crate::trace::RETENTION_REASON_MAX_SESSIONS),
1377 (old, crate::trace::RETENTION_REASON_MAX_AGE),
1378 ]
1379 );
1380 }
1381
1382 #[test]
1385 fn prune_formations_skips_a_live_directory_and_keeps_its_rank() {
1386 let root = tempfile::tempdir().unwrap();
1387 let project = tempfile::tempdir().unwrap();
1388 let older = frm(NOW - 3000, 1);
1389 let live = frm(NOW - 2000, 2);
1390 let current = frm(NOW, 3);
1391 make_owned(root.path(), project.path(), &[&older, &live, ¤t]);
1392
1393 let asked = std::cell::RefCell::new(Vec::new());
1394 let pruned = prune_formations(
1395 root.path(),
1396 ¤t,
1397 project.path(),
1398 &retain(Some(1), None),
1399 NOW,
1400 |id, marker| {
1401 assert_eq!(marker.launcher_pid, 1);
1402 asked.borrow_mut().push(id.clone());
1403 *id == live
1404 },
1405 );
1406
1407 assert_eq!(
1408 removed(&pruned),
1409 [(older.clone(), crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1410 );
1411 assert_eq!(*asked.borrow(), [live.clone(), older]);
1412 assert_eq!(
1413 names(root.path()),
1414 vec![live.as_str().to_string(), current.as_str().to_string()]
1415 );
1416
1417 let kept = frm(NOW - 2500, 4);
1419 make_owned(root.path(), project.path(), &[&kept]);
1420 let pruned = prune_formations(
1421 root.path(),
1422 ¤t,
1423 project.path(),
1424 &retain(Some(3), None),
1425 NOW,
1426 |id, _| *id == live,
1427 );
1428 assert!(pruned.is_empty(), "{pruned:?}");
1429 let pruned = prune_formations(
1430 root.path(),
1431 ¤t,
1432 project.path(),
1433 &retain(Some(2), None),
1434 NOW,
1435 |id, _| *id == live,
1436 );
1437 assert_eq!(
1438 removed(&pruned),
1439 [(kept, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1440 );
1441 }
1442
1443 #[test]
1446 fn prune_formations_never_ranks_or_removes_unowned_or_foreign_directories() {
1447 let root = tempfile::tempdir().unwrap();
1448 let project = tempfile::tempdir().unwrap();
1449 let other = tempfile::tempdir().unwrap();
1450 let owned_old = frm(NOW - 3 * DAY_MS, 1);
1451 let owned_kept = frm(NOW - 2 * DAY_MS, 2);
1452 let unmarked = frm(NOW - 2 * DAY_MS + 1, 3);
1453 let garbled = frm(NOW - 2 * DAY_MS + 2, 4);
1454 let foreign = frm(NOW - 2 * DAY_MS + 3, 5);
1455 let current = frm(NOW, 6);
1456 make_owned(
1457 root.path(),
1458 project.path(),
1459 &[&owned_old, &owned_kept, ¤t],
1460 );
1461 make_formation(root.path(), &unmarked, None);
1462 make_formation(root.path(), &garbled, Some("not json {"));
1463 make_formation(root.path(), &foreign, Some(&marker_naming(other.path())));
1464
1465 let pruned = prune_formations(
1466 root.path(),
1467 ¤t,
1468 project.path(),
1469 &retain(Some(2), None),
1470 NOW,
1471 |_, _| false,
1472 );
1473 assert_eq!(
1474 removed(&pruned),
1475 [(owned_old, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1476 );
1477
1478 let pruned = prune_formations(
1479 root.path(),
1480 ¤t,
1481 project.path(),
1482 &retain(Some(1), Some(DAY_MS / 1000)),
1483 NOW,
1484 |_, _| false,
1485 );
1486 assert_eq!(
1487 removed(&pruned),
1488 [(owned_kept, crate::trace::RETENTION_REASON_MAX_AGE)]
1489 );
1490 for id in [&unmarked, &garbled, &foreign, ¤t] {
1491 assert!(root.path().join(id.as_str()).is_dir(), "{id}");
1492 }
1493 }
1494
1495 #[test]
1498 fn prune_formations_ignores_a_symlink_a_regular_file_and_a_name_that_is_not_an_id() {
1499 let root = tempfile::tempdir().unwrap();
1500 let elsewhere = tempfile::tempdir().unwrap();
1501 let project = tempfile::tempdir().unwrap();
1502 let marker = marker_naming(project.path());
1503 let target = make_formation(elsewhere.path(), &frm(NOW - 3 * DAY_MS, 1), Some(&marker));
1504 let linked = frm(NOW - 3 * DAY_MS, 2);
1505 std::os::unix::fs::symlink(&target, root.path().join(linked.as_str())).unwrap();
1506 std::fs::write(root.path().join(frm(NOW - 3 * DAY_MS, 3).as_str()), &marker).unwrap();
1507 for name in ["not-a-formation", "frm_ABC", "frm_0123"] {
1508 std::fs::create_dir(root.path().join(name)).unwrap();
1509 std::fs::write(
1510 root.path()
1511 .join(name)
1512 .join(crate::formation_launch::FORMATION_MARKER_FILE),
1513 &marker,
1514 )
1515 .unwrap();
1516 }
1517 let before = names(root.path());
1518
1519 let pruned = prune_formations(
1520 root.path(),
1521 &frm(NOW, 9),
1522 project.path(),
1523 &retain(Some(1), Some(1)),
1524 NOW,
1525 |_, _| false,
1526 );
1527
1528 assert!(pruned.is_empty(), "{pruned:?}");
1529 assert_eq!(names(root.path()), before);
1530 assert!(target.join("worker/.murmur").is_dir());
1531 }
1532
1533 #[test]
1535 fn prune_formations_with_neither_key_or_no_directory_removes_nothing() {
1536 let root = tempfile::tempdir().unwrap();
1537 let project = tempfile::tempdir().unwrap();
1538 let old = frm(NOW - 3 * DAY_MS, 1);
1539 make_owned(root.path(), project.path(), &[&old]);
1540
1541 let pruned = prune_formations(
1542 root.path(),
1543 &frm(NOW, 2),
1544 project.path(),
1545 &retain(None, None),
1546 NOW,
1547 |_, _| panic!("nothing is a candidate without a policy"),
1548 );
1549 assert!(pruned.is_empty());
1550 assert!(root.path().join(old.as_str()).is_dir());
1551
1552 assert!(prune_formations(
1553 &root.path().join("missing"),
1554 &frm(NOW, 2),
1555 project.path(),
1556 &retain(Some(1), Some(1)),
1557 NOW,
1558 |_, _| false,
1559 )
1560 .is_empty());
1561 }
1562
1563 #[test]
1566 fn prune_formations_compares_project_directories_canonically() {
1567 let root = tempfile::tempdir().unwrap();
1568 let project = tempfile::tempdir().unwrap();
1569 let links = tempfile::tempdir().unwrap();
1570 let link = links.path().join("project");
1571 std::os::unix::fs::symlink(project.path(), &link).unwrap();
1572 let old = frm(NOW - 3 * DAY_MS, 1);
1573 make_owned(
1574 root.path(),
1575 &std::fs::canonicalize(project.path()).unwrap(),
1576 &[&old],
1577 );
1578
1579 let pruned = prune_formations(
1580 root.path(),
1581 &frm(NOW, 2),
1582 &link,
1583 &retain(None, Some(1)),
1584 NOW,
1585 |_, _| false,
1586 );
1587 assert_eq!(
1588 removed(&pruned),
1589 [(old, crate::trace::RETENTION_REASON_MAX_AGE)]
1590 );
1591 }
1592
1593 #[test]
1595 fn prune_formations_leaves_a_directory_it_could_not_remove_out_of_the_result() {
1596 use std::os::unix::fs::PermissionsExt;
1597
1598 let root = tempfile::tempdir().unwrap();
1599 let project = tempfile::tempdir().unwrap();
1600 let stuck = frm(NOW - 3 * DAY_MS, 1);
1601 let gone = frm(NOW - 3 * DAY_MS + 1, 2);
1602 make_owned(root.path(), project.path(), &[&stuck, &gone]);
1603 let stuck_dir = root.path().join(stuck.as_str());
1604 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o500)).unwrap();
1605 if std::fs::write(stuck_dir.join("probe"), b"").is_ok() {
1607 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
1608 return;
1609 }
1610
1611 let pruned = prune_formations(
1612 root.path(),
1613 &frm(NOW, 3),
1614 project.path(),
1615 &retain(None, Some(1)),
1616 NOW,
1617 |_, _| false,
1618 );
1619 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
1620
1621 assert_eq!(
1622 removed(&pruned),
1623 [(gone, crate::trace::RETENTION_REASON_MAX_AGE)]
1624 );
1625 assert!(stuck_dir.is_dir());
1626 }
1627}