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 if let Some(dir) = path.parent() {
471 crate::record_seal::seal_conversation(dir, crate::record_seal::CLI_SEALER);
472 }
473 outcome.blobs_removed = sweep_orphan_blobs(path);
474 Ok(outcome)
475 }
476 }
477}
478
479fn sweep_orphan_blobs(path: &Path) -> u64 {
483 let Some(blobs) = path
484 .parent()
485 .map(|dir| dir.join(crate::trace_blobs::BLOB_DIR_NAME))
486 else {
487 return 0;
488 };
489 let Ok(entries) = std::fs::read_dir(&blobs) else {
490 return 0;
491 };
492 let Ok(raw) = std::fs::read_to_string(path) else {
493 return 0;
494 };
495 let referenced: std::collections::HashSet<String> = raw
496 .lines()
497 .filter_map(parse_message_line)
498 .flat_map(|message| {
499 crate::images::referenced_blobs(&message)
500 .map(str::to_string)
501 .collect::<Vec<_>>()
502 })
503 .collect();
504 let mut removed = 0;
505 for entry in entries.flatten() {
506 let name = entry.file_name();
507 let Some(name) = name.to_str() else {
508 continue;
509 };
510 if !murmur_artifact::is_sha256_hex(name) || referenced.contains(name) {
511 continue;
512 }
513 if std::fs::remove_file(entry.path()).is_ok() {
514 removed += 1;
515 }
516 }
517 removed
518}
519
520pub(crate) fn stage_truncation(
526 path: &Path,
527 keep: u32,
528 capsule: &str,
529 at_ms: u64,
530) -> Result<Option<(StagedRewrite, TruncationOutcome)>, String> {
531 let raw = match std::fs::read_to_string(path) {
532 Ok(raw) => raw,
533 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
534 Err(err) => return Err(format!("{}: {err}", path.display())),
535 };
536
537 let existing = read_header(path);
538 let lines: Vec<&str> = raw.lines().collect();
539 let message_indices: Vec<usize> = lines
542 .iter()
543 .enumerate()
544 .filter(|(_, line)| parse_message_line(line).is_some())
545 .map(|(index, _)| index)
546 .collect();
547
548 let total = message_indices.len();
549 let keep = keep as usize;
550 if total <= keep {
551 return Ok(None);
552 }
553 let dropped = total - keep;
554 let first_kept_index = message_indices[dropped];
555
556 let id_of = |index: usize| -> String {
557 parse_message_line(lines[index])
558 .as_ref()
559 .and_then(|message| crate::agent::message_id(message).map(str::to_string))
560 .unwrap_or_default()
561 };
562 let last_dropped_id = id_of(message_indices[dropped - 1]);
563 let oldest_surviving_id = message_indices
564 .get(dropped)
565 .map(|i| id_of(*i))
566 .unwrap_or_default();
567
568 let header = RecordHeader {
569 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
570 capsule: existing
571 .as_ref()
572 .map(|h| h.capsule.clone())
573 .unwrap_or_else(|| capsule.to_string()),
574 created_ms: existing.as_ref().map(|h| h.created_ms).unwrap_or(at_ms),
575 truncated: Some(TruncationMarker {
576 dropped: existing
579 .as_ref()
580 .and_then(|h| h.truncated.as_ref())
581 .map(|marker| marker.dropped)
582 .unwrap_or(0)
583 + dropped as u64,
584 oldest_surviving_id: oldest_surviving_id.clone(),
585 last_dropped_id: last_dropped_id.clone(),
586 at_ms,
587 }),
588 };
589
590 let mut contents = header_line(&header)?;
591 for line in &lines[first_kept_index..] {
592 contents.push_str(line);
593 contents.push('\n');
594 }
595
596 let staged = StagedRewrite::stage_with_mode(
597 path,
598 contents.as_bytes(),
599 Some(crate::conversation::RECORD_FILE_MODE),
600 )?;
601 Ok(Some((
602 staged,
603 TruncationOutcome {
604 dropped: dropped as u64,
605 kept: keep as u64,
606 oldest_surviving_id,
607 last_dropped_id,
608 blobs_removed: 0,
609 },
610 )))
611}
612
613pub(crate) struct StagedRewrite {
619 target: PathBuf,
620 temp: PathBuf,
621 committed: bool,
622}
623
624impl StagedRewrite {
625 pub(crate) fn stage_with_mode(
628 target: &Path,
629 contents: &[u8],
630 mode: Option<u32>,
631 ) -> Result<Self, String> {
632 use std::io::Write;
633 use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
634
635 let dir = target
636 .parent()
637 .ok_or_else(|| format!("{} has no parent directory", target.display()))?;
638 let name = target
639 .file_name()
640 .map(|name| name.to_string_lossy().into_owned())
641 .unwrap_or_else(|| "record".to_string());
642 let temp = dir.join(format!(
643 ".{name}.{}.{}.tmp",
644 std::process::id(),
645 uuid::Uuid::now_v7().simple()
646 ));
647 let mut options = std::fs::OpenOptions::new();
648 options.write(true).create(true).truncate(true);
649 if let Some(mode) = mode {
650 options.mode(mode);
651 }
652 let mut file = options
653 .open(&temp)
654 .map_err(|err| format!("{}: {err}", temp.display()))?;
655 file.write_all(contents)
658 .and_then(|()| file.sync_all())
659 .map_err(|err| format!("{}: {err}", temp.display()))?;
660 if let Some(mode) = mode {
664 std::fs::set_permissions(&temp, std::fs::Permissions::from_mode(mode))
665 .map_err(|err| format!("{}: {err}", temp.display()))?;
666 }
667 Ok(Self {
668 target: target.to_path_buf(),
669 temp,
670 committed: false,
671 })
672 }
673
674 #[cfg(test)]
675 pub(crate) fn temp_path(&self) -> &Path {
676 &self.temp
677 }
678
679 pub(crate) fn commit(mut self) -> Result<(), String> {
680 std::fs::rename(&self.temp, &self.target)
681 .map_err(|err| format!("{}: {err}", self.target.display()))?;
682 self.committed = true;
683 Ok(())
684 }
685}
686
687impl Drop for StagedRewrite {
688 fn drop(&mut self) {
689 if !self.committed {
690 let _ = std::fs::remove_file(&self.temp);
691 }
692 }
693}
694
695pub fn list_records() -> Result<Vec<RecordSummary>, String> {
703 let root = conversations_root()?;
704 let mut summaries = Vec::new();
705 for record in child_dirs(&root) {
706 let store = root.join(&record);
707 for (context_id, dir) in context_dirs(&store) {
708 let path = dir.join(RECORD_FILE_NAME);
709 if !path.exists() {
710 continue;
711 }
712 let header = read_header(&path);
713 summaries.push(RecordSummary {
714 record: record.clone(),
715 context_id,
716 messages: count_messages(&path),
717 bytes: std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0),
718 last_touched_ms: last_touched_ms(&path),
719 capsule: header.as_ref().map(|h| h.capsule.clone()),
720 truncation: header.and_then(|h| h.truncated),
721 path,
722 });
723 }
724 }
725 summaries.sort_by(|a, b| {
726 a.record
727 .cmp(&b.record)
728 .then_with(|| a.context_id.cmp(&b.context_id))
729 });
730 Ok(summaries)
731}
732
733pub fn remove_record(record: &str, context_id: &str) -> Result<RemovedRecord, String> {
738 let dir = conversations_root()?.join(record).join(context_id);
739 if !dir.is_dir() {
740 return Err(format!("{} is not a context directory", dir.display()));
741 }
742 let messages = count_messages(&dir.join(RECORD_FILE_NAME));
743 std::fs::remove_dir_all(&dir).map_err(|err| format!("{}: {err}", dir.display()))?;
744 Ok(RemovedRecord {
745 record: record.to_string(),
746 context_id: context_id.to_string(),
747 path: dir,
748 messages,
749 })
750}
751
752pub fn locate_message(message_id: &str) -> Result<Vec<MessageLocation>, String> {
764 let minted_ms = message_id_timestamp_ms(message_id);
765 let mut found = Vec::new();
766 for summary in list_records()? {
767 let raw = std::fs::read_to_string(&summary.path).unwrap_or_default();
768 let mut position = 0u64;
769 let mut at = None;
770 for line in raw.lines() {
771 let Some(message) = parse_message_line(line) else {
772 continue;
773 };
774 position += 1;
775 if crate::agent::message_id(&message) == Some(message_id) {
776 at = Some(position);
777 }
778 }
779 let status = match (at, summary.truncation.as_ref()) {
780 (Some(position), _) => MessageStatus::Present {
781 position,
782 total: position.max(summary.messages),
783 },
784 (None, Some(marker)) => {
785 let dropped_at = message_id_timestamp_ms(&marker.last_dropped_id);
786 match (minted_ms, dropped_at) {
787 (Some(minted), Some(last)) if minted <= last => MessageStatus::Truncated {
788 dropped: marker.dropped,
789 oldest_surviving_id: marker.oldest_surviving_id.clone(),
790 },
791 _ => MessageStatus::Unknown,
792 }
793 }
794 (None, None) => MessageStatus::Unknown,
795 };
796 if status != MessageStatus::Unknown {
797 found.push(MessageLocation {
798 record: summary.record,
799 context_id: summary.context_id,
800 path: summary.path,
801 status,
802 });
803 }
804 }
805 Ok(found)
806}
807
808pub fn find_context_dirs(context_id: &str) -> Result<Vec<ContextDir>, String> {
815 Ok(find_context_dirs_in(&conversations_root()?, context_id))
816}
817
818fn find_context_dirs_in(root: &Path, context_id: &str) -> Vec<ContextDir> {
819 let mut found = Vec::new();
820 for record in child_dirs(root) {
821 for (name, path) in context_dirs(&root.join(&record)) {
822 if name != context_id {
823 continue;
824 }
825 if path.join(RECORD_FILE_NAME).exists() || path.join(HARNESS_SESSION_FILE_NAME).exists()
826 {
827 found.push(ContextDir {
828 record: record.clone(),
829 context_id: name,
830 path,
831 });
832 }
833 }
834 }
835 found
836}
837
838pub fn read_context_dir(dir: &Path) -> Result<ContextContents, String> {
846 let record = dir.join(RECORD_FILE_NAME);
847 let raw = match std::fs::read_to_string(&record) {
848 Ok(raw) => raw,
849 Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
850 Err(err) => return Err(format!("{}: {err}", record.display())),
851 };
852 Ok(ContextContents {
853 header: raw.lines().next().and_then(parse_header),
854 messages: record_messages(&raw, |_| {}),
855 harness_session: read_entry(&dir.join(HARNESS_SESSION_FILE_NAME)),
856 })
857}
858
859pub fn conversations_root() -> Result<PathBuf, String> {
863 Ok(crate::state_store::murmur_home_dir()?.join(CONVERSATION_ROOT_DIR))
864}
865
866pub fn now_ms() -> u64 {
868 std::time::SystemTime::now()
869 .duration_since(std::time::UNIX_EPOCH)
870 .map(|elapsed| elapsed.as_millis() as u64)
871 .unwrap_or(0)
872}
873
874fn child_dirs(dir: &Path) -> Vec<String> {
876 let mut names: Vec<String> = match std::fs::read_dir(dir) {
877 Ok(entries) => entries
878 .filter_map(Result::ok)
879 .filter(|entry| entry.path().is_dir())
880 .map(|entry| entry.file_name().to_string_lossy().into_owned())
881 .filter(|name| !name.starts_with('.'))
882 .collect(),
883 Err(_) => Vec::new(),
884 };
885 names.sort();
886 names
887}
888
889fn context_dirs(record_root: &Path) -> Vec<(String, PathBuf)> {
891 child_dirs(record_root)
892 .into_iter()
893 .map(|name| {
894 let path = record_root.join(&name);
895 (name, path)
896 })
897 .collect()
898}
899
900fn count_messages(path: &Path) -> u64 {
903 std::fs::read_to_string(path)
904 .map(|raw| {
905 raw.lines()
906 .filter(|line| parse_message_line(line).is_some())
907 .count() as u64
908 })
909 .unwrap_or(0)
910}
911
912fn last_touched_ms(path: &Path) -> Option<u64> {
914 std::fs::metadata(path)
915 .and_then(|metadata| metadata.modified())
916 .ok()?
917 .duration_since(std::time::UNIX_EPOCH)
918 .ok()
919 .map(|elapsed| elapsed.as_millis() as u64)
920}
921
922#[cfg(test)]
923mod tests {
924 use super::*;
925 use murmur_artifact::{ContextRetainConfig, TraceRetainConfig};
926 use serde_json::json;
927
928 fn ses(ms: u64, tail: u64) -> String {
931 format!("ses_{ms:012x}{tail:020x}")
932 }
933
934 fn msg(ms: u64, tail: u64) -> String {
936 format!("msg_{ms:012x}{tail:020x}")
937 }
938
939 fn message_line(id: &str, text: &str) -> String {
940 json!({"role": "user", "content": [{"type": "text", "text": text}], "id": id}).to_string()
941 }
942
943 fn make_sessions(root: &Path, names: &[String]) {
944 for name in names {
945 let dir = root.join(name);
946 std::fs::create_dir_all(dir.join("blobs")).unwrap();
947 std::fs::write(dir.join("trace.jsonl"), "{}\n").unwrap();
948 std::fs::write(dir.join("blobs").join("deadbeef"), b"body").unwrap();
949 }
950 }
951
952 fn names(root: &Path) -> Vec<String> {
953 let mut found: Vec<String> = std::fs::read_dir(root)
954 .unwrap()
955 .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
956 .collect();
957 found.sort();
958 found
959 }
960
961 const NOW: u64 = 1_756_400_000_000;
962
963 #[test]
967 fn max_sessions_leaves_exactly_the_newest_n_including_the_current() {
968 let root = tempfile::tempdir().unwrap();
969 let all: Vec<String> = (1..=6).map(|n| ses(NOW - (6 - n) * 1000, n)).collect();
970 make_sessions(root.path(), &all);
971
972 let pruned = prune_sessions(
973 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
974 None,
975 &all[5],
976 &TraceRetainConfig {
977 max_sessions: Some(3),
978 max_age_secs: None,
979 },
980 NOW,
981 );
982
983 assert_eq!(
984 names(root.path()),
985 vec![all[3].clone(), all[4].clone(), all[5].clone()]
986 );
987 let mut removed: Vec<String> = pruned.iter().map(|p| p.name.clone()).collect();
988 removed.sort();
989 assert_eq!(
990 removed,
991 vec![all[0].clone(), all[1].clone(), all[2].clone()]
992 );
993 assert!(pruned
994 .iter()
995 .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_SESSIONS));
996 for name in &all[..3] {
997 assert!(!root.path().join(name).exists(), "{name} went whole");
998 }
999 }
1000
1001 #[test]
1005 fn max_age_reads_the_id_and_never_the_filesystem() {
1006 let root = tempfile::tempdir().unwrap();
1007 let old_a = ses(NOW - 7_200_000, 1);
1008 let old_b = ses(NOW - 7_100_000, 2);
1009 let recent = ses(NOW - 60_000, 3);
1010 let current = ses(NOW, 4);
1011 make_sessions(
1012 root.path(),
1013 &[
1014 old_a.clone(),
1015 old_b.clone(),
1016 recent.clone(),
1017 current.clone(),
1018 ],
1019 );
1020
1021 let pruned = prune_sessions(
1022 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1023 None,
1024 ¤t,
1025 &TraceRetainConfig {
1026 max_sessions: None,
1027 max_age_secs: Some(900),
1028 },
1029 NOW,
1030 );
1031
1032 assert_eq!(pruned.len(), 2, "{pruned:?}");
1033 assert!(pruned
1034 .iter()
1035 .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_AGE));
1036 assert_eq!(names(root.path()), vec![recent, current]);
1037 }
1038
1039 #[test]
1042 fn both_trace_keys_keep_only_sessions_inside_both() {
1043 let root = tempfile::tempdir().unwrap();
1044 let old = [ses(NOW - 7_200_000, 1), ses(NOW - 7_100_000, 2)];
1045 let recent = [
1046 ses(NOW - 300_000, 3),
1047 ses(NOW - 200_000, 4),
1048 ses(NOW - 100_000, 5),
1049 ];
1050 let current = ses(NOW, 6);
1051 let mut all = old.to_vec();
1052 all.extend(recent.iter().cloned());
1053 all.push(current.clone());
1054 make_sessions(root.path(), &all);
1055
1056 let pruned = prune_sessions(
1057 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1058 None,
1059 ¤t,
1060 &TraceRetainConfig {
1061 max_sessions: Some(2),
1062 max_age_secs: Some(900),
1063 },
1064 NOW,
1065 );
1066
1067 assert_eq!(names(root.path()), vec![recent[2].clone(), current.clone()]);
1068 assert_eq!(pruned.len(), 4);
1069 for session in &pruned {
1070 let expected = if old.contains(&session.name) {
1071 crate::trace::RETENTION_REASON_MAX_AGE
1072 } else {
1073 crate::trace::RETENTION_REASON_MAX_SESSIONS
1074 };
1075 assert_eq!(session.reason, expected, "{session:?}");
1076 }
1077 }
1078
1079 #[test]
1082 fn nothing_at_or_after_the_current_session_id_is_a_candidate() {
1083 let root = tempfile::tempdir().unwrap();
1084 let current = ses(NOW, 5);
1085 let sibling = ses(NOW + 1, 6);
1087 let older = ses(NOW - 1000, 4);
1088 make_sessions(
1089 root.path(),
1090 &[current.clone(), sibling.clone(), older.clone()],
1091 );
1092
1093 let pruned = prune_sessions(
1094 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1095 None,
1096 ¤t,
1097 &TraceRetainConfig {
1098 max_sessions: Some(1),
1099 max_age_secs: Some(1),
1100 },
1101 NOW,
1102 );
1103
1104 assert_eq!(pruned.len(), 1);
1105 assert_eq!(pruned[0].name, older);
1106 assert_eq!(names(root.path()), vec![current.clone(), sibling]);
1107 assert!(root.path().join(¤t).join("trace.jsonl").exists());
1108 }
1109
1110 #[test]
1114 fn a_pruned_session_takes_its_records_and_its_created_work_folder() {
1115 let records = tempfile::tempdir().unwrap();
1116 let workdirs = tempfile::tempdir().unwrap();
1117 let all: Vec<String> = (1..=3).map(|n| ses(NOW - (3 - n) * 1000, n)).collect();
1118 make_sessions(records.path(), &all);
1119 for name in &all {
1120 std::fs::create_dir_all(workdirs.path().join(name).join("out")).unwrap();
1121 }
1122 let workdir_only = ses(NOW - 9_000, 9);
1123 std::fs::create_dir_all(workdirs.path().join(&workdir_only)).unwrap();
1124
1125 let pruned = prune_sessions(
1126 &crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
1127 Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
1128 &all[2],
1129 &TraceRetainConfig {
1130 max_sessions: Some(2),
1131 max_age_secs: None,
1132 },
1133 NOW,
1134 );
1135
1136 assert_eq!(pruned.len(), 1);
1137 assert_eq!(pruned[0].name, all[0]);
1138 assert_eq!(names(records.path()), vec![all[1].clone(), all[2].clone()]);
1139 let mut kept = vec![all[1].clone(), all[2].clone(), workdir_only];
1140 kept.sort();
1141 assert_eq!(names(workdirs.path()), kept);
1142 }
1143
1144 #[test]
1148 fn retention_prunes_a_created_work_folder_and_never_the_operators() {
1149 let records = tempfile::tempdir().unwrap();
1150 let operator = tempfile::tempdir().unwrap();
1151 let manifest_dir = tempfile::tempdir().unwrap();
1152 let old = ses(NOW - 2000, 1);
1153 let current = ses(NOW, 2);
1154 let policy = TraceRetainConfig {
1155 max_sessions: Some(1),
1156 max_age_secs: None,
1157 };
1158 let open = |path: &std::path::Path| crate::workdir_fs::WorkdirRoot::open(path).unwrap();
1159
1160 make_sessions(records.path(), &[old.clone(), current.clone()]);
1161 let planted = [
1162 operator.path().join(".murmur").join(&old),
1163 operator.path().join(&old),
1164 ];
1165 for dir in &planted {
1166 std::fs::create_dir_all(dir).unwrap();
1167 std::fs::write(dir.join("keep"), b"x").unwrap();
1168 }
1169 assert_eq!(
1170 created_workdirs(Some(operator.path()), manifest_dir.path()),
1171 None
1172 );
1173 let pruned = prune_sessions(
1174 &open(records.path()),
1175 created_workdirs(Some(operator.path()), manifest_dir.path())
1176 .map(|dir| open(&dir))
1177 .as_ref(),
1178 ¤t,
1179 &policy,
1180 NOW,
1181 );
1182 assert_eq!(pruned.len(), 1);
1183 assert_eq!(names(records.path()), vec![current.clone()]);
1184 for dir in &planted {
1185 assert!(dir.join("keep").is_file(), "{} survives", dir.display());
1186 }
1187
1188 make_sessions(records.path(), std::slice::from_ref(&old));
1189 let workdirs = created_workdirs(None, manifest_dir.path()).unwrap();
1190 assert_eq!(workdirs, manifest_dir.path().join("workdir"));
1191 std::fs::create_dir_all(workdirs.join(&old)).unwrap();
1192 std::fs::create_dir_all(workdirs.join(¤t)).unwrap();
1193 let pruned = prune_sessions(
1194 &open(records.path()),
1195 Some(&open(&workdirs)),
1196 ¤t,
1197 &policy,
1198 NOW,
1199 );
1200 assert_eq!(pruned.len(), 1);
1201 assert!(!workdirs.join(&old).exists());
1202 assert!(workdirs.join(¤t).exists());
1203 }
1204
1205 #[test]
1208 fn pruning_a_session_removes_its_temp_folder() {
1209 use std::os::unix::fs::PermissionsExt;
1210
1211 let root = tempfile::tempdir().unwrap();
1212 let old = ses(NOW - 2000, 1);
1213 let current = ses(NOW, 2);
1214 make_sessions(root.path(), &[old.clone(), current.clone()]);
1215 let folder = root
1216 .path()
1217 .join(&old)
1218 .join(crate::temp_dir::SESSION_TEMP_DIR_NAME);
1219 let read_only = folder.join("read-only");
1220 let sealed_shut = folder.join("sealed-shut");
1221 std::fs::create_dir_all(&read_only).unwrap();
1222 std::fs::create_dir_all(&sealed_shut).unwrap();
1223 std::fs::write(read_only.join("file"), b"left behind").unwrap();
1224 std::fs::set_permissions(&read_only, std::fs::Permissions::from_mode(0o555)).unwrap();
1225 std::fs::set_permissions(&sealed_shut, std::fs::Permissions::from_mode(0o000)).unwrap();
1226
1227 let pruned = prune_sessions(
1228 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1229 None,
1230 ¤t,
1231 &TraceRetainConfig {
1232 max_sessions: Some(1),
1233 max_age_secs: None,
1234 },
1235 NOW,
1236 );
1237
1238 assert_eq!(pruned.len(), 1, "{pruned:?}");
1239 assert_eq!(pruned[0].name, old);
1240 assert!(!root.path().join(&old).exists(), "the session went whole");
1241 assert_eq!(names(root.path()), vec![current]);
1242 }
1243
1244 #[test]
1247 fn a_link_in_place_of_a_created_work_folder_is_not_followed() {
1248 let records = tempfile::tempdir().unwrap();
1249 let workdirs = tempfile::tempdir().unwrap();
1250 let elsewhere = tempfile::tempdir().unwrap();
1251 std::fs::write(elsewhere.path().join("keep"), b"x").unwrap();
1252 let all: Vec<String> = (1..=2).map(|n| ses(NOW - (2 - n) * 1000, n)).collect();
1253 make_sessions(records.path(), &all);
1254 std::os::unix::fs::symlink(elsewhere.path(), workdirs.path().join(&all[0])).unwrap();
1255
1256 let pruned = prune_sessions(
1257 &crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
1258 Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
1259 &all[1],
1260 &TraceRetainConfig {
1261 max_sessions: Some(1),
1262 max_age_secs: None,
1263 },
1264 NOW,
1265 );
1266
1267 assert_eq!(pruned.len(), 1);
1268 assert!(elsewhere.path().join("keep").exists());
1269 }
1270
1271 #[test]
1274 fn an_empty_policy_removes_nothing() {
1275 let root = tempfile::tempdir().unwrap();
1276 let all = [ses(NOW - 9_000_000, 1), ses(NOW, 2)];
1277 make_sessions(root.path(), &all);
1278 assert!(prune_sessions(
1279 &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1280 None,
1281 &all[1],
1282 &TraceRetainConfig {
1283 max_sessions: None,
1284 max_age_secs: None
1285 },
1286 NOW
1287 )
1288 .is_empty());
1289 assert_eq!(names(root.path()), all.to_vec());
1290 }
1291
1292 fn write_record(dir: &Path, capsule: Option<&str>, count: u64) -> PathBuf {
1296 std::fs::create_dir_all(dir).unwrap();
1297 let path = dir.join(RECORD_FILE_NAME);
1298 let mut contents = String::new();
1299 if let Some(capsule) = capsule {
1300 contents.push_str(
1301 &header_line(&RecordHeader {
1302 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
1303 capsule: capsule.to_string(),
1304 created_ms: NOW,
1305 truncated: None,
1306 })
1307 .unwrap(),
1308 );
1309 }
1310 for i in 1..=count {
1311 contents.push_str(&message_line(
1312 &msg(NOW - (count - i) * 1000, i),
1313 &format!("m{i}"),
1314 ));
1315 contents.push('\n');
1316 }
1317 std::fs::write(&path, contents).unwrap();
1318 path
1319 }
1320
1321 fn image_line(id: &str, shas: &[&str]) -> String {
1323 let mut content = vec![serde_json::json!({"type": "text", "text": "shot"})];
1324 for sha in shas {
1325 content.push(serde_json::json!({
1326 "type": "image",
1327 "source": {"type": "file", "sha256": sha, "media_type": "image/png", "bytes": 4},
1328 }));
1329 }
1330 serde_json::json!({"id": id, "role": "tool", "content": content}).to_string()
1331 }
1332
1333 #[test]
1336 fn truncation_removes_blobs_no_surviving_line_names() {
1337 let dir = tempfile::tempdir().unwrap();
1338 let path = dir.path().join(RECORD_FILE_NAME);
1339 let (old, shared, new) = ("a".repeat(64), "b".repeat(64), "c".repeat(64));
1340 let lines = [
1341 image_line(&msg(NOW - 3000, 1), &[&old, &shared]),
1342 image_line(&msg(NOW - 2000, 2), &[&old]),
1343 image_line(&msg(NOW - 1000, 3), &[&shared]),
1344 image_line(&msg(NOW, 4), &[&new]),
1345 ];
1346 std::fs::write(&path, lines.join("\n") + "\n").unwrap();
1347 let blobs = dir.path().join("blobs");
1348 std::fs::create_dir(&blobs).unwrap();
1349 for name in [&old, &shared, &new, &"notes.txt".to_string()] {
1350 std::fs::write(blobs.join(name), b"\x89PNG").unwrap();
1351 }
1352
1353 let outcome = truncate_record(&path, 2, "capsule").unwrap();
1354 assert_eq!(outcome.dropped, 2);
1355 assert_eq!(outcome.blobs_removed, 1);
1356 assert!(!blobs.join(&old).exists());
1357 assert!(blobs.join(&shared).exists(), "a kept line still names it");
1358 assert!(blobs.join(&new).exists());
1359 assert!(blobs.join("notes.txt").exists());
1360 }
1361
1362 #[test]
1365 fn truncation_reseals_the_record() {
1366 use crate::record_seal::{self, PublicKey, Verdict};
1367 let home = tempfile::tempdir().unwrap();
1368 record_seal::set_test_seal_home(Some(home.path().to_path_buf()));
1369 let dir = tempfile::tempdir().unwrap();
1370 let context = dir.path().join("capsule/ctx_1");
1371 std::fs::create_dir_all(&context).unwrap();
1372 let path = write_record(&context, Some("capsule"), 5);
1373 let before = std::fs::read_to_string(&path).unwrap();
1374 truncate_record(&path, 2, "capsule").unwrap();
1375 let after = std::fs::read_to_string(&path).unwrap();
1376 for kept in after
1377 .lines()
1378 .filter(|line| parse_message_line(line).is_some())
1379 {
1380 assert!(before.lines().any(|line| line == kept), "{kept}");
1381 }
1382 let root = crate::workdir_fs::WorkdirRoot::open(&context).unwrap();
1383 let seal = record_seal::read_seal(&root, crate::conversation::RECORD_FILE_NAME)
1384 .unwrap()
1385 .expect("the truncation sealed the record");
1386 assert_eq!(seal.stream, "conversation:capsule/ctx_1");
1387 let key = PublicKey::load(&record_seal::public_key_path(home.path())).unwrap();
1388 let report = record_seal::verify_listed(
1389 &seal.stream,
1390 after.as_bytes(),
1391 false,
1392 Some(&seal),
1393 Some(&key),
1394 None,
1395 );
1396 assert_eq!(report.verdict, Verdict::Verified, "{report:#?}");
1397 }
1398
1399 #[test]
1401 fn truncation_without_blobs_removes_nothing() {
1402 let dir = tempfile::tempdir().unwrap();
1403 let path = write_record(dir.path(), Some("capsule"), 5);
1404 let outcome = truncate_record(&path, 2, "capsule").unwrap();
1405 assert_eq!(outcome.dropped, 3);
1406 assert_eq!(outcome.blobs_removed, 0);
1407 assert!(!dir.path().join("blobs").exists());
1408 }
1409
1410 #[test]
1413 fn max_messages_leaves_the_newest_n_with_their_ids_intact() {
1414 let dir = tempfile::tempdir().unwrap();
1415 let path = write_record(dir.path(), Some("capsule"), 10);
1416 let before: Vec<String> = std::fs::read_to_string(&path)
1417 .unwrap()
1418 .lines()
1419 .filter_map(parse_message_line)
1420 .map(|m| m["id"].as_str().unwrap().to_string())
1421 .collect();
1422
1423 let outcome = truncate_record(&path, 3, "capsule").unwrap();
1424 assert_eq!(outcome.dropped, 7);
1425 assert_eq!(outcome.kept, 3);
1426 assert_eq!(outcome.last_dropped_id, before[6]);
1427 assert_eq!(outcome.oldest_surviving_id, before[7]);
1428
1429 let raw = std::fs::read_to_string(&path).unwrap();
1430 let lines: Vec<&str> = raw.lines().collect();
1431 assert_eq!(
1432 lines.len(),
1433 4,
1434 "a header plus exactly three messages: {lines:?}"
1435 );
1436 for line in &lines {
1437 serde_json::from_str::<serde_json::Value>(line).expect("every line is JSON");
1438 }
1439 let header = parse_header(lines[0]).expect("the first line is the header");
1440 assert_eq!(header.capsule, "capsule");
1441 let marker = header.truncated.expect("the header records the drop");
1442 assert_eq!(marker.dropped, 7);
1443 assert_eq!(marker.last_dropped_id, before[6]);
1444 assert_eq!(marker.oldest_surviving_id, before[7]);
1445
1446 let after: Vec<String> = lines[1..]
1447 .iter()
1448 .map(|line| {
1449 parse_message_line(line).expect("a message with a string role")["id"]
1450 .as_str()
1451 .unwrap()
1452 .to_string()
1453 })
1454 .collect();
1455 assert_eq!(after, before[7..], "surviving ids are the bytes they were");
1456 }
1457
1458 #[test]
1461 fn a_header_line_is_not_counted_as_a_message() {
1462 let dir = tempfile::tempdir().unwrap();
1463 let headerless = write_record(&dir.path().join("a"), None, 4);
1464 let owned = write_record(&dir.path().join("b"), Some("capsule"), 4);
1465 assert_eq!(count_messages(&headerless), 4);
1466 assert_eq!(count_messages(&owned), 4);
1467 }
1468
1469 #[test]
1472 fn a_second_truncation_accumulates_the_dropped_count() {
1473 let dir = tempfile::tempdir().unwrap();
1474 let path = write_record(dir.path(), Some("capsule"), 10);
1475 truncate_record(&path, 6, "capsule").unwrap();
1476 truncate_record(&path, 2, "capsule").unwrap();
1477 let marker = read_header(&path).unwrap().truncated.unwrap();
1478 assert_eq!(marker.dropped, 8);
1479 assert_eq!(count_messages(&path), 2);
1480 }
1481
1482 #[test]
1483 fn truncation_rewrite_keeps_the_record_owner_only() {
1484 use std::os::unix::fs::PermissionsExt;
1485 let dir = tempfile::tempdir().unwrap();
1486 let path = write_record(dir.path(), Some("capsule"), 5);
1487 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
1488
1489 truncate_record(&path, 2, "capsule").unwrap();
1490
1491 assert_eq!(count_messages(&path), 2);
1492 assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
1493 }
1494
1495 #[test]
1497 fn a_record_inside_the_limit_is_not_rewritten() {
1498 let dir = tempfile::tempdir().unwrap();
1499 let path = write_record(dir.path(), Some("capsule"), 3);
1500 let before = std::fs::read(&path).unwrap();
1501 let outcome = truncate_record(&path, 3, "capsule").unwrap();
1502 assert_eq!(outcome.dropped, 0);
1503 assert_eq!(std::fs::read(&path).unwrap(), before);
1504 }
1505
1506 #[test]
1510 fn a_staged_truncation_leaves_the_original_intact_until_the_rename() {
1511 let dir = tempfile::tempdir().unwrap();
1512 let path = write_record(dir.path(), Some("capsule"), 10);
1513 let before = std::fs::read(&path).unwrap();
1514
1515 let (staged, outcome) = stage_truncation(&path, 3, "capsule", NOW).unwrap().unwrap();
1516 assert_eq!(outcome.dropped, 7);
1517
1518 let temp = staged.temp_path().to_path_buf();
1519 assert_eq!(
1520 temp.parent(),
1521 path.parent(),
1522 "the temp file is in the record's own directory, so the rename is atomic"
1523 );
1524 assert_ne!(temp.file_name().unwrap(), RECORD_FILE_NAME);
1525
1526 assert_eq!(
1527 std::fs::read(&path).unwrap(),
1528 before,
1529 "the original is byte-identical before the rename"
1530 );
1531 assert_eq!(count_messages(&path), 10);
1532 for line in std::fs::read_to_string(&path).unwrap().lines() {
1533 serde_json::from_str::<serde_json::Value>(line).expect("the original parses whole");
1534 }
1535
1536 drop(staged);
1538 assert!(!temp.exists());
1539 assert_eq!(std::fs::read(&path).unwrap(), before);
1540 assert_eq!(count_messages(&path), 10);
1541 }
1542
1543 #[test]
1549 fn record_pruning_skips_another_capsules_records_and_every_unowned_one() {
1550 let root = tempfile::tempdir().unwrap();
1551 write_record(&root.path().join("mine"), Some("mine"), 3);
1552 write_record(&root.path().join("theirs"), Some("theirs"), 3);
1553 write_record(&root.path().join("unowned"), None, 3);
1554 write_record(&root.path().join("current"), Some("mine"), 3);
1555
1556 let far_future = now_ms() + 3_600_000;
1558 let pruned = prune_records(
1559 root.path(),
1560 "mine",
1561 Some("current"),
1562 &ContextRetainConfig {
1563 max_messages: None,
1564 max_age_secs: Some(1),
1565 },
1566 far_future,
1567 );
1568
1569 assert_eq!(pruned.len(), 1, "{pruned:?}");
1570 assert_eq!(pruned[0].context_id, "mine");
1571 assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
1572 assert_eq!(names(root.path()), vec!["current", "theirs", "unowned"]);
1573 }
1574
1575 #[test]
1578 fn a_record_written_inside_the_window_is_kept() {
1579 let root = tempfile::tempdir().unwrap();
1580 write_record(&root.path().join("fresh"), Some("mine"), 3);
1581 let pruned = prune_records(
1582 root.path(),
1583 "mine",
1584 None,
1585 &ContextRetainConfig {
1586 max_messages: None,
1587 max_age_secs: Some(90 * 86_400),
1588 },
1589 now_ms(),
1590 );
1591 assert!(pruned.is_empty());
1592 assert_eq!(names(root.path()), vec!["fresh"]);
1593 }
1594
1595 #[test]
1598 fn max_messages_alone_removes_no_record() {
1599 let root = tempfile::tempdir().unwrap();
1600 write_record(&root.path().join("ctx"), Some("mine"), 100);
1601 assert!(prune_records(
1602 root.path(),
1603 "mine",
1604 None,
1605 &ContextRetainConfig {
1606 max_messages: Some(1),
1607 max_age_secs: None
1608 },
1609 now_ms() + 3_600_000,
1610 )
1611 .is_empty());
1612 assert_eq!(names(root.path()), vec!["ctx"]);
1613 }
1614
1615 fn frm(ms: u64, tail: u64) -> FormationId {
1619 FormationId::parse(&format!("frm_{ms:012x}{tail:020x}")).unwrap()
1620 }
1621
1622 fn retain(max_sessions: Option<u32>, max_age_secs: Option<u64>) -> TraceRetainConfig {
1623 TraceRetainConfig {
1624 max_sessions,
1625 max_age_secs,
1626 }
1627 }
1628
1629 fn make_formation(root: &Path, id: &FormationId, marker: Option<&str>) -> PathBuf {
1632 let dir = root.join(id.as_str());
1633 let session = dir.join("worker/.murmur").join(ses(NOW, 1));
1634 std::fs::create_dir_all(&session).unwrap();
1635 std::fs::write(session.join("trace.jsonl"), "{}\n").unwrap();
1636 if let Some(marker) = marker {
1637 std::fs::write(
1638 dir.join(crate::formation_launch::FORMATION_MARKER_FILE),
1639 marker,
1640 )
1641 .unwrap();
1642 }
1643 dir
1644 }
1645
1646 fn marker_naming(project: &Path) -> String {
1647 serde_json::to_string(&FormationMarker {
1648 project_dir: project.to_path_buf(),
1649 launcher_pid: 1,
1650 launcher_start: String::new(),
1651 })
1652 .unwrap()
1653 }
1654
1655 fn make_owned(root: &Path, project: &Path, ids: &[&FormationId]) {
1657 for id in ids {
1658 make_formation(root, id, Some(&marker_naming(project)));
1659 }
1660 }
1661
1662 fn removed(pruned: &[PrunedFormation]) -> Vec<(FormationId, &'static str)> {
1663 pruned
1664 .iter()
1665 .map(|p| (p.formation_id.clone(), p.reason))
1666 .collect()
1667 }
1668
1669 const DAY_MS: u64 = 86_400_000;
1670
1671 #[test]
1674 fn prune_formations_never_removes_the_current_or_a_later_formation() {
1675 let root = tempfile::tempdir().unwrap();
1676 let project = tempfile::tempdir().unwrap();
1677 let old = frm(NOW - 3 * DAY_MS, 1);
1678 let current = frm(NOW - 2 * DAY_MS, 2);
1679 let later = frm(NOW - 2 * DAY_MS + 1000, 3);
1680 make_owned(root.path(), project.path(), &[&old, ¤t, &later]);
1681
1682 let pruned = prune_formations(
1683 root.path(),
1684 ¤t,
1685 project.path(),
1686 &retain(Some(1), Some(1)),
1687 NOW,
1688 |_, _| false,
1689 );
1690
1691 assert_eq!(
1692 removed(&pruned),
1693 [(old.clone(), crate::trace::RETENTION_REASON_MAX_AGE)]
1694 );
1695 assert_eq!(pruned[0].path, root.path().join(old.as_str()));
1696 assert_eq!(
1697 names(root.path()),
1698 vec![current.as_str().to_string(), later.as_str().to_string()]
1699 );
1700 }
1701
1702 #[test]
1705 fn prune_formations_keeps_the_newest_n_with_the_current_counted() {
1706 let root = tempfile::tempdir().unwrap();
1707 let project = tempfile::tempdir().unwrap();
1708 let all: Vec<FormationId> = (1..=5).map(|n| frm(NOW - (5 - n) * 1000, n)).collect();
1709 make_owned(root.path(), project.path(), &all.iter().collect::<Vec<_>>());
1710
1711 let pruned = prune_formations(
1712 root.path(),
1713 &all[3],
1714 project.path(),
1715 &retain(Some(3), None),
1716 NOW,
1717 |_, _| false,
1718 );
1719
1720 let reason = crate::trace::RETENTION_REASON_MAX_SESSIONS;
1721 assert_eq!(
1722 removed(&pruned),
1723 [(all[1].clone(), reason), (all[0].clone(), reason)]
1724 );
1725 assert_eq!(
1726 names(root.path()),
1727 all[2..]
1728 .iter()
1729 .map(|id| id.as_str().to_string())
1730 .collect::<Vec<_>>()
1731 );
1732 }
1733
1734 #[test]
1737 fn prune_formations_reads_age_from_the_id() {
1738 let root = tempfile::tempdir().unwrap();
1739 let project = tempfile::tempdir().unwrap();
1740 let old_a = frm(NOW - 7_200_000, 1);
1741 let old_b = frm(NOW - 7_100_000, 2);
1742 let recent = frm(NOW - 60_000, 3);
1743 let current = frm(NOW, 4);
1744 make_owned(
1745 root.path(),
1746 project.path(),
1747 &[&old_a, &old_b, &recent, ¤t],
1748 );
1749
1750 let pruned = prune_formations(
1751 root.path(),
1752 ¤t,
1753 project.path(),
1754 &retain(None, Some(900)),
1755 NOW,
1756 |_, _| false,
1757 );
1758
1759 let reason = crate::trace::RETENTION_REASON_MAX_AGE;
1760 assert_eq!(removed(&pruned), [(old_b, reason), (old_a, reason)]);
1761 assert_eq!(
1762 names(root.path()),
1763 vec![recent.as_str().to_string(), current.as_str().to_string()]
1764 );
1765 }
1766
1767 #[test]
1770 fn prune_formations_attributes_reasons_age_first() {
1771 let root = tempfile::tempdir().unwrap();
1772 let project = tempfile::tempdir().unwrap();
1773 let old = frm(NOW - 7_200_000, 1);
1774 let recent = frm(NOW - 60_000, 2);
1775 let current = frm(NOW, 3);
1776 make_owned(root.path(), project.path(), &[&old, &recent, ¤t]);
1777
1778 let pruned = prune_formations(
1779 root.path(),
1780 ¤t,
1781 project.path(),
1782 &retain(Some(1), Some(900)),
1783 NOW,
1784 |_, _| false,
1785 );
1786
1787 assert_eq!(
1788 removed(&pruned),
1789 [
1790 (recent, crate::trace::RETENTION_REASON_MAX_SESSIONS),
1791 (old, crate::trace::RETENTION_REASON_MAX_AGE),
1792 ]
1793 );
1794 }
1795
1796 #[test]
1799 fn prune_formations_skips_a_live_directory_and_keeps_its_rank() {
1800 let root = tempfile::tempdir().unwrap();
1801 let project = tempfile::tempdir().unwrap();
1802 let older = frm(NOW - 3000, 1);
1803 let live = frm(NOW - 2000, 2);
1804 let current = frm(NOW, 3);
1805 make_owned(root.path(), project.path(), &[&older, &live, ¤t]);
1806
1807 let asked = std::cell::RefCell::new(Vec::new());
1808 let pruned = prune_formations(
1809 root.path(),
1810 ¤t,
1811 project.path(),
1812 &retain(Some(1), None),
1813 NOW,
1814 |id, marker| {
1815 assert_eq!(marker.launcher_pid, 1);
1816 asked.borrow_mut().push(id.clone());
1817 *id == live
1818 },
1819 );
1820
1821 assert_eq!(
1822 removed(&pruned),
1823 [(older.clone(), crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1824 );
1825 assert_eq!(*asked.borrow(), [live.clone(), older]);
1826 assert_eq!(
1827 names(root.path()),
1828 vec![live.as_str().to_string(), current.as_str().to_string()]
1829 );
1830
1831 let kept = frm(NOW - 2500, 4);
1833 make_owned(root.path(), project.path(), &[&kept]);
1834 let pruned = prune_formations(
1835 root.path(),
1836 ¤t,
1837 project.path(),
1838 &retain(Some(3), None),
1839 NOW,
1840 |id, _| *id == live,
1841 );
1842 assert!(pruned.is_empty(), "{pruned:?}");
1843 let pruned = prune_formations(
1844 root.path(),
1845 ¤t,
1846 project.path(),
1847 &retain(Some(2), None),
1848 NOW,
1849 |id, _| *id == live,
1850 );
1851 assert_eq!(
1852 removed(&pruned),
1853 [(kept, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1854 );
1855 }
1856
1857 #[test]
1860 fn prune_formations_never_ranks_or_removes_unowned_or_foreign_directories() {
1861 let root = tempfile::tempdir().unwrap();
1862 let project = tempfile::tempdir().unwrap();
1863 let other = tempfile::tempdir().unwrap();
1864 let owned_old = frm(NOW - 3 * DAY_MS, 1);
1865 let owned_kept = frm(NOW - 2 * DAY_MS, 2);
1866 let unmarked = frm(NOW - 2 * DAY_MS + 1, 3);
1867 let garbled = frm(NOW - 2 * DAY_MS + 2, 4);
1868 let foreign = frm(NOW - 2 * DAY_MS + 3, 5);
1869 let current = frm(NOW, 6);
1870 make_owned(
1871 root.path(),
1872 project.path(),
1873 &[&owned_old, &owned_kept, ¤t],
1874 );
1875 make_formation(root.path(), &unmarked, None);
1876 make_formation(root.path(), &garbled, Some("not json {"));
1877 make_formation(root.path(), &foreign, Some(&marker_naming(other.path())));
1878
1879 let pruned = prune_formations(
1880 root.path(),
1881 ¤t,
1882 project.path(),
1883 &retain(Some(2), None),
1884 NOW,
1885 |_, _| false,
1886 );
1887 assert_eq!(
1888 removed(&pruned),
1889 [(owned_old, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1890 );
1891
1892 let pruned = prune_formations(
1893 root.path(),
1894 ¤t,
1895 project.path(),
1896 &retain(Some(1), Some(DAY_MS / 1000)),
1897 NOW,
1898 |_, _| false,
1899 );
1900 assert_eq!(
1901 removed(&pruned),
1902 [(owned_kept, crate::trace::RETENTION_REASON_MAX_AGE)]
1903 );
1904 for id in [&unmarked, &garbled, &foreign, ¤t] {
1905 assert!(root.path().join(id.as_str()).is_dir(), "{id}");
1906 }
1907 }
1908
1909 #[test]
1912 fn prune_formations_ignores_a_symlink_a_regular_file_and_a_name_that_is_not_an_id() {
1913 let root = tempfile::tempdir().unwrap();
1914 let elsewhere = tempfile::tempdir().unwrap();
1915 let project = tempfile::tempdir().unwrap();
1916 let marker = marker_naming(project.path());
1917 let target = make_formation(elsewhere.path(), &frm(NOW - 3 * DAY_MS, 1), Some(&marker));
1918 let linked = frm(NOW - 3 * DAY_MS, 2);
1919 std::os::unix::fs::symlink(&target, root.path().join(linked.as_str())).unwrap();
1920 std::fs::write(root.path().join(frm(NOW - 3 * DAY_MS, 3).as_str()), &marker).unwrap();
1921 for name in ["not-a-formation", "frm_ABC", "frm_0123"] {
1922 std::fs::create_dir(root.path().join(name)).unwrap();
1923 std::fs::write(
1924 root.path()
1925 .join(name)
1926 .join(crate::formation_launch::FORMATION_MARKER_FILE),
1927 &marker,
1928 )
1929 .unwrap();
1930 }
1931 let before = names(root.path());
1932
1933 let pruned = prune_formations(
1934 root.path(),
1935 &frm(NOW, 9),
1936 project.path(),
1937 &retain(Some(1), Some(1)),
1938 NOW,
1939 |_, _| false,
1940 );
1941
1942 assert!(pruned.is_empty(), "{pruned:?}");
1943 assert_eq!(names(root.path()), before);
1944 assert!(target.join("worker/.murmur").is_dir());
1945 }
1946
1947 #[test]
1949 fn prune_formations_with_neither_key_or_no_directory_removes_nothing() {
1950 let root = tempfile::tempdir().unwrap();
1951 let project = tempfile::tempdir().unwrap();
1952 let old = frm(NOW - 3 * DAY_MS, 1);
1953 make_owned(root.path(), project.path(), &[&old]);
1954
1955 let pruned = prune_formations(
1956 root.path(),
1957 &frm(NOW, 2),
1958 project.path(),
1959 &retain(None, None),
1960 NOW,
1961 |_, _| panic!("nothing is a candidate without a policy"),
1962 );
1963 assert!(pruned.is_empty());
1964 assert!(root.path().join(old.as_str()).is_dir());
1965
1966 assert!(prune_formations(
1967 &root.path().join("missing"),
1968 &frm(NOW, 2),
1969 project.path(),
1970 &retain(Some(1), Some(1)),
1971 NOW,
1972 |_, _| false,
1973 )
1974 .is_empty());
1975 }
1976
1977 #[test]
1980 fn prune_formations_compares_project_directories_canonically() {
1981 let root = tempfile::tempdir().unwrap();
1982 let project = tempfile::tempdir().unwrap();
1983 let links = tempfile::tempdir().unwrap();
1984 let link = links.path().join("project");
1985 std::os::unix::fs::symlink(project.path(), &link).unwrap();
1986 let old = frm(NOW - 3 * DAY_MS, 1);
1987 make_owned(
1988 root.path(),
1989 &std::fs::canonicalize(project.path()).unwrap(),
1990 &[&old],
1991 );
1992
1993 let pruned = prune_formations(
1994 root.path(),
1995 &frm(NOW, 2),
1996 &link,
1997 &retain(None, Some(1)),
1998 NOW,
1999 |_, _| false,
2000 );
2001 assert_eq!(
2002 removed(&pruned),
2003 [(old, crate::trace::RETENTION_REASON_MAX_AGE)]
2004 );
2005 }
2006
2007 #[test]
2009 fn prune_formations_leaves_a_directory_it_could_not_remove_out_of_the_result() {
2010 use std::os::unix::fs::PermissionsExt;
2011
2012 let root = tempfile::tempdir().unwrap();
2013 let project = tempfile::tempdir().unwrap();
2014 let stuck = frm(NOW - 3 * DAY_MS, 1);
2015 let gone = frm(NOW - 3 * DAY_MS + 1, 2);
2016 make_owned(root.path(), project.path(), &[&stuck, &gone]);
2017 let stuck_dir = root.path().join(stuck.as_str());
2018 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o500)).unwrap();
2019 if std::fs::write(stuck_dir.join("probe"), b"").is_ok() {
2021 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
2022 return;
2023 }
2024
2025 let pruned = prune_formations(
2026 root.path(),
2027 &frm(NOW, 3),
2028 project.path(),
2029 &retain(None, Some(1)),
2030 NOW,
2031 |_, _| false,
2032 );
2033 std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
2034
2035 assert_eq!(
2036 removed(&pruned),
2037 [(gone, crate::trace::RETENTION_REASON_MAX_AGE)]
2038 );
2039 assert!(stuck_dir.is_dir());
2040 }
2041
2042 fn harness_entry_json(kind: &str, session_id: &str) -> String {
2045 json!({"type": kind, "session_id": session_id, "harness": "h", "driver": "d",
2046 "created_ms": 1, "updated_ms": 2})
2047 .to_string()
2048 }
2049
2050 #[test]
2053 fn the_context_finder_sees_either_file_in_every_store() {
2054 let root = tempfile::tempdir().unwrap();
2055 let dir = |store: &str, context: &str| {
2056 let path = root.path().join(store).join(context);
2057 std::fs::create_dir_all(&path).unwrap();
2058 path
2059 };
2060 std::fs::write(dir("b", "ctx").join(RECORD_FILE_NAME), "").unwrap();
2061 std::fs::write(
2062 dir("a", "ctx").join(HARNESS_SESSION_FILE_NAME),
2063 harness_entry_json(crate::harness_session::HARNESS_SESSION_TYPE, "s"),
2064 )
2065 .unwrap();
2066 dir("c", "ctx");
2067 std::fs::write(dir(".hidden", "ctx").join(RECORD_FILE_NAME), "").unwrap();
2068 std::fs::write(dir("a", ".ctx").join(RECORD_FILE_NAME), "").unwrap();
2069
2070 let found = find_context_dirs_in(root.path(), "ctx");
2071 let stores: Vec<&str> = found.iter().map(|dir| dir.record.as_str()).collect();
2072 assert_eq!(stores, ["a", "b"]);
2073 assert!(found.iter().all(|dir| dir.context_id == "ctx"));
2074 assert_eq!(found[1].path, root.path().join("b").join("ctx"));
2075 assert!(find_context_dirs_in(root.path(), ".ctx").is_empty());
2076 assert!(find_context_dirs_in(root.path(), "other").is_empty());
2077 assert!(find_context_dirs_in(&root.path().join("absent"), "ctx").is_empty());
2078 }
2079
2080 #[test]
2083 fn the_context_reader_reads_as_the_runtime_reads() {
2084 let dir = tempfile::tempdir().unwrap();
2085 let header = RecordHeader {
2086 kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
2087 capsule: "owner".to_string(),
2088 created_ms: 7,
2089 truncated: None,
2090 };
2091 let first = json!({"role": "user", "content": "hi", "id": "msg_1", "kept": {"x": 1}});
2092 let second = json!({"role": "assistant", "content": [{"type": "text", "text": "yo"}]});
2093 std::fs::write(
2094 dir.path().join(RECORD_FILE_NAME),
2095 format!(
2096 "{}{first}\n\n{second}\n{}\n{{\"role\": \"us",
2097 header_line(&header).unwrap(),
2098 header_line(&header).unwrap().trim_end()
2099 ),
2100 )
2101 .unwrap();
2102
2103 let contents = read_context_dir(dir.path()).unwrap();
2104 assert_eq!(contents.header, Some(header.clone()));
2105 assert_eq!(contents.messages, vec![first.clone(), second]);
2106 assert_eq!(contents.harness_session, None);
2107
2108 std::fs::write(
2109 dir.path().join(RECORD_FILE_NAME),
2110 format!("{first}\n{}", header_line(&header).unwrap()),
2111 )
2112 .unwrap();
2113 let contents = read_context_dir(dir.path()).unwrap();
2114 assert_eq!(contents.header, None, "a header below line 1 is no header");
2115 assert_eq!(contents.messages, vec![first]);
2116
2117 let empty = tempfile::tempdir().unwrap();
2118 let contents = read_context_dir(empty.path()).unwrap();
2119 assert_eq!(contents.header, None);
2120 assert!(contents.messages.is_empty());
2121
2122 std::fs::create_dir(empty.path().join(RECORD_FILE_NAME)).unwrap();
2123 assert!(
2124 read_context_dir(empty.path()).is_err(),
2125 "an unreadable record"
2126 );
2127 }
2128
2129 #[test]
2132 fn the_context_reader_reads_no_session_where_the_runtime_does() {
2133 let dir = tempfile::tempdir().unwrap();
2134 let file = dir.path().join(HARNESS_SESSION_FILE_NAME);
2135 assert_eq!(read_context_dir(dir.path()).unwrap().harness_session, None);
2136
2137 let session_type = crate::harness_session::HARNESS_SESSION_TYPE;
2138 for (body, expected) in [
2139 ("{not json".to_string(), false),
2140 (harness_entry_json("another-tool", "s"), false),
2141 (harness_entry_json(session_type, ""), false),
2142 (harness_entry_json(session_type, "s"), true),
2143 ] {
2144 std::fs::write(&file, &body).unwrap();
2145 let session = read_context_dir(dir.path()).unwrap().harness_session;
2146 assert_eq!(session.is_some(), expected, "{body}");
2147 if let Some(session) = session {
2148 assert_eq!(
2149 (session.session_id.as_str(), session.harness.as_str()),
2150 ("s", "h")
2151 );
2152 assert_eq!((session.created_ms, session.updated_ms), (1, 2));
2153 }
2154 }
2155 }
2156}