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            // The rewrite dropped lines a seal covered; the record is sealed again at once.
470            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
479/// Remove every blob in `blobs/` beside the record at `path` that none of its lines' image blocks
480/// names, returning how many were removed. A record with no `blobs/` directory removes nothing; a
481/// name that is not a sha256, and a blob that cannot be removed, are left where they are.
482fn 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
520/// The half of [`truncate_record`] that touches nothing the reader can see: the rewritten record
521/// is on disk under a temporary name in the record's own directory, and the original is still the
522/// original until [`StagedRewrite::commit`] renames over it.
523///
524/// `None` means there was nothing to drop.
525pub(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    // Every line that is a message, by its index in the file. Non-message lines — the header,
540    // and a torn final line — are not messages to any reader and are not counted here either.
541    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            // Cumulative over the record's life: an operator reading `dropped` wants to know how
577            // much of this conversation is gone, not how much the last truncation took.
578            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
613/// A rewrite that exists on disk but has not replaced anything yet.
614///
615/// Staged as `<record dir>/.conversation.jsonl.<pid>.<nonce>.tmp` — in the record's own
616/// directory, so the rename that commits it is within one filesystem and therefore atomic, and
617/// under a name no reader mistakes for a record.
618pub(crate) struct StagedRewrite {
619    target: PathBuf,
620    temp: PathBuf,
621    committed: bool,
622}
623
624impl StagedRewrite {
625    /// Stage `contents` for `target`, with the staged file's mode set explicitly for a target whose
626    /// permissions are part of its contract. `None` leaves the mode to the process umask.
627    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        // The fsync is the point: a rename is only atomic with respect to a file whose bytes are
656        // already durable. Without it a crash can leave the new name pointing at a short file.
657        file.write_all(contents)
658            .and_then(|()| file.sync_all())
659            .map_err(|err| format!("{}: {err}", temp.display()))?;
660        // Re-asserted after the write: `mode` on the open only applies when this call created the
661        // file, and a mode that is part of the target's contract must hold for a staged file an
662        // earlier run left behind too.
663        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
695// ── The record store, as the CLI reads it ────────────────────────────────────
696
697/// Every conversation record on this host, record store by record store.
698///
699/// Ordered by record then context id, so two runs of `mur conversation ls` print the same table.
700/// A host with no `~/.murmur/conversations/` has no records, which is an empty list rather than
701/// an error.
702pub 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
733/// Remove one context directory whole, and report what it held.
734///
735/// The only way to reclaim a record no capsule owns — one written before the header line
736/// existed, whose capsule has since been retired. Automatic pruning never touches those.
737pub 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
752/// Every record that has something to say about `message_id`.
753///
754/// A record answers [`MessageStatus::Present`] when the id is one of its lines, and
755/// [`MessageStatus::Truncated`] when it is not but the record has dropped a message at least as
756/// new as this one — the id's own uuid v7 timestamp against the marker's `last_dropped_id`.
757/// Records with nothing to say are left out, so an empty list is
758/// [`MessageStatus::Unknown`] everywhere.
759///
760/// That classification is the whole reason the marker exists: an artifact that stored
761/// `source_id: msg_X` and now finds nothing must be told the conversation was trimmed, not that
762/// its reference was never real.
763pub 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
808/// Every context directory named `context_id`, in every record store, ordered by record store.
809///
810/// A context counts when its directory holds `conversation.jsonl` or `harness-session.json`, so a
811/// context on `inference.transport: process`, which has only the second, is found as well;
812/// [`list_records`] counts the first alone. An empty list is no such context. The `Err` is `HOME`
813/// being unusable.
814pub 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
838/// What the context directory `dir` holds: the header and messages of its `conversation.jsonl`
839/// and its harness session.
840///
841/// Reads with the runtime's own rules — [`parse_header`] on line 1, [`record_messages`] for the
842/// messages, [`read_entry`] for the harness session — and writes nothing. A missing
843/// `conversation.jsonl` is no header and no messages. The `Err` is `conversation.jsonl` existing
844/// and not being readable, and names the path.
845pub 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
859// ── Shared helpers ───────────────────────────────────────────────────────────
860
861/// `~/.murmur/conversations`, resolved and not created.
862pub fn conversations_root() -> Result<PathBuf, String> {
863    Ok(crate::state_store::murmur_home_dir()?.join(CONVERSATION_ROOT_DIR))
864}
865
866/// Milliseconds since the epoch, for the callers that have no fixture to take a clock from.
867pub 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
874/// Immediate subdirectory names of `dir`, sorted, skipping dotted entries.
875fn 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
889/// Every context directory under one record store, as `(context id, path)`.
890fn 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
900/// Message lines in `path`. The header line is not a message and is never counted, which is what
901/// keeps it out of the `total` every reader of a record reports.
902fn 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
912/// Last write to `path`, in milliseconds since the epoch.
913fn 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    /// A `ses_` name whose uuid-v7 timestamp is exactly `ms`, and whose tail distinguishes it
929    /// from any sibling minted in the same millisecond.
930    fn ses(ms: u64, tail: u64) -> String {
931        format!("ses_{ms:012x}{tail:020x}")
932    }
933
934    /// A `msg_` id whose uuid-v7 timestamp is exactly `ms`.
935    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    /// the count is over every entry present, the running session included,
964    /// so `max_sessions: 3` leaves three directories and one of them is the current one. What went
965    /// went whole — no orphaned `trace.jsonl`, no orphaned `blobs/`.
966    #[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    /// the decision comes out of the id. Every directory here is created in
1002    /// the same instant, so every mtime is equally fresh; the two whose ids encode two hours ago
1003    /// are the two that go.
1004    #[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            &current,
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    /// the two keys are ANDed. An old-id directory never survives on rank, and a recent-id
1040    /// directory outside the count does not survive on age.
1041    #[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            &current,
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    /// the floor. Nothing at or after the current session's own id is ever a candidate, which
1080    /// is also what keeps a sibling capsule's concurrently running session safe with no lock file.
1081    #[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        // A sibling launched a millisecond later, while this session was still staging.
1086        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            &current,
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(&current).join("trace.jsonl").exists());
1108    }
1109
1110    /// A pruned session goes whole: its records directory and its created work folder of the same
1111    /// name both go, and a work folder with no records directory is never a candidate, whatever
1112    /// its name says.
1113    #[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    /// Under `--workdir` retention is handed no work folders: an old session's records go, and a
1145    /// planted `.murmur/<old-id>/` and `<old-id>/` in the operator's folder both survive. Without
1146    /// it, the old session's runtime-created `workdir/<old-id>` goes with its records.
1147    #[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            &current,
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(&current)).unwrap();
1193        let pruned = prune_sessions(
1194            &open(records.path()),
1195            Some(&open(&workdirs)),
1196            &current,
1197            &policy,
1198            NOW,
1199        );
1200        assert_eq!(pruned.len(), 1);
1201        assert!(!workdirs.join(&old).exists());
1202        assert!(workdirs.join(&current).exists());
1203    }
1204
1205    /// A pruned session's temporary folder goes with its records directory, even when a command
1206    /// left a read-only directory holding a file and a directory no one can open in it.
1207    #[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            &current,
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    /// A link planted where a created work folder belongs is refused, not followed: the
1245    /// directory it points at survives the pruning of the session it is named for.
1246    #[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    /// A policy with neither key set — which the manifest parser refuses, but which the runtime
1272    /// must still treat as inert rather than as "prune everything".
1273    #[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    // ── Truncation ───────────────────────────────────────────────────────────
1293
1294    /// A record with `count` messages, owned by `capsule` when `owned`.
1295    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    /// A record line whose tool message carries one image block per sha in `shas`.
1322    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    /// A blob only dropped lines name is removed after the rewrite; one a kept line still names
1334    /// stays, and so does a file whose name is no sha256.
1335    #[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    /// A truncation reseals the record at once: it verifies afterwards, and the lines it kept
1363    /// are byte for byte the lines it had.
1364    #[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    /// A record with no `blobs/` truncates exactly as before and removes nothing.
1400    #[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    /// the newest N survive with their ids intact, the file is a header plus exactly N
1411    /// message lines, and every line parses.
1412    #[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    /// The header does not count as a message: `count_messages` — the same rule
1459    /// `crate::conversation::read_record` applies to `total` — is unchanged by it.
1460    #[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    /// Truncating twice accumulates: `dropped` is what this record has lost over its life, not
1470    /// what the last rewrite took.
1471    #[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    /// A record already at or under the limit is left untouched, byte for byte.
1496    #[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    /// a crash mid-truncate leaves the original intact. The staged rewrite is a temp file in
1507    /// the record's own directory — same filesystem, so the rename that commits it is atomic —
1508    /// and until that rename the original parses whole and holds its original message count.
1509    #[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        // Dropping without committing is the crash: the original stands and no debris is left.
1537        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    // ── Record pruning ───────────────────────────────────────────────────────
1544
1545    /// automatic record pruning touches only a record whose header names this
1546    /// capsule. A record another capsule owns and a record no header owns both survive, however
1547    /// old they are — and the current launch's own context is never removed.
1548    #[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        // Far enough in the future that every mtime is outside a one-second window.
1557        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    /// a record inside the window stays. Age is last write, so a record written a moment ago
1576    /// survives a 90-day policy however long ago its conversation started.
1577    #[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    /// `max_messages` alone never removes a context directory: it is enforced on the record the
1596    /// launch opens, not by the age sweep.
1597    #[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    // ── Formation pruning ────────────────────────────────────────────────────
1616
1617    /// A formation id whose uuid-v7 timestamp is exactly `ms`.
1618    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    /// `<root>/<id>/worker/.murmur/ses_…/trace.jsonl`, with `marker` as `formation.json` when
1630    /// given.
1631    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    /// Owned formations of `project` under `root`, one per id.
1656    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    /// The current formation and every one minted after it stay, under both keys at their
1672    /// tightest and every directory past both.
1673    #[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, &current, &later]);
1681
1682        let pruned = prune_formations(
1683            root.path(),
1684            &current,
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    /// `max_sessions: 3` leaves the three newest of this project's directories, the current one
1703    /// and one minted after it among them, and what went is listed newest first.
1704    #[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    /// Age comes out of the id: every directory here is made in the same instant, and the two
1735    /// whose ids encode two hours ago are the two that go.
1736    #[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, &current],
1748        );
1749
1750        let pruned = prune_formations(
1751            root.path(),
1752            &current,
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    /// A directory past both limits is reported as too old; one past the count alone, as over the
1768    /// count.
1769    #[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, &current]);
1777
1778        let pruned = prune_formations(
1779            root.path(),
1780            &current,
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    /// A live formation is never removed, and still counts: the one ranked behind it goes under
1797    /// `max_sessions: 2` although only two directories are not live.
1798    #[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, &current]);
1806
1807        let asked = std::cell::RefCell::new(Vec::new());
1808        let pruned = prune_formations(
1809            root.path(),
1810            &current,
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        // Ranked behind the live one, a directory inside the count is kept.
1832        let kept = frm(NOW - 2500, 4);
1833        make_owned(root.path(), project.path(), &[&kept]);
1834        let pruned = prune_formations(
1835            root.path(),
1836            &current,
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            &current,
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    /// No marker, a marker that does not parse, and a marker naming another project: none is
1858    /// removed, and none takes a rank from this project's own directories.
1859    #[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, &current],
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            &current,
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            &current,
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, &current] {
1905            assert!(root.path().join(id.as_str()).is_dir(), "{id}");
1906        }
1907    }
1908
1909    /// A symlink, a regular file, and a directory whose name is not a formation id are never
1910    /// candidates, marker or not; the symlink's target is left alone too.
1911    #[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    /// A policy with neither key is no policy, and a missing formations directory holds nothing.
1948    #[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    /// The project directory is compared canonicalized: a launch through a symlink to the project
1978    /// owns what a launch through its real path marked.
1979    #[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    /// A directory that cannot be removed whole is not reported as removed.
2008    #[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        // Root removes it regardless, so there is nothing to observe.
2020        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    // ── The record store, as `mur conversation show` reads it ────────────────
2043
2044    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    /// A context counts with either file; a context with neither, and anything dotted, does not;
2051    /// and an id present in two stores is found in both, store by store.
2052    #[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    /// Every key of every message is kept; the header is read from line 1 only and is not a
2081    /// message; a torn last line is skipped; and a missing record is no header and no messages.
2082    #[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    /// The harness session is read by the runtime's own rule: no file, a file that is not JSON,
2130    /// another tool's file and an entry naming no session all read as no session.
2131    #[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}