Skip to main content

capsule_runtime/
retention.rs

1//! Retention: what bounds the three stores that grow, and what it costs to bound them.
2//!
3//! Three stores, two operations, because they are different shapes:
4//!
5//! * A **session directory** — `<workdir>/.murmur/<ses_…>/` — is an independent unit. Nothing
6//!   references it and nothing spans two of them, so it is pruned by being deleted whole, taking
7//!   its `trace.jsonl` and its `blobs/` with it.
8//! * A **formation directory** — `~/.murmur/formations/<frm_…>/` — holds one ended formation's
9//!   peer directories, and with them every peer's sessions. It is the other half of the evidence
10//!   whose first half is the entry member's session in the project, so it is pruned whole, by the
11//!   entry member's `trace.retain`, and only by the launcher of a later formation from the same
12//!   project once every member of that later one is reaped.
13//! * A **conversation record** — `~/.murmur/conversations/<record>/<ctx>/conversation.jsonl` — is
14//!   one unit that grows. Deleting it destroys the conversation an operator wanted kept, so it is
15//!   pruned by truncating its front: the oldest messages go, the recent ones stay, and every
16//!   surviving message keeps the id it has always had.
17//!
18//! Three properties hold across everything below:
19//!
20//! * **No defaults.** Every entry point takes the policy it enforces. An absent policy is an
21//!   absent call: nothing here has a fallback that deletes.
22//! * **No clock of its own.** Every prune function takes `now_ms`, so age is a fixture's to supply
23//!   and a test never sleeps.
24//! * **No trace of its own.** Every prune function returns what it removed and writes nothing;
25//!   the caller holds the [`crate::trace::TraceWriter`] and records the deletion against the
26//!   session that performed it, or reports it as the launcher that performed it.
27//!
28//! Session and formation age is read out of the `ses_` or `frm_` id — a uuid v7 whose first 12
29//! hex characters are a millisecond timestamp — so both policies are computed from one directory
30//! listing with no `stat`. Record age is the mtime of `conversation.jsonl`, because a context id
31//! is not necessarily a uuid and "untouched since" is the useful question for a record that spans
32//! weeks.
33
34use 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/// A session directory that was removed, and why.
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct PrunedSession {
51    /// The `ses_…` directory name, as it appeared under the sessions root.
52    pub name: String,
53    /// [`crate::trace::RETENTION_REASON_MAX_SESSIONS`] or
54    /// [`crate::trace::RETENTION_REASON_MAX_AGE`].
55    pub reason: &'static str,
56}
57
58/// A formation directory that was removed, and why.
59#[derive(Debug, Clone, PartialEq, Eq)]
60pub struct PrunedFormation {
61    /// The formation whose directory went.
62    pub formation_id: FormationId,
63    /// The directory that was removed, whole: `<formations dir>/<frm_id>`.
64    pub path: PathBuf,
65    /// [`crate::trace::RETENTION_REASON_MAX_AGE`] or
66    /// [`crate::trace::RETENTION_REASON_MAX_SESSIONS`].
67    pub reason: &'static str,
68}
69
70/// A context directory that was removed, and why.
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct PrunedRecord {
73    /// The context id, which is the directory name under the record root.
74    pub context_id: String,
75    /// The directory that was removed, whole.
76    pub path: PathBuf,
77    /// Always [`crate::trace::RETENTION_REASON_MAX_AGE`]: a record is never removed on a count.
78    pub reason: &'static str,
79}
80
81/// What one truncation dropped and what it left.
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct TruncationOutcome {
84    /// Messages this truncation removed from the front. `0` means the file was left untouched.
85    pub dropped: u64,
86    /// Messages left in the record.
87    pub kept: u64,
88    /// The `msg_` id of the oldest surviving message, empty when nothing survives with an id.
89    pub oldest_surviving_id: String,
90    /// The `msg_` id of the newest dropped message, empty when nothing dropped carried one.
91    pub last_dropped_id: String,
92    /// Image blobs removed from the record's `blobs/` because no surviving line refers to them.
93    pub blobs_removed: u64,
94}
95
96/// Where one message id turned up, and in what state.
97#[derive(Debug, Clone, PartialEq, Eq)]
98pub enum MessageStatus {
99    /// The id is a line in the record. `position` is 1-based, oldest first.
100    Present { position: u64, total: u64 },
101    /// The id is not a line, the header carries a truncation marker, and the id was minted at or
102    /// before the newest dropped message. A reference an artifact stored is dangling because the
103    /// record was trimmed — not because the id was never real.
104    Truncated {
105        dropped: u64,
106        oldest_surviving_id: String,
107    },
108    /// Anything else: no such id, in a record that has dropped nothing that old.
109    Unknown,
110}
111
112/// One record a message id was looked for in, and what that record had to say about it.
113#[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/// One context's record, as `mur conversation ls` reports it.
122#[derive(Debug, Clone, PartialEq, Eq)]
123pub struct RecordSummary {
124    /// Record store: the directory under `~/.murmur/conversations/`.
125    pub record: String,
126    /// Context id: the directory under the record store.
127    pub context_id: String,
128    pub path: PathBuf,
129    /// Message lines. The header line is not one and is never counted.
130    pub messages: u64,
131    /// Size of `conversation.jsonl` on disk.
132    pub bytes: u64,
133    /// Last write to `conversation.jsonl`, in milliseconds since the epoch. `None` when the
134    /// host's mtime is unreadable or predates the epoch.
135    pub last_touched_ms: Option<u64>,
136    /// The capsule the header line names, and `None` for a record written before the header
137    /// existed. An unowned record is never pruned automatically.
138    pub capsule: Option<String>,
139    /// What this record has already dropped, or `None` if it has never been truncated.
140    pub truncation: Option<TruncationMarker>,
141}
142
143/// A context directory `mur conversation rm` removed.
144#[derive(Debug, Clone, PartialEq, Eq)]
145pub struct RemovedRecord {
146    pub record: String,
147    pub context_id: String,
148    /// The directory that was removed, whole.
149    pub path: PathBuf,
150    /// Messages it held.
151    pub messages: u64,
152}
153
154/// One context directory of the record store: a directory under a record store holding
155/// `conversation.jsonl`, `harness-session.json`, or both.
156#[derive(Debug, Clone, PartialEq, Eq)]
157pub struct ContextDir {
158    /// Record store: the directory under `~/.murmur/conversations/`.
159    pub record: String,
160    /// Context id: the directory under the record store.
161    pub context_id: String,
162    /// The context directory itself.
163    pub path: PathBuf,
164}
165
166/// What one context directory holds, read the way the runtime reads it.
167#[derive(Debug, Clone, PartialEq)]
168pub struct ContextContents {
169    /// The header line, when `conversation.jsonl`'s first line is one.
170    pub header: Option<RecordHeader>,
171    /// Every message line of `conversation.jsonl`, oldest first, as recorded. Empty when there is
172    /// no `conversation.jsonl`, which is the case on `inference.transport: process`.
173    pub messages: Vec<serde_json::Value>,
174    /// The context's harness session, `None` wherever the runtime reads "no session".
175    pub harness_session: Option<HarnessSessionEntry>,
176}
177
178// ── Session pruning ──────────────────────────────────────────────────────────
179
180/// The millisecond timestamp inside a `ses_` id, or `None` when the name is not one.
181///
182/// `ses_` ids are uuid v7 in simple form, so their first 12 hex characters are the mint time in
183/// milliseconds. This is the whole of the session age policy: no `stat`, no walk, and a lexical
184/// sort of the names is a chronological sort of the sessions.
185pub fn session_id_timestamp_ms(name: &str) -> Option<u64> {
186    uuid_v7_timestamp_ms(name, "ses_")
187}
188
189/// The millisecond timestamp in the first 12 hex characters after `prefix`, the mint time of a
190/// uuid v7 in simple form. `None` when `id` does not start with `prefix` or they are not hex.
191pub(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
196/// The directory holding the work folders the runtime creates, which [`prune_sessions`] prunes
197/// beside the records group: `<manifest_dir>/workdir` for a launch without `--workdir`, and
198/// `None` under `--workdir`, whose folder is the operator's and is never touched.
199pub(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
206/// Remove the sessions of one records group that `policy` does not keep: each pruned session's
207/// records directory in `records_group` and, when `created_workdirs` is given
208/// ([`created_workdirs`]), the work folder of the same name the runtime created there, so a
209/// session goes whole.
210///
211/// The decision is made on `records_group`'s `ses_*` listing alone: a session whose work folder
212/// is already gone still goes, and a work folder with no records directory is never a candidate.
213/// Both removals go by descriptor through their roots, with no link followed.
214///
215/// `current_session_id` is a hard floor: no directory whose name sorts at or after it is ever a
216/// candidate, whatever the policy says. That is what keeps the running session's own trace, and
217/// what keeps a sibling capsule launched a moment ago from having its live session deleted —
218/// uuid v7 ordering makes "minted after mine" a property of the name, so no lock file is needed.
219///
220/// Both keys are ANDed: a session survives only if it is inside every limit declared. The count
221/// is over every entry present, the current session included, so `max_sessions: 3` leaves three
222/// directories and one of them is the running one. Reasons are attributed age-first, because
223/// "too old" is the more specific statement about a directory that fails both.
224///
225/// A records directory that cannot be removed is skipped and left out of the returned list, and
226/// its work folder is kept with it: retention never fails a launch, and a trace event must name
227/// only what actually went.
228pub 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    // Newest first, which for uuid v7 names is both the lexical and the chronological order.
249    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        // The floor. `>=` rather than `>` so the current session is excluded by the same rule
258        // that excludes everything minted after it.
259        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
287// ── Formation pruning ────────────────────────────────────────────────────────
288
289/// Remove the formation directories under `formations_dir` that belong to `project_dir` and that
290/// `policy` — the entry member's `trace.retain` — does not keep.
291///
292/// A directory is this project's when it is a real directory, not a symlink, its name is a
293/// formation id, and its [`FormationMarker`] parses and names `project_dir`, both sides
294/// canonicalized. Only those are ranked. A directory with no marker or an unreadable one is
295/// unowned, and one whose marker names another project is foreign: neither is ranked, and neither
296/// is ever removed. Every directory made before markers existed is unowned.
297///
298/// The policy reads as it does for sessions:
299///
300/// * `max_age_secs` is measured from the formation id's mint time, with no `stat`.
301/// * `max_sessions` counts this project's directories present, newest first, `current` and any
302///   minted after it included.
303/// * Both keys are ANDed, and reasons are attributed age-first.
304///
305/// `current` is a hard floor: no directory whose id sorts at or after it is ever removed. A
306/// directory `is_live` answers `true` for is skipped and keeps its rank, so a formation still
307/// running is never removed and still counts against `max_sessions`.
308///
309/// A removed formation's records subtree, the records group of its directory and every member's
310/// group beneath it, goes with it. A missing `formations_dir` removes nothing. A directory that
311/// cannot be removed is left out of the returned list, which is newest first.
312pub 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    // Newest first, which for uuid v7 ids is both the lexical and the chronological order.
341    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        // Every member's records group mirrors a directory beneath this one, so the formation's
367        // records subtree goes with it. Resolved before the removal, while the path still
368        // canonicalizes to what the members were launched in.
369        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
384// ── Record pruning ───────────────────────────────────────────────────────────
385
386/// Remove the context directories under `record_root` that `policy.max_age_secs` does not keep.
387///
388/// Three things are never touched:
389///
390/// * **The context this launch is using** (`current_context_id`) — retention must not delete the
391///   conversation it is about to continue.
392/// * **A record no header line owns.** Every record written before the header existed is
393///   unowned, and stays unowned until the capsule that writes it adopts it. `mur conversation rm`
394///   is what reaches an abandoned one.
395/// * **A record whose header names another capsule.** Two capsules pointed at one
396///   `context.record_store` is a deliberate operator act; one silently pruning the other's
397///   history is not.
398///
399/// `policy.max_messages` is not enforced here: it truncates the record this launch opens, at the
400/// point it is opened, which is O(one record) instead of O(every conversation) — see
401/// [`truncate_record`].
402pub 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
441/// Drop everything before the newest `keep` messages of `path`, atomically.
442///
443/// The file is rewritten, not edited in place: the kept tail plus a header line recording the
444/// drop is staged in the record's own directory, fsynced, and renamed over the original. Same
445/// directory means same filesystem, which is what makes the rename atomic — a crash at any point
446/// leaves the original whole and the temp file orphaned, never a half-written record.
447///
448/// Every surviving message keeps the exact `id` bytes its line carried; truncation drops
449/// messages and never renumbers them.
450///
451/// `capsule` is used only when the record has no header yet: an existing header's `capsule` is
452/// preserved, so truncating never transfers ownership of a record between capsules.
453///
454/// A record already at or under `keep` is left untouched and reported as `dropped: 0`.
455///
456/// Once the rewrite commits, every blob in the record's `blobs/` that no surviving line's image
457/// block names is removed, and counted in `blobs_removed`. A blob a kept line still names stays.
458pub fn truncate_record(path: &Path, keep: u32, capsule: &str) -> Result<TruncationOutcome, String> {
459    match stage_truncation(path, keep, capsule, now_ms())? {
460        None => Ok(TruncationOutcome {
461            dropped: 0,
462            kept: count_messages(path),
463            oldest_surviving_id: String::new(),
464            last_dropped_id: String::new(),
465            blobs_removed: 0,
466        }),
467        Some((staged, mut outcome)) => {
468            staged.commit()?;
469            outcome.blobs_removed = sweep_orphan_blobs(path);
470            Ok(outcome)
471        }
472    }
473}
474
475/// Remove every blob in `blobs/` beside the record at `path` that none of its lines' image blocks
476/// names, returning how many were removed. A record with no `blobs/` directory removes nothing; a
477/// name that is not a sha256, and a blob that cannot be removed, are left where they are.
478fn sweep_orphan_blobs(path: &Path) -> u64 {
479    let Some(blobs) = path
480        .parent()
481        .map(|dir| dir.join(crate::trace_blobs::BLOB_DIR_NAME))
482    else {
483        return 0;
484    };
485    let Ok(entries) = std::fs::read_dir(&blobs) else {
486        return 0;
487    };
488    let Ok(raw) = std::fs::read_to_string(path) else {
489        return 0;
490    };
491    let referenced: std::collections::HashSet<String> = raw
492        .lines()
493        .filter_map(parse_message_line)
494        .flat_map(|message| {
495            crate::images::referenced_blobs(&message)
496                .map(str::to_string)
497                .collect::<Vec<_>>()
498        })
499        .collect();
500    let mut removed = 0;
501    for entry in entries.flatten() {
502        let name = entry.file_name();
503        let Some(name) = name.to_str() else {
504            continue;
505        };
506        if !crate::images::is_sha256_hex(name) || referenced.contains(name) {
507            continue;
508        }
509        if std::fs::remove_file(entry.path()).is_ok() {
510            removed += 1;
511        }
512    }
513    removed
514}
515
516/// The half of [`truncate_record`] that touches nothing the reader can see: the rewritten record
517/// is on disk under a temporary name in the record's own directory, and the original is still the
518/// original until [`StagedRewrite::commit`] renames over it.
519///
520/// `None` means there was nothing to drop.
521pub(crate) fn stage_truncation(
522    path: &Path,
523    keep: u32,
524    capsule: &str,
525    at_ms: u64,
526) -> Result<Option<(StagedRewrite, TruncationOutcome)>, String> {
527    let raw = match std::fs::read_to_string(path) {
528        Ok(raw) => raw,
529        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
530        Err(err) => return Err(format!("{}: {err}", path.display())),
531    };
532
533    let existing = read_header(path);
534    let lines: Vec<&str> = raw.lines().collect();
535    // Every line that is a message, by its index in the file. Non-message lines — the header,
536    // and a torn final line — are not messages to any reader and are not counted here either.
537    let message_indices: Vec<usize> = lines
538        .iter()
539        .enumerate()
540        .filter(|(_, line)| parse_message_line(line).is_some())
541        .map(|(index, _)| index)
542        .collect();
543
544    let total = message_indices.len();
545    let keep = keep as usize;
546    if total <= keep {
547        return Ok(None);
548    }
549    let dropped = total - keep;
550    let first_kept_index = message_indices[dropped];
551
552    let id_of = |index: usize| -> String {
553        parse_message_line(lines[index])
554            .as_ref()
555            .and_then(|message| crate::agent::message_id(message).map(str::to_string))
556            .unwrap_or_default()
557    };
558    let last_dropped_id = id_of(message_indices[dropped - 1]);
559    let oldest_surviving_id = message_indices
560        .get(dropped)
561        .map(|i| id_of(*i))
562        .unwrap_or_default();
563
564    let header = RecordHeader {
565        kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
566        capsule: existing
567            .as_ref()
568            .map(|h| h.capsule.clone())
569            .unwrap_or_else(|| capsule.to_string()),
570        created_ms: existing.as_ref().map(|h| h.created_ms).unwrap_or(at_ms),
571        truncated: Some(TruncationMarker {
572            // Cumulative over the record's life: an operator reading `dropped` wants to know how
573            // much of this conversation is gone, not how much the last truncation took.
574            dropped: existing
575                .as_ref()
576                .and_then(|h| h.truncated.as_ref())
577                .map(|marker| marker.dropped)
578                .unwrap_or(0)
579                + dropped as u64,
580            oldest_surviving_id: oldest_surviving_id.clone(),
581            last_dropped_id: last_dropped_id.clone(),
582            at_ms,
583        }),
584    };
585
586    let mut contents = header_line(&header)?;
587    for line in &lines[first_kept_index..] {
588        contents.push_str(line);
589        contents.push('\n');
590    }
591
592    let staged = StagedRewrite::stage_with_mode(
593        path,
594        contents.as_bytes(),
595        Some(crate::conversation::RECORD_FILE_MODE),
596    )?;
597    Ok(Some((
598        staged,
599        TruncationOutcome {
600            dropped: dropped as u64,
601            kept: keep as u64,
602            oldest_surviving_id,
603            last_dropped_id,
604            blobs_removed: 0,
605        },
606    )))
607}
608
609/// A rewrite that exists on disk but has not replaced anything yet.
610///
611/// Staged as `<record dir>/.conversation.jsonl.<pid>.<nonce>.tmp` — in the record's own
612/// directory, so the rename that commits it is within one filesystem and therefore atomic, and
613/// under a name no reader mistakes for a record.
614pub(crate) struct StagedRewrite {
615    target: PathBuf,
616    temp: PathBuf,
617    committed: bool,
618}
619
620impl StagedRewrite {
621    /// Stage `contents` for `target`, with the staged file's mode set explicitly for a target whose
622    /// permissions are part of its contract. `None` leaves the mode to the process umask.
623    pub(crate) fn stage_with_mode(
624        target: &Path,
625        contents: &[u8],
626        mode: Option<u32>,
627    ) -> Result<Self, String> {
628        use std::io::Write;
629        use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
630
631        let dir = target
632            .parent()
633            .ok_or_else(|| format!("{} has no parent directory", target.display()))?;
634        let name = target
635            .file_name()
636            .map(|name| name.to_string_lossy().into_owned())
637            .unwrap_or_else(|| "record".to_string());
638        let temp = dir.join(format!(
639            ".{name}.{}.{}.tmp",
640            std::process::id(),
641            uuid::Uuid::now_v7().simple()
642        ));
643        let mut options = std::fs::OpenOptions::new();
644        options.write(true).create(true).truncate(true);
645        if let Some(mode) = mode {
646            options.mode(mode);
647        }
648        let mut file = options
649            .open(&temp)
650            .map_err(|err| format!("{}: {err}", temp.display()))?;
651        // The fsync is the point: a rename is only atomic with respect to a file whose bytes are
652        // already durable. Without it a crash can leave the new name pointing at a short file.
653        file.write_all(contents)
654            .and_then(|()| file.sync_all())
655            .map_err(|err| format!("{}: {err}", temp.display()))?;
656        // Re-asserted after the write: `mode` on the open only applies when this call created the
657        // file, and a mode that is part of the target's contract must hold for a staged file an
658        // earlier run left behind too.
659        if let Some(mode) = mode {
660            std::fs::set_permissions(&temp, std::fs::Permissions::from_mode(mode))
661                .map_err(|err| format!("{}: {err}", temp.display()))?;
662        }
663        Ok(Self {
664            target: target.to_path_buf(),
665            temp,
666            committed: false,
667        })
668    }
669
670    #[cfg(test)]
671    pub(crate) fn temp_path(&self) -> &Path {
672        &self.temp
673    }
674
675    pub(crate) fn commit(mut self) -> Result<(), String> {
676        std::fs::rename(&self.temp, &self.target)
677            .map_err(|err| format!("{}: {err}", self.target.display()))?;
678        self.committed = true;
679        Ok(())
680    }
681}
682
683impl Drop for StagedRewrite {
684    fn drop(&mut self) {
685        if !self.committed {
686            let _ = std::fs::remove_file(&self.temp);
687        }
688    }
689}
690
691// ── The record store, as the CLI reads it ────────────────────────────────────
692
693/// Every conversation record on this host, record store by record store.
694///
695/// Ordered by record then context id, so two runs of `mur conversation ls` print the same table.
696/// A host with no `~/.murmur/conversations/` has no records, which is an empty list rather than
697/// an error.
698pub fn list_records() -> Result<Vec<RecordSummary>, String> {
699    let root = conversations_root()?;
700    let mut summaries = Vec::new();
701    for record in child_dirs(&root) {
702        let store = root.join(&record);
703        for (context_id, dir) in context_dirs(&store) {
704            let path = dir.join(RECORD_FILE_NAME);
705            if !path.exists() {
706                continue;
707            }
708            let header = read_header(&path);
709            summaries.push(RecordSummary {
710                record: record.clone(),
711                context_id,
712                messages: count_messages(&path),
713                bytes: std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0),
714                last_touched_ms: last_touched_ms(&path),
715                capsule: header.as_ref().map(|h| h.capsule.clone()),
716                truncation: header.and_then(|h| h.truncated),
717                path,
718            });
719        }
720    }
721    summaries.sort_by(|a, b| {
722        a.record
723            .cmp(&b.record)
724            .then_with(|| a.context_id.cmp(&b.context_id))
725    });
726    Ok(summaries)
727}
728
729/// Remove one context directory whole, and report what it held.
730///
731/// The only way to reclaim a record no capsule owns — one written before the header line
732/// existed, whose capsule has since been retired. Automatic pruning never touches those.
733pub fn remove_record(record: &str, context_id: &str) -> Result<RemovedRecord, String> {
734    let dir = conversations_root()?.join(record).join(context_id);
735    if !dir.is_dir() {
736        return Err(format!("{} is not a context directory", dir.display()));
737    }
738    let messages = count_messages(&dir.join(RECORD_FILE_NAME));
739    std::fs::remove_dir_all(&dir).map_err(|err| format!("{}: {err}", dir.display()))?;
740    Ok(RemovedRecord {
741        record: record.to_string(),
742        context_id: context_id.to_string(),
743        path: dir,
744        messages,
745    })
746}
747
748/// Every record that has something to say about `message_id`.
749///
750/// A record answers [`MessageStatus::Present`] when the id is one of its lines, and
751/// [`MessageStatus::Truncated`] when it is not but the record has dropped a message at least as
752/// new as this one — the id's own uuid v7 timestamp against the marker's `last_dropped_id`.
753/// Records with nothing to say are left out, so an empty list is
754/// [`MessageStatus::Unknown`] everywhere.
755///
756/// That classification is the whole reason the marker exists: an artifact that stored
757/// `source_id: msg_X` and now finds nothing must be told the conversation was trimmed, not that
758/// its reference was never real.
759pub fn locate_message(message_id: &str) -> Result<Vec<MessageLocation>, String> {
760    let minted_ms = message_id_timestamp_ms(message_id);
761    let mut found = Vec::new();
762    for summary in list_records()? {
763        let raw = std::fs::read_to_string(&summary.path).unwrap_or_default();
764        let mut position = 0u64;
765        let mut at = None;
766        for line in raw.lines() {
767            let Some(message) = parse_message_line(line) else {
768                continue;
769            };
770            position += 1;
771            if crate::agent::message_id(&message) == Some(message_id) {
772                at = Some(position);
773            }
774        }
775        let status = match (at, summary.truncation.as_ref()) {
776            (Some(position), _) => MessageStatus::Present {
777                position,
778                total: position.max(summary.messages),
779            },
780            (None, Some(marker)) => {
781                let dropped_at = message_id_timestamp_ms(&marker.last_dropped_id);
782                match (minted_ms, dropped_at) {
783                    (Some(minted), Some(last)) if minted <= last => MessageStatus::Truncated {
784                        dropped: marker.dropped,
785                        oldest_surviving_id: marker.oldest_surviving_id.clone(),
786                    },
787                    _ => MessageStatus::Unknown,
788                }
789            }
790            (None, None) => MessageStatus::Unknown,
791        };
792        if status != MessageStatus::Unknown {
793            found.push(MessageLocation {
794                record: summary.record,
795                context_id: summary.context_id,
796                path: summary.path,
797                status,
798            });
799        }
800    }
801    Ok(found)
802}
803
804/// Every context directory named `context_id`, in every record store, ordered by record store.
805///
806/// A context counts when its directory holds `conversation.jsonl` or `harness-session.json`, so a
807/// context on `inference.transport: process`, which has only the second, is found as well;
808/// [`list_records`] counts the first alone. An empty list is no such context. The `Err` is `HOME`
809/// being unusable.
810pub fn find_context_dirs(context_id: &str) -> Result<Vec<ContextDir>, String> {
811    Ok(find_context_dirs_in(&conversations_root()?, context_id))
812}
813
814fn find_context_dirs_in(root: &Path, context_id: &str) -> Vec<ContextDir> {
815    let mut found = Vec::new();
816    for record in child_dirs(root) {
817        for (name, path) in context_dirs(&root.join(&record)) {
818            if name != context_id {
819                continue;
820            }
821            if path.join(RECORD_FILE_NAME).exists() || path.join(HARNESS_SESSION_FILE_NAME).exists()
822            {
823                found.push(ContextDir {
824                    record: record.clone(),
825                    context_id: name,
826                    path,
827                });
828            }
829        }
830    }
831    found
832}
833
834/// What the context directory `dir` holds: the header and messages of its `conversation.jsonl`
835/// and its harness session.
836///
837/// Reads with the runtime's own rules — [`parse_header`] on line 1, [`record_messages`] for the
838/// messages, [`read_entry`] for the harness session — and writes nothing. A missing
839/// `conversation.jsonl` is no header and no messages. The `Err` is `conversation.jsonl` existing
840/// and not being readable, and names the path.
841pub fn read_context_dir(dir: &Path) -> Result<ContextContents, String> {
842    let record = dir.join(RECORD_FILE_NAME);
843    let raw = match std::fs::read_to_string(&record) {
844        Ok(raw) => raw,
845        Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
846        Err(err) => return Err(format!("{}: {err}", record.display())),
847    };
848    Ok(ContextContents {
849        header: raw.lines().next().and_then(parse_header),
850        messages: record_messages(&raw, |_| {}),
851        harness_session: read_entry(&dir.join(HARNESS_SESSION_FILE_NAME)),
852    })
853}
854
855// ── Shared helpers ───────────────────────────────────────────────────────────
856
857/// `~/.murmur/conversations`, resolved and not created.
858pub fn conversations_root() -> Result<PathBuf, String> {
859    Ok(crate::state_store::murmur_home_dir()?.join(CONVERSATION_ROOT_DIR))
860}
861
862/// Milliseconds since the epoch, for the callers that have no fixture to take a clock from.
863pub fn now_ms() -> u64 {
864    std::time::SystemTime::now()
865        .duration_since(std::time::UNIX_EPOCH)
866        .map(|elapsed| elapsed.as_millis() as u64)
867        .unwrap_or(0)
868}
869
870/// Immediate subdirectory names of `dir`, sorted, skipping dotted entries.
871fn child_dirs(dir: &Path) -> Vec<String> {
872    let mut names: Vec<String> = match std::fs::read_dir(dir) {
873        Ok(entries) => entries
874            .filter_map(Result::ok)
875            .filter(|entry| entry.path().is_dir())
876            .map(|entry| entry.file_name().to_string_lossy().into_owned())
877            .filter(|name| !name.starts_with('.'))
878            .collect(),
879        Err(_) => Vec::new(),
880    };
881    names.sort();
882    names
883}
884
885/// Every context directory under one record store, as `(context id, path)`.
886fn context_dirs(record_root: &Path) -> Vec<(String, PathBuf)> {
887    child_dirs(record_root)
888        .into_iter()
889        .map(|name| {
890            let path = record_root.join(&name);
891            (name, path)
892        })
893        .collect()
894}
895
896/// Message lines in `path`. The header line is not a message and is never counted, which is what
897/// keeps it out of the `total` every reader of a record reports.
898fn count_messages(path: &Path) -> u64 {
899    std::fs::read_to_string(path)
900        .map(|raw| {
901            raw.lines()
902                .filter(|line| parse_message_line(line).is_some())
903                .count() as u64
904        })
905        .unwrap_or(0)
906}
907
908/// Last write to `path`, in milliseconds since the epoch.
909fn last_touched_ms(path: &Path) -> Option<u64> {
910    std::fs::metadata(path)
911        .and_then(|metadata| metadata.modified())
912        .ok()?
913        .duration_since(std::time::UNIX_EPOCH)
914        .ok()
915        .map(|elapsed| elapsed.as_millis() as u64)
916}
917
918#[cfg(test)]
919mod tests {
920    use super::*;
921    use murmur_artifact::{ContextRetainConfig, TraceRetainConfig};
922    use serde_json::json;
923
924    /// A `ses_` name whose uuid-v7 timestamp is exactly `ms`, and whose tail distinguishes it
925    /// from any sibling minted in the same millisecond.
926    fn ses(ms: u64, tail: u64) -> String {
927        format!("ses_{ms:012x}{tail:020x}")
928    }
929
930    /// A `msg_` id whose uuid-v7 timestamp is exactly `ms`.
931    fn msg(ms: u64, tail: u64) -> String {
932        format!("msg_{ms:012x}{tail:020x}")
933    }
934
935    fn message_line(id: &str, text: &str) -> String {
936        json!({"role": "user", "content": [{"type": "text", "text": text}], "id": id}).to_string()
937    }
938
939    fn make_sessions(root: &Path, names: &[String]) {
940        for name in names {
941            let dir = root.join(name);
942            std::fs::create_dir_all(dir.join("blobs")).unwrap();
943            std::fs::write(dir.join("trace.jsonl"), "{}\n").unwrap();
944            std::fs::write(dir.join("blobs").join("deadbeef"), b"body").unwrap();
945        }
946    }
947
948    fn names(root: &Path) -> Vec<String> {
949        let mut found: Vec<String> = std::fs::read_dir(root)
950            .unwrap()
951            .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
952            .collect();
953        found.sort();
954        found
955    }
956
957    const NOW: u64 = 1_756_400_000_000;
958
959    /// the count is over every entry present, the running session included,
960    /// so `max_sessions: 3` leaves three directories and one of them is the current one. What went
961    /// went whole — no orphaned `trace.jsonl`, no orphaned `blobs/`.
962    #[test]
963    fn max_sessions_leaves_exactly_the_newest_n_including_the_current() {
964        let root = tempfile::tempdir().unwrap();
965        let all: Vec<String> = (1..=6).map(|n| ses(NOW - (6 - n) * 1000, n)).collect();
966        make_sessions(root.path(), &all);
967
968        let pruned = prune_sessions(
969            &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
970            None,
971            &all[5],
972            &TraceRetainConfig {
973                max_sessions: Some(3),
974                max_age_secs: None,
975            },
976            NOW,
977        );
978
979        assert_eq!(
980            names(root.path()),
981            vec![all[3].clone(), all[4].clone(), all[5].clone()]
982        );
983        let mut removed: Vec<String> = pruned.iter().map(|p| p.name.clone()).collect();
984        removed.sort();
985        assert_eq!(
986            removed,
987            vec![all[0].clone(), all[1].clone(), all[2].clone()]
988        );
989        assert!(pruned
990            .iter()
991            .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_SESSIONS));
992        for name in &all[..3] {
993            assert!(!root.path().join(name).exists(), "{name} went whole");
994        }
995    }
996
997    /// the decision comes out of the id. Every directory here is created in
998    /// the same instant, so every mtime is equally fresh; the two whose ids encode two hours ago
999    /// are the two that go.
1000    #[test]
1001    fn max_age_reads_the_id_and_never_the_filesystem() {
1002        let root = tempfile::tempdir().unwrap();
1003        let old_a = ses(NOW - 7_200_000, 1);
1004        let old_b = ses(NOW - 7_100_000, 2);
1005        let recent = ses(NOW - 60_000, 3);
1006        let current = ses(NOW, 4);
1007        make_sessions(
1008            root.path(),
1009            &[
1010                old_a.clone(),
1011                old_b.clone(),
1012                recent.clone(),
1013                current.clone(),
1014            ],
1015        );
1016
1017        let pruned = prune_sessions(
1018            &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1019            None,
1020            &current,
1021            &TraceRetainConfig {
1022                max_sessions: None,
1023                max_age_secs: Some(900),
1024            },
1025            NOW,
1026        );
1027
1028        assert_eq!(pruned.len(), 2, "{pruned:?}");
1029        assert!(pruned
1030            .iter()
1031            .all(|p| p.reason == crate::trace::RETENTION_REASON_MAX_AGE));
1032        assert_eq!(names(root.path()), vec![recent, current]);
1033    }
1034
1035    /// the two keys are ANDed. An old-id directory never survives on rank, and a recent-id
1036    /// directory outside the count does not survive on age.
1037    #[test]
1038    fn both_trace_keys_keep_only_sessions_inside_both() {
1039        let root = tempfile::tempdir().unwrap();
1040        let old = [ses(NOW - 7_200_000, 1), ses(NOW - 7_100_000, 2)];
1041        let recent = [
1042            ses(NOW - 300_000, 3),
1043            ses(NOW - 200_000, 4),
1044            ses(NOW - 100_000, 5),
1045        ];
1046        let current = ses(NOW, 6);
1047        let mut all = old.to_vec();
1048        all.extend(recent.iter().cloned());
1049        all.push(current.clone());
1050        make_sessions(root.path(), &all);
1051
1052        let pruned = prune_sessions(
1053            &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1054            None,
1055            &current,
1056            &TraceRetainConfig {
1057                max_sessions: Some(2),
1058                max_age_secs: Some(900),
1059            },
1060            NOW,
1061        );
1062
1063        assert_eq!(names(root.path()), vec![recent[2].clone(), current.clone()]);
1064        assert_eq!(pruned.len(), 4);
1065        for session in &pruned {
1066            let expected = if old.contains(&session.name) {
1067                crate::trace::RETENTION_REASON_MAX_AGE
1068            } else {
1069                crate::trace::RETENTION_REASON_MAX_SESSIONS
1070            };
1071            assert_eq!(session.reason, expected, "{session:?}");
1072        }
1073    }
1074
1075    /// the floor. Nothing at or after the current session's own id is ever a candidate, which
1076    /// is also what keeps a sibling capsule's concurrently running session safe with no lock file.
1077    #[test]
1078    fn nothing_at_or_after_the_current_session_id_is_a_candidate() {
1079        let root = tempfile::tempdir().unwrap();
1080        let current = ses(NOW, 5);
1081        // A sibling launched a millisecond later, while this session was still staging.
1082        let sibling = ses(NOW + 1, 6);
1083        let older = ses(NOW - 1000, 4);
1084        make_sessions(
1085            root.path(),
1086            &[current.clone(), sibling.clone(), older.clone()],
1087        );
1088
1089        let pruned = prune_sessions(
1090            &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1091            None,
1092            &current,
1093            &TraceRetainConfig {
1094                max_sessions: Some(1),
1095                max_age_secs: Some(1),
1096            },
1097            NOW,
1098        );
1099
1100        assert_eq!(pruned.len(), 1);
1101        assert_eq!(pruned[0].name, older);
1102        assert_eq!(names(root.path()), vec![current.clone(), sibling]);
1103        assert!(root.path().join(&current).join("trace.jsonl").exists());
1104    }
1105
1106    /// A pruned session goes whole: its records directory and its created work folder of the same
1107    /// name both go, and a work folder with no records directory is never a candidate, whatever
1108    /// its name says.
1109    #[test]
1110    fn a_pruned_session_takes_its_records_and_its_created_work_folder() {
1111        let records = tempfile::tempdir().unwrap();
1112        let workdirs = tempfile::tempdir().unwrap();
1113        let all: Vec<String> = (1..=3).map(|n| ses(NOW - (3 - n) * 1000, n)).collect();
1114        make_sessions(records.path(), &all);
1115        for name in &all {
1116            std::fs::create_dir_all(workdirs.path().join(name).join("out")).unwrap();
1117        }
1118        let workdir_only = ses(NOW - 9_000, 9);
1119        std::fs::create_dir_all(workdirs.path().join(&workdir_only)).unwrap();
1120
1121        let pruned = prune_sessions(
1122            &crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
1123            Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
1124            &all[2],
1125            &TraceRetainConfig {
1126                max_sessions: Some(2),
1127                max_age_secs: None,
1128            },
1129            NOW,
1130        );
1131
1132        assert_eq!(pruned.len(), 1);
1133        assert_eq!(pruned[0].name, all[0]);
1134        assert_eq!(names(records.path()), vec![all[1].clone(), all[2].clone()]);
1135        let mut kept = vec![all[1].clone(), all[2].clone(), workdir_only];
1136        kept.sort();
1137        assert_eq!(names(workdirs.path()), kept);
1138    }
1139
1140    /// Under `--workdir` retention is handed no work folders: an old session's records go, and a
1141    /// planted `.murmur/<old-id>/` and `<old-id>/` in the operator's folder both survive. Without
1142    /// it, the old session's runtime-created `workdir/<old-id>` goes with its records.
1143    #[test]
1144    fn retention_prunes_a_created_work_folder_and_never_the_operators() {
1145        let records = tempfile::tempdir().unwrap();
1146        let operator = tempfile::tempdir().unwrap();
1147        let manifest_dir = tempfile::tempdir().unwrap();
1148        let old = ses(NOW - 2000, 1);
1149        let current = ses(NOW, 2);
1150        let policy = TraceRetainConfig {
1151            max_sessions: Some(1),
1152            max_age_secs: None,
1153        };
1154        let open = |path: &std::path::Path| crate::workdir_fs::WorkdirRoot::open(path).unwrap();
1155
1156        make_sessions(records.path(), &[old.clone(), current.clone()]);
1157        let planted = [
1158            operator.path().join(".murmur").join(&old),
1159            operator.path().join(&old),
1160        ];
1161        for dir in &planted {
1162            std::fs::create_dir_all(dir).unwrap();
1163            std::fs::write(dir.join("keep"), b"x").unwrap();
1164        }
1165        assert_eq!(
1166            created_workdirs(Some(operator.path()), manifest_dir.path()),
1167            None
1168        );
1169        let pruned = prune_sessions(
1170            &open(records.path()),
1171            created_workdirs(Some(operator.path()), manifest_dir.path())
1172                .map(|dir| open(&dir))
1173                .as_ref(),
1174            &current,
1175            &policy,
1176            NOW,
1177        );
1178        assert_eq!(pruned.len(), 1);
1179        assert_eq!(names(records.path()), vec![current.clone()]);
1180        for dir in &planted {
1181            assert!(dir.join("keep").is_file(), "{} survives", dir.display());
1182        }
1183
1184        make_sessions(records.path(), std::slice::from_ref(&old));
1185        let workdirs = created_workdirs(None, manifest_dir.path()).unwrap();
1186        assert_eq!(workdirs, manifest_dir.path().join("workdir"));
1187        std::fs::create_dir_all(workdirs.join(&old)).unwrap();
1188        std::fs::create_dir_all(workdirs.join(&current)).unwrap();
1189        let pruned = prune_sessions(
1190            &open(records.path()),
1191            Some(&open(&workdirs)),
1192            &current,
1193            &policy,
1194            NOW,
1195        );
1196        assert_eq!(pruned.len(), 1);
1197        assert!(!workdirs.join(&old).exists());
1198        assert!(workdirs.join(&current).exists());
1199    }
1200
1201    /// A pruned session's temporary folder goes with its records directory, even when a command
1202    /// left a read-only directory holding a file and a directory no one can open in it.
1203    #[test]
1204    fn pruning_a_session_removes_its_temp_folder() {
1205        use std::os::unix::fs::PermissionsExt;
1206
1207        let root = tempfile::tempdir().unwrap();
1208        let old = ses(NOW - 2000, 1);
1209        let current = ses(NOW, 2);
1210        make_sessions(root.path(), &[old.clone(), current.clone()]);
1211        let folder = root
1212            .path()
1213            .join(&old)
1214            .join(crate::temp_dir::SESSION_TEMP_DIR_NAME);
1215        let read_only = folder.join("read-only");
1216        let sealed_shut = folder.join("sealed-shut");
1217        std::fs::create_dir_all(&read_only).unwrap();
1218        std::fs::create_dir_all(&sealed_shut).unwrap();
1219        std::fs::write(read_only.join("file"), b"left behind").unwrap();
1220        std::fs::set_permissions(&read_only, std::fs::Permissions::from_mode(0o555)).unwrap();
1221        std::fs::set_permissions(&sealed_shut, std::fs::Permissions::from_mode(0o000)).unwrap();
1222
1223        let pruned = prune_sessions(
1224            &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1225            None,
1226            &current,
1227            &TraceRetainConfig {
1228                max_sessions: Some(1),
1229                max_age_secs: None,
1230            },
1231            NOW,
1232        );
1233
1234        assert_eq!(pruned.len(), 1, "{pruned:?}");
1235        assert_eq!(pruned[0].name, old);
1236        assert!(!root.path().join(&old).exists(), "the session went whole");
1237        assert_eq!(names(root.path()), vec![current]);
1238    }
1239
1240    /// A link planted where a created work folder belongs is refused, not followed: the
1241    /// directory it points at survives the pruning of the session it is named for.
1242    #[test]
1243    fn a_link_in_place_of_a_created_work_folder_is_not_followed() {
1244        let records = tempfile::tempdir().unwrap();
1245        let workdirs = tempfile::tempdir().unwrap();
1246        let elsewhere = tempfile::tempdir().unwrap();
1247        std::fs::write(elsewhere.path().join("keep"), b"x").unwrap();
1248        let all: Vec<String> = (1..=2).map(|n| ses(NOW - (2 - n) * 1000, n)).collect();
1249        make_sessions(records.path(), &all);
1250        std::os::unix::fs::symlink(elsewhere.path(), workdirs.path().join(&all[0])).unwrap();
1251
1252        let pruned = prune_sessions(
1253            &crate::workdir_fs::WorkdirRoot::open(records.path()).unwrap(),
1254            Some(&crate::workdir_fs::WorkdirRoot::open(workdirs.path()).unwrap()),
1255            &all[1],
1256            &TraceRetainConfig {
1257                max_sessions: Some(1),
1258                max_age_secs: None,
1259            },
1260            NOW,
1261        );
1262
1263        assert_eq!(pruned.len(), 1);
1264        assert!(elsewhere.path().join("keep").exists());
1265    }
1266
1267    /// A policy with neither key set — which the manifest parser refuses, but which the runtime
1268    /// must still treat as inert rather than as "prune everything".
1269    #[test]
1270    fn an_empty_policy_removes_nothing() {
1271        let root = tempfile::tempdir().unwrap();
1272        let all = [ses(NOW - 9_000_000, 1), ses(NOW, 2)];
1273        make_sessions(root.path(), &all);
1274        assert!(prune_sessions(
1275            &crate::workdir_fs::WorkdirRoot::open(root.path()).unwrap(),
1276            None,
1277            &all[1],
1278            &TraceRetainConfig {
1279                max_sessions: None,
1280                max_age_secs: None
1281            },
1282            NOW
1283        )
1284        .is_empty());
1285        assert_eq!(names(root.path()), all.to_vec());
1286    }
1287
1288    // ── Truncation ───────────────────────────────────────────────────────────
1289
1290    /// A record with `count` messages, owned by `capsule` when `owned`.
1291    fn write_record(dir: &Path, capsule: Option<&str>, count: u64) -> PathBuf {
1292        std::fs::create_dir_all(dir).unwrap();
1293        let path = dir.join(RECORD_FILE_NAME);
1294        let mut contents = String::new();
1295        if let Some(capsule) = capsule {
1296            contents.push_str(
1297                &header_line(&RecordHeader {
1298                    kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
1299                    capsule: capsule.to_string(),
1300                    created_ms: NOW,
1301                    truncated: None,
1302                })
1303                .unwrap(),
1304            );
1305        }
1306        for i in 1..=count {
1307            contents.push_str(&message_line(
1308                &msg(NOW - (count - i) * 1000, i),
1309                &format!("m{i}"),
1310            ));
1311            contents.push('\n');
1312        }
1313        std::fs::write(&path, contents).unwrap();
1314        path
1315    }
1316
1317    /// A record line whose tool message carries one image block per sha in `shas`.
1318    fn image_line(id: &str, shas: &[&str]) -> String {
1319        let mut content = vec![serde_json::json!({"type": "text", "text": "shot"})];
1320        for sha in shas {
1321            content.push(serde_json::json!({
1322                "type": "image",
1323                "source": {"type": "file", "sha256": sha, "media_type": "image/png", "bytes": 4},
1324            }));
1325        }
1326        serde_json::json!({"id": id, "role": "tool", "content": content}).to_string()
1327    }
1328
1329    /// A blob only dropped lines name is removed after the rewrite; one a kept line still names
1330    /// stays, and so does a file whose name is no sha256.
1331    #[test]
1332    fn truncation_removes_blobs_no_surviving_line_names() {
1333        let dir = tempfile::tempdir().unwrap();
1334        let path = dir.path().join(RECORD_FILE_NAME);
1335        let (old, shared, new) = ("a".repeat(64), "b".repeat(64), "c".repeat(64));
1336        let lines = [
1337            image_line(&msg(NOW - 3000, 1), &[&old, &shared]),
1338            image_line(&msg(NOW - 2000, 2), &[&old]),
1339            image_line(&msg(NOW - 1000, 3), &[&shared]),
1340            image_line(&msg(NOW, 4), &[&new]),
1341        ];
1342        std::fs::write(&path, lines.join("\n") + "\n").unwrap();
1343        let blobs = dir.path().join("blobs");
1344        std::fs::create_dir(&blobs).unwrap();
1345        for name in [&old, &shared, &new, &"notes.txt".to_string()] {
1346            std::fs::write(blobs.join(name), b"\x89PNG").unwrap();
1347        }
1348
1349        let outcome = truncate_record(&path, 2, "capsule").unwrap();
1350        assert_eq!(outcome.dropped, 2);
1351        assert_eq!(outcome.blobs_removed, 1);
1352        assert!(!blobs.join(&old).exists());
1353        assert!(blobs.join(&shared).exists(), "a kept line still names it");
1354        assert!(blobs.join(&new).exists());
1355        assert!(blobs.join("notes.txt").exists());
1356    }
1357
1358    /// A record with no `blobs/` truncates exactly as before and removes nothing.
1359    #[test]
1360    fn truncation_without_blobs_removes_nothing() {
1361        let dir = tempfile::tempdir().unwrap();
1362        let path = write_record(dir.path(), Some("capsule"), 5);
1363        let outcome = truncate_record(&path, 2, "capsule").unwrap();
1364        assert_eq!(outcome.dropped, 3);
1365        assert_eq!(outcome.blobs_removed, 0);
1366        assert!(!dir.path().join("blobs").exists());
1367    }
1368
1369    /// the newest N survive with their ids intact, the file is a header plus exactly N
1370    /// message lines, and every line parses.
1371    #[test]
1372    fn max_messages_leaves_the_newest_n_with_their_ids_intact() {
1373        let dir = tempfile::tempdir().unwrap();
1374        let path = write_record(dir.path(), Some("capsule"), 10);
1375        let before: Vec<String> = std::fs::read_to_string(&path)
1376            .unwrap()
1377            .lines()
1378            .filter_map(parse_message_line)
1379            .map(|m| m["id"].as_str().unwrap().to_string())
1380            .collect();
1381
1382        let outcome = truncate_record(&path, 3, "capsule").unwrap();
1383        assert_eq!(outcome.dropped, 7);
1384        assert_eq!(outcome.kept, 3);
1385        assert_eq!(outcome.last_dropped_id, before[6]);
1386        assert_eq!(outcome.oldest_surviving_id, before[7]);
1387
1388        let raw = std::fs::read_to_string(&path).unwrap();
1389        let lines: Vec<&str> = raw.lines().collect();
1390        assert_eq!(
1391            lines.len(),
1392            4,
1393            "a header plus exactly three messages: {lines:?}"
1394        );
1395        for line in &lines {
1396            serde_json::from_str::<serde_json::Value>(line).expect("every line is JSON");
1397        }
1398        let header = parse_header(lines[0]).expect("the first line is the header");
1399        assert_eq!(header.capsule, "capsule");
1400        let marker = header.truncated.expect("the header records the drop");
1401        assert_eq!(marker.dropped, 7);
1402        assert_eq!(marker.last_dropped_id, before[6]);
1403        assert_eq!(marker.oldest_surviving_id, before[7]);
1404
1405        let after: Vec<String> = lines[1..]
1406            .iter()
1407            .map(|line| {
1408                parse_message_line(line).expect("a message with a string role")["id"]
1409                    .as_str()
1410                    .unwrap()
1411                    .to_string()
1412            })
1413            .collect();
1414        assert_eq!(after, before[7..], "surviving ids are the bytes they were");
1415    }
1416
1417    /// The header does not count as a message: `count_messages` — the same rule
1418    /// `crate::conversation::read_record` applies to `total` — is unchanged by it.
1419    #[test]
1420    fn a_header_line_is_not_counted_as_a_message() {
1421        let dir = tempfile::tempdir().unwrap();
1422        let headerless = write_record(&dir.path().join("a"), None, 4);
1423        let owned = write_record(&dir.path().join("b"), Some("capsule"), 4);
1424        assert_eq!(count_messages(&headerless), 4);
1425        assert_eq!(count_messages(&owned), 4);
1426    }
1427
1428    /// Truncating twice accumulates: `dropped` is what this record has lost over its life, not
1429    /// what the last rewrite took.
1430    #[test]
1431    fn a_second_truncation_accumulates_the_dropped_count() {
1432        let dir = tempfile::tempdir().unwrap();
1433        let path = write_record(dir.path(), Some("capsule"), 10);
1434        truncate_record(&path, 6, "capsule").unwrap();
1435        truncate_record(&path, 2, "capsule").unwrap();
1436        let marker = read_header(&path).unwrap().truncated.unwrap();
1437        assert_eq!(marker.dropped, 8);
1438        assert_eq!(count_messages(&path), 2);
1439    }
1440
1441    #[test]
1442    fn truncation_rewrite_keeps_the_record_owner_only() {
1443        use std::os::unix::fs::PermissionsExt;
1444        let dir = tempfile::tempdir().unwrap();
1445        let path = write_record(dir.path(), Some("capsule"), 5);
1446        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
1447
1448        truncate_record(&path, 2, "capsule").unwrap();
1449
1450        assert_eq!(count_messages(&path), 2);
1451        assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
1452    }
1453
1454    /// A record already at or under the limit is left untouched, byte for byte.
1455    #[test]
1456    fn a_record_inside_the_limit_is_not_rewritten() {
1457        let dir = tempfile::tempdir().unwrap();
1458        let path = write_record(dir.path(), Some("capsule"), 3);
1459        let before = std::fs::read(&path).unwrap();
1460        let outcome = truncate_record(&path, 3, "capsule").unwrap();
1461        assert_eq!(outcome.dropped, 0);
1462        assert_eq!(std::fs::read(&path).unwrap(), before);
1463    }
1464
1465    /// a crash mid-truncate leaves the original intact. The staged rewrite is a temp file in
1466    /// the record's own directory — same filesystem, so the rename that commits it is atomic —
1467    /// and until that rename the original parses whole and holds its original message count.
1468    #[test]
1469    fn a_staged_truncation_leaves_the_original_intact_until_the_rename() {
1470        let dir = tempfile::tempdir().unwrap();
1471        let path = write_record(dir.path(), Some("capsule"), 10);
1472        let before = std::fs::read(&path).unwrap();
1473
1474        let (staged, outcome) = stage_truncation(&path, 3, "capsule", NOW).unwrap().unwrap();
1475        assert_eq!(outcome.dropped, 7);
1476
1477        let temp = staged.temp_path().to_path_buf();
1478        assert_eq!(
1479            temp.parent(),
1480            path.parent(),
1481            "the temp file is in the record's own directory, so the rename is atomic"
1482        );
1483        assert_ne!(temp.file_name().unwrap(), RECORD_FILE_NAME);
1484
1485        assert_eq!(
1486            std::fs::read(&path).unwrap(),
1487            before,
1488            "the original is byte-identical before the rename"
1489        );
1490        assert_eq!(count_messages(&path), 10);
1491        for line in std::fs::read_to_string(&path).unwrap().lines() {
1492            serde_json::from_str::<serde_json::Value>(line).expect("the original parses whole");
1493        }
1494
1495        // Dropping without committing is the crash: the original stands and no debris is left.
1496        drop(staged);
1497        assert!(!temp.exists());
1498        assert_eq!(std::fs::read(&path).unwrap(), before);
1499        assert_eq!(count_messages(&path), 10);
1500    }
1501
1502    // ── Record pruning ───────────────────────────────────────────────────────
1503
1504    /// automatic record pruning touches only a record whose header names this
1505    /// capsule. A record another capsule owns and a record no header owns both survive, however
1506    /// old they are — and the current launch's own context is never removed.
1507    #[test]
1508    fn record_pruning_skips_another_capsules_records_and_every_unowned_one() {
1509        let root = tempfile::tempdir().unwrap();
1510        write_record(&root.path().join("mine"), Some("mine"), 3);
1511        write_record(&root.path().join("theirs"), Some("theirs"), 3);
1512        write_record(&root.path().join("unowned"), None, 3);
1513        write_record(&root.path().join("current"), Some("mine"), 3);
1514
1515        // Far enough in the future that every mtime is outside a one-second window.
1516        let far_future = now_ms() + 3_600_000;
1517        let pruned = prune_records(
1518            root.path(),
1519            "mine",
1520            Some("current"),
1521            &ContextRetainConfig {
1522                max_messages: None,
1523                max_age_secs: Some(1),
1524            },
1525            far_future,
1526        );
1527
1528        assert_eq!(pruned.len(), 1, "{pruned:?}");
1529        assert_eq!(pruned[0].context_id, "mine");
1530        assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
1531        assert_eq!(names(root.path()), vec!["current", "theirs", "unowned"]);
1532    }
1533
1534    /// a record inside the window stays. Age is last write, so a record written a moment ago
1535    /// survives a 90-day policy however long ago its conversation started.
1536    #[test]
1537    fn a_record_written_inside_the_window_is_kept() {
1538        let root = tempfile::tempdir().unwrap();
1539        write_record(&root.path().join("fresh"), Some("mine"), 3);
1540        let pruned = prune_records(
1541            root.path(),
1542            "mine",
1543            None,
1544            &ContextRetainConfig {
1545                max_messages: None,
1546                max_age_secs: Some(90 * 86_400),
1547            },
1548            now_ms(),
1549        );
1550        assert!(pruned.is_empty());
1551        assert_eq!(names(root.path()), vec!["fresh"]);
1552    }
1553
1554    /// `max_messages` alone never removes a context directory: it is enforced on the record the
1555    /// launch opens, not by the age sweep.
1556    #[test]
1557    fn max_messages_alone_removes_no_record() {
1558        let root = tempfile::tempdir().unwrap();
1559        write_record(&root.path().join("ctx"), Some("mine"), 100);
1560        assert!(prune_records(
1561            root.path(),
1562            "mine",
1563            None,
1564            &ContextRetainConfig {
1565                max_messages: Some(1),
1566                max_age_secs: None
1567            },
1568            now_ms() + 3_600_000,
1569        )
1570        .is_empty());
1571        assert_eq!(names(root.path()), vec!["ctx"]);
1572    }
1573
1574    // ── Formation pruning ────────────────────────────────────────────────────
1575
1576    /// A formation id whose uuid-v7 timestamp is exactly `ms`.
1577    fn frm(ms: u64, tail: u64) -> FormationId {
1578        FormationId::parse(&format!("frm_{ms:012x}{tail:020x}")).unwrap()
1579    }
1580
1581    fn retain(max_sessions: Option<u32>, max_age_secs: Option<u64>) -> TraceRetainConfig {
1582        TraceRetainConfig {
1583            max_sessions,
1584            max_age_secs,
1585        }
1586    }
1587
1588    /// `<root>/<id>/worker/.murmur/ses_…/trace.jsonl`, with `marker` as `formation.json` when
1589    /// given.
1590    fn make_formation(root: &Path, id: &FormationId, marker: Option<&str>) -> PathBuf {
1591        let dir = root.join(id.as_str());
1592        let session = dir.join("worker/.murmur").join(ses(NOW, 1));
1593        std::fs::create_dir_all(&session).unwrap();
1594        std::fs::write(session.join("trace.jsonl"), "{}\n").unwrap();
1595        if let Some(marker) = marker {
1596            std::fs::write(
1597                dir.join(crate::formation_launch::FORMATION_MARKER_FILE),
1598                marker,
1599            )
1600            .unwrap();
1601        }
1602        dir
1603    }
1604
1605    fn marker_naming(project: &Path) -> String {
1606        serde_json::to_string(&FormationMarker {
1607            project_dir: project.to_path_buf(),
1608            launcher_pid: 1,
1609            launcher_start: String::new(),
1610        })
1611        .unwrap()
1612    }
1613
1614    /// Owned formations of `project` under `root`, one per id.
1615    fn make_owned(root: &Path, project: &Path, ids: &[&FormationId]) {
1616        for id in ids {
1617            make_formation(root, id, Some(&marker_naming(project)));
1618        }
1619    }
1620
1621    fn removed(pruned: &[PrunedFormation]) -> Vec<(FormationId, &'static str)> {
1622        pruned
1623            .iter()
1624            .map(|p| (p.formation_id.clone(), p.reason))
1625            .collect()
1626    }
1627
1628    const DAY_MS: u64 = 86_400_000;
1629
1630    /// The current formation and every one minted after it stay, under both keys at their
1631    /// tightest and every directory past both.
1632    #[test]
1633    fn prune_formations_never_removes_the_current_or_a_later_formation() {
1634        let root = tempfile::tempdir().unwrap();
1635        let project = tempfile::tempdir().unwrap();
1636        let old = frm(NOW - 3 * DAY_MS, 1);
1637        let current = frm(NOW - 2 * DAY_MS, 2);
1638        let later = frm(NOW - 2 * DAY_MS + 1000, 3);
1639        make_owned(root.path(), project.path(), &[&old, &current, &later]);
1640
1641        let pruned = prune_formations(
1642            root.path(),
1643            &current,
1644            project.path(),
1645            &retain(Some(1), Some(1)),
1646            NOW,
1647            |_, _| false,
1648        );
1649
1650        assert_eq!(
1651            removed(&pruned),
1652            [(old.clone(), crate::trace::RETENTION_REASON_MAX_AGE)]
1653        );
1654        assert_eq!(pruned[0].path, root.path().join(old.as_str()));
1655        assert_eq!(
1656            names(root.path()),
1657            vec![current.as_str().to_string(), later.as_str().to_string()]
1658        );
1659    }
1660
1661    /// `max_sessions: 3` leaves the three newest of this project's directories, the current one
1662    /// and one minted after it among them, and what went is listed newest first.
1663    #[test]
1664    fn prune_formations_keeps_the_newest_n_with_the_current_counted() {
1665        let root = tempfile::tempdir().unwrap();
1666        let project = tempfile::tempdir().unwrap();
1667        let all: Vec<FormationId> = (1..=5).map(|n| frm(NOW - (5 - n) * 1000, n)).collect();
1668        make_owned(root.path(), project.path(), &all.iter().collect::<Vec<_>>());
1669
1670        let pruned = prune_formations(
1671            root.path(),
1672            &all[3],
1673            project.path(),
1674            &retain(Some(3), None),
1675            NOW,
1676            |_, _| false,
1677        );
1678
1679        let reason = crate::trace::RETENTION_REASON_MAX_SESSIONS;
1680        assert_eq!(
1681            removed(&pruned),
1682            [(all[1].clone(), reason), (all[0].clone(), reason)]
1683        );
1684        assert_eq!(
1685            names(root.path()),
1686            all[2..]
1687                .iter()
1688                .map(|id| id.as_str().to_string())
1689                .collect::<Vec<_>>()
1690        );
1691    }
1692
1693    /// Age comes out of the id: every directory here is made in the same instant, and the two
1694    /// whose ids encode two hours ago are the two that go.
1695    #[test]
1696    fn prune_formations_reads_age_from_the_id() {
1697        let root = tempfile::tempdir().unwrap();
1698        let project = tempfile::tempdir().unwrap();
1699        let old_a = frm(NOW - 7_200_000, 1);
1700        let old_b = frm(NOW - 7_100_000, 2);
1701        let recent = frm(NOW - 60_000, 3);
1702        let current = frm(NOW, 4);
1703        make_owned(
1704            root.path(),
1705            project.path(),
1706            &[&old_a, &old_b, &recent, &current],
1707        );
1708
1709        let pruned = prune_formations(
1710            root.path(),
1711            &current,
1712            project.path(),
1713            &retain(None, Some(900)),
1714            NOW,
1715            |_, _| false,
1716        );
1717
1718        let reason = crate::trace::RETENTION_REASON_MAX_AGE;
1719        assert_eq!(removed(&pruned), [(old_b, reason), (old_a, reason)]);
1720        assert_eq!(
1721            names(root.path()),
1722            vec![recent.as_str().to_string(), current.as_str().to_string()]
1723        );
1724    }
1725
1726    /// A directory past both limits is reported as too old; one past the count alone, as over the
1727    /// count.
1728    #[test]
1729    fn prune_formations_attributes_reasons_age_first() {
1730        let root = tempfile::tempdir().unwrap();
1731        let project = tempfile::tempdir().unwrap();
1732        let old = frm(NOW - 7_200_000, 1);
1733        let recent = frm(NOW - 60_000, 2);
1734        let current = frm(NOW, 3);
1735        make_owned(root.path(), project.path(), &[&old, &recent, &current]);
1736
1737        let pruned = prune_formations(
1738            root.path(),
1739            &current,
1740            project.path(),
1741            &retain(Some(1), Some(900)),
1742            NOW,
1743            |_, _| false,
1744        );
1745
1746        assert_eq!(
1747            removed(&pruned),
1748            [
1749                (recent, crate::trace::RETENTION_REASON_MAX_SESSIONS),
1750                (old, crate::trace::RETENTION_REASON_MAX_AGE),
1751            ]
1752        );
1753    }
1754
1755    /// A live formation is never removed, and still counts: the one ranked behind it goes under
1756    /// `max_sessions: 2` although only two directories are not live.
1757    #[test]
1758    fn prune_formations_skips_a_live_directory_and_keeps_its_rank() {
1759        let root = tempfile::tempdir().unwrap();
1760        let project = tempfile::tempdir().unwrap();
1761        let older = frm(NOW - 3000, 1);
1762        let live = frm(NOW - 2000, 2);
1763        let current = frm(NOW, 3);
1764        make_owned(root.path(), project.path(), &[&older, &live, &current]);
1765
1766        let asked = std::cell::RefCell::new(Vec::new());
1767        let pruned = prune_formations(
1768            root.path(),
1769            &current,
1770            project.path(),
1771            &retain(Some(1), None),
1772            NOW,
1773            |id, marker| {
1774                assert_eq!(marker.launcher_pid, 1);
1775                asked.borrow_mut().push(id.clone());
1776                *id == live
1777            },
1778        );
1779
1780        assert_eq!(
1781            removed(&pruned),
1782            [(older.clone(), crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1783        );
1784        assert_eq!(*asked.borrow(), [live.clone(), older]);
1785        assert_eq!(
1786            names(root.path()),
1787            vec![live.as_str().to_string(), current.as_str().to_string()]
1788        );
1789
1790        // Ranked behind the live one, a directory inside the count is kept.
1791        let kept = frm(NOW - 2500, 4);
1792        make_owned(root.path(), project.path(), &[&kept]);
1793        let pruned = prune_formations(
1794            root.path(),
1795            &current,
1796            project.path(),
1797            &retain(Some(3), None),
1798            NOW,
1799            |id, _| *id == live,
1800        );
1801        assert!(pruned.is_empty(), "{pruned:?}");
1802        let pruned = prune_formations(
1803            root.path(),
1804            &current,
1805            project.path(),
1806            &retain(Some(2), None),
1807            NOW,
1808            |id, _| *id == live,
1809        );
1810        assert_eq!(
1811            removed(&pruned),
1812            [(kept, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1813        );
1814    }
1815
1816    /// No marker, a marker that does not parse, and a marker naming another project: none is
1817    /// removed, and none takes a rank from this project's own directories.
1818    #[test]
1819    fn prune_formations_never_ranks_or_removes_unowned_or_foreign_directories() {
1820        let root = tempfile::tempdir().unwrap();
1821        let project = tempfile::tempdir().unwrap();
1822        let other = tempfile::tempdir().unwrap();
1823        let owned_old = frm(NOW - 3 * DAY_MS, 1);
1824        let owned_kept = frm(NOW - 2 * DAY_MS, 2);
1825        let unmarked = frm(NOW - 2 * DAY_MS + 1, 3);
1826        let garbled = frm(NOW - 2 * DAY_MS + 2, 4);
1827        let foreign = frm(NOW - 2 * DAY_MS + 3, 5);
1828        let current = frm(NOW, 6);
1829        make_owned(
1830            root.path(),
1831            project.path(),
1832            &[&owned_old, &owned_kept, &current],
1833        );
1834        make_formation(root.path(), &unmarked, None);
1835        make_formation(root.path(), &garbled, Some("not json {"));
1836        make_formation(root.path(), &foreign, Some(&marker_naming(other.path())));
1837
1838        let pruned = prune_formations(
1839            root.path(),
1840            &current,
1841            project.path(),
1842            &retain(Some(2), None),
1843            NOW,
1844            |_, _| false,
1845        );
1846        assert_eq!(
1847            removed(&pruned),
1848            [(owned_old, crate::trace::RETENTION_REASON_MAX_SESSIONS)]
1849        );
1850
1851        let pruned = prune_formations(
1852            root.path(),
1853            &current,
1854            project.path(),
1855            &retain(Some(1), Some(DAY_MS / 1000)),
1856            NOW,
1857            |_, _| false,
1858        );
1859        assert_eq!(
1860            removed(&pruned),
1861            [(owned_kept, crate::trace::RETENTION_REASON_MAX_AGE)]
1862        );
1863        for id in [&unmarked, &garbled, &foreign, &current] {
1864            assert!(root.path().join(id.as_str()).is_dir(), "{id}");
1865        }
1866    }
1867
1868    /// A symlink, a regular file, and a directory whose name is not a formation id are never
1869    /// candidates, marker or not; the symlink's target is left alone too.
1870    #[test]
1871    fn prune_formations_ignores_a_symlink_a_regular_file_and_a_name_that_is_not_an_id() {
1872        let root = tempfile::tempdir().unwrap();
1873        let elsewhere = tempfile::tempdir().unwrap();
1874        let project = tempfile::tempdir().unwrap();
1875        let marker = marker_naming(project.path());
1876        let target = make_formation(elsewhere.path(), &frm(NOW - 3 * DAY_MS, 1), Some(&marker));
1877        let linked = frm(NOW - 3 * DAY_MS, 2);
1878        std::os::unix::fs::symlink(&target, root.path().join(linked.as_str())).unwrap();
1879        std::fs::write(root.path().join(frm(NOW - 3 * DAY_MS, 3).as_str()), &marker).unwrap();
1880        for name in ["not-a-formation", "frm_ABC", "frm_0123"] {
1881            std::fs::create_dir(root.path().join(name)).unwrap();
1882            std::fs::write(
1883                root.path()
1884                    .join(name)
1885                    .join(crate::formation_launch::FORMATION_MARKER_FILE),
1886                &marker,
1887            )
1888            .unwrap();
1889        }
1890        let before = names(root.path());
1891
1892        let pruned = prune_formations(
1893            root.path(),
1894            &frm(NOW, 9),
1895            project.path(),
1896            &retain(Some(1), Some(1)),
1897            NOW,
1898            |_, _| false,
1899        );
1900
1901        assert!(pruned.is_empty(), "{pruned:?}");
1902        assert_eq!(names(root.path()), before);
1903        assert!(target.join("worker/.murmur").is_dir());
1904    }
1905
1906    /// A policy with neither key is no policy, and a missing formations directory holds nothing.
1907    #[test]
1908    fn prune_formations_with_neither_key_or_no_directory_removes_nothing() {
1909        let root = tempfile::tempdir().unwrap();
1910        let project = tempfile::tempdir().unwrap();
1911        let old = frm(NOW - 3 * DAY_MS, 1);
1912        make_owned(root.path(), project.path(), &[&old]);
1913
1914        let pruned = prune_formations(
1915            root.path(),
1916            &frm(NOW, 2),
1917            project.path(),
1918            &retain(None, None),
1919            NOW,
1920            |_, _| panic!("nothing is a candidate without a policy"),
1921        );
1922        assert!(pruned.is_empty());
1923        assert!(root.path().join(old.as_str()).is_dir());
1924
1925        assert!(prune_formations(
1926            &root.path().join("missing"),
1927            &frm(NOW, 2),
1928            project.path(),
1929            &retain(Some(1), Some(1)),
1930            NOW,
1931            |_, _| false,
1932        )
1933        .is_empty());
1934    }
1935
1936    /// The project directory is compared canonicalized: a launch through a symlink to the project
1937    /// owns what a launch through its real path marked.
1938    #[test]
1939    fn prune_formations_compares_project_directories_canonically() {
1940        let root = tempfile::tempdir().unwrap();
1941        let project = tempfile::tempdir().unwrap();
1942        let links = tempfile::tempdir().unwrap();
1943        let link = links.path().join("project");
1944        std::os::unix::fs::symlink(project.path(), &link).unwrap();
1945        let old = frm(NOW - 3 * DAY_MS, 1);
1946        make_owned(
1947            root.path(),
1948            &std::fs::canonicalize(project.path()).unwrap(),
1949            &[&old],
1950        );
1951
1952        let pruned = prune_formations(
1953            root.path(),
1954            &frm(NOW, 2),
1955            &link,
1956            &retain(None, Some(1)),
1957            NOW,
1958            |_, _| false,
1959        );
1960        assert_eq!(
1961            removed(&pruned),
1962            [(old, crate::trace::RETENTION_REASON_MAX_AGE)]
1963        );
1964    }
1965
1966    /// A directory that cannot be removed whole is not reported as removed.
1967    #[test]
1968    fn prune_formations_leaves_a_directory_it_could_not_remove_out_of_the_result() {
1969        use std::os::unix::fs::PermissionsExt;
1970
1971        let root = tempfile::tempdir().unwrap();
1972        let project = tempfile::tempdir().unwrap();
1973        let stuck = frm(NOW - 3 * DAY_MS, 1);
1974        let gone = frm(NOW - 3 * DAY_MS + 1, 2);
1975        make_owned(root.path(), project.path(), &[&stuck, &gone]);
1976        let stuck_dir = root.path().join(stuck.as_str());
1977        std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o500)).unwrap();
1978        // Root removes it regardless, so there is nothing to observe.
1979        if std::fs::write(stuck_dir.join("probe"), b"").is_ok() {
1980            std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
1981            return;
1982        }
1983
1984        let pruned = prune_formations(
1985            root.path(),
1986            &frm(NOW, 3),
1987            project.path(),
1988            &retain(None, Some(1)),
1989            NOW,
1990            |_, _| false,
1991        );
1992        std::fs::set_permissions(&stuck_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
1993
1994        assert_eq!(
1995            removed(&pruned),
1996            [(gone, crate::trace::RETENTION_REASON_MAX_AGE)]
1997        );
1998        assert!(stuck_dir.is_dir());
1999    }
2000
2001    // ── The record store, as `mur conversation show` reads it ────────────────
2002
2003    fn harness_entry_json(kind: &str, session_id: &str) -> String {
2004        json!({"type": kind, "session_id": session_id, "harness": "h", "driver": "d",
2005               "created_ms": 1, "updated_ms": 2})
2006        .to_string()
2007    }
2008
2009    /// A context counts with either file; a context with neither, and anything dotted, does not;
2010    /// and an id present in two stores is found in both, store by store.
2011    #[test]
2012    fn the_context_finder_sees_either_file_in_every_store() {
2013        let root = tempfile::tempdir().unwrap();
2014        let dir = |store: &str, context: &str| {
2015            let path = root.path().join(store).join(context);
2016            std::fs::create_dir_all(&path).unwrap();
2017            path
2018        };
2019        std::fs::write(dir("b", "ctx").join(RECORD_FILE_NAME), "").unwrap();
2020        std::fs::write(
2021            dir("a", "ctx").join(HARNESS_SESSION_FILE_NAME),
2022            harness_entry_json(crate::harness_session::HARNESS_SESSION_TYPE, "s"),
2023        )
2024        .unwrap();
2025        dir("c", "ctx");
2026        std::fs::write(dir(".hidden", "ctx").join(RECORD_FILE_NAME), "").unwrap();
2027        std::fs::write(dir("a", ".ctx").join(RECORD_FILE_NAME), "").unwrap();
2028
2029        let found = find_context_dirs_in(root.path(), "ctx");
2030        let stores: Vec<&str> = found.iter().map(|dir| dir.record.as_str()).collect();
2031        assert_eq!(stores, ["a", "b"]);
2032        assert!(found.iter().all(|dir| dir.context_id == "ctx"));
2033        assert_eq!(found[1].path, root.path().join("b").join("ctx"));
2034        assert!(find_context_dirs_in(root.path(), ".ctx").is_empty());
2035        assert!(find_context_dirs_in(root.path(), "other").is_empty());
2036        assert!(find_context_dirs_in(&root.path().join("absent"), "ctx").is_empty());
2037    }
2038
2039    /// Every key of every message is kept; the header is read from line 1 only and is not a
2040    /// message; a torn last line is skipped; and a missing record is no header and no messages.
2041    #[test]
2042    fn the_context_reader_reads_as_the_runtime_reads() {
2043        let dir = tempfile::tempdir().unwrap();
2044        let header = RecordHeader {
2045            kind: crate::conversation::RECORD_HEADER_TYPE.to_string(),
2046            capsule: "owner".to_string(),
2047            created_ms: 7,
2048            truncated: None,
2049        };
2050        let first = json!({"role": "user", "content": "hi", "id": "msg_1", "kept": {"x": 1}});
2051        let second = json!({"role": "assistant", "content": [{"type": "text", "text": "yo"}]});
2052        std::fs::write(
2053            dir.path().join(RECORD_FILE_NAME),
2054            format!(
2055                "{}{first}\n\n{second}\n{}\n{{\"role\": \"us",
2056                header_line(&header).unwrap(),
2057                header_line(&header).unwrap().trim_end()
2058            ),
2059        )
2060        .unwrap();
2061
2062        let contents = read_context_dir(dir.path()).unwrap();
2063        assert_eq!(contents.header, Some(header.clone()));
2064        assert_eq!(contents.messages, vec![first.clone(), second]);
2065        assert_eq!(contents.harness_session, None);
2066
2067        std::fs::write(
2068            dir.path().join(RECORD_FILE_NAME),
2069            format!("{first}\n{}", header_line(&header).unwrap()),
2070        )
2071        .unwrap();
2072        let contents = read_context_dir(dir.path()).unwrap();
2073        assert_eq!(contents.header, None, "a header below line 1 is no header");
2074        assert_eq!(contents.messages, vec![first]);
2075
2076        let empty = tempfile::tempdir().unwrap();
2077        let contents = read_context_dir(empty.path()).unwrap();
2078        assert_eq!(contents.header, None);
2079        assert!(contents.messages.is_empty());
2080
2081        std::fs::create_dir(empty.path().join(RECORD_FILE_NAME)).unwrap();
2082        assert!(
2083            read_context_dir(empty.path()).is_err(),
2084            "an unreadable record"
2085        );
2086    }
2087
2088    /// The harness session is read by the runtime's own rule: no file, a file that is not JSON,
2089    /// another tool's file and an entry naming no session all read as no session.
2090    #[test]
2091    fn the_context_reader_reads_no_session_where_the_runtime_does() {
2092        let dir = tempfile::tempdir().unwrap();
2093        let file = dir.path().join(HARNESS_SESSION_FILE_NAME);
2094        assert_eq!(read_context_dir(dir.path()).unwrap().harness_session, None);
2095
2096        let session_type = crate::harness_session::HARNESS_SESSION_TYPE;
2097        for (body, expected) in [
2098            ("{not json".to_string(), false),
2099            (harness_entry_json("another-tool", "s"), false),
2100            (harness_entry_json(session_type, ""), false),
2101            (harness_entry_json(session_type, "s"), true),
2102        ] {
2103            std::fs::write(&file, &body).unwrap();
2104            let session = read_context_dir(dir.path()).unwrap().harness_session;
2105            assert_eq!(session.is_some(), expected, "{body}");
2106            if let Some(session) = session {
2107                assert_eq!(
2108                    (session.session_id.as_str(), session.harness.as_str()),
2109                    ("s", "h")
2110                );
2111                assert_eq!((session.created_ms, session.updated_ms), (1, 2));
2112            }
2113        }
2114    }
2115}