Skip to main content

capsule_runtime/
conversation.rs

1//! The durable conversation record: `~/.murmur/conversations/<record>/<context-id>/conversation.jsonl`,
2//! one JSON object per line, in the order the messages entered the context.
3//!
4//! Three properties decide the layout:
5//!
6//! * **Outside every workdir, and never preopened.** A record outlives the session that wrote it,
7//!   so it cannot live in a session directory; and it is the whole conversation, so no artifact
8//!   gets it through a filesystem grant. `murmur:conversation/read` is the only way in.
9//! * **Keyed by capsule and context, not by workdir.** `<record>` defaults to the capsule name and
10//!   is overridable with `context.record_store`; `<context-id>` is the task's context id, which is
11//!   client-supplied over A2A and settable with `mur run --context`. Two runs sharing a context id
12//!   share one record.
13//! * **Append-only, and never fatal.** The runtime is the only writer. A directory that cannot be
14//!   created, a line that cannot be appended and a line that will not parse are each reported once
15//!   and then survived: a capsule whose record is unwritable must still do its work.
16//!
17//! Resolution and creation are separate, as they are for [`crate::state_store`]: [`record_root`]
18//! resolves and validates without touching the filesystem, and only the first append creates
19//! anything.
20
21use std::collections::HashSet;
22use std::io::Write;
23use std::path::{Path, PathBuf};
24
25use serde_json::Value;
26
27use crate::bindings::hook::murmur::conversation::read::{Message, MessagePage};
28use crate::errors::RuntimeError;
29use crate::workdir_fs::WorkdirRoot;
30
31/// Directory under the murmur home holding every record, one subdirectory per record name.
32pub(crate) const CONVERSATION_ROOT_DIR: &str = "conversations";
33
34/// The record file inside `<record>/<context-id>/`.
35pub(crate) const RECORD_FILE_NAME: &str = "conversation.jsonl";
36
37/// The `type` value on a record's header line, and the only value a header is recognised by.
38pub(crate) const RECORD_HEADER_TYPE: &str = "murmur.record";
39
40/// What a context is called in a diagnostic when the task resolved no context id. Shared so the
41/// refusal to resume and the failure to continue a harness session name the same non-context.
42pub(crate) const UNRESOLVED_CONTEXT: &str = "<unresolved>";
43
44/// Mode applied to the conversation root, each record directory and each context directory:
45/// owner-only, because a record is the whole of one capsule's conversation.
46const RECORD_DIR_MODE: u32 = 0o700;
47
48/// Mode every record file is held at, on each append and each rewrite.
49pub(crate) const RECORD_FILE_MODE: u32 = 0o600;
50
51/// Prefix on every host-minted cursor, so a value the host did not mint is refused rather than
52/// parsed into a position.
53const CURSOR_PREFIX: &str = "mc_";
54
55/// Inclusive ceiling on `read-messages`' `limit`. A page is copied into the guest's memory whole,
56/// so the host caps it rather than trusting a caller's number.
57const MAX_PAGE_LIMIT: u32 = 100;
58
59/// What every refusal of a record or context name says. The rule itself is
60/// [`crate::state_store::validate_store_name`] — the one definition of "a single usable path
61/// segment" — and this is the wording an operator gets, which names all of it at once because the
62/// value they wrote is in front of them.
63const SEGMENT_RULE: &str = "must be a single path segment: no '/', no '.' or '..', not absolute, \
64                            and not starting with a dot";
65
66/// What one record has already dropped from its front.
67///
68/// Cumulative over the record's life: `dropped` is every message truncation has ever taken from
69/// this record, not what the last one took.
70#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
71pub struct TruncationMarker {
72    pub dropped: u64,
73    /// The `msg_` id of the oldest message still in the file.
74    pub oldest_surviving_id: String,
75    /// The `msg_` id of the newest message that went. A reference to any message minted at or
76    /// before this one reads as truncated rather than as unknown — see
77    /// [`crate::retention::locate_message`].
78    pub last_dropped_id: String,
79    /// When the most recent truncation ran, in milliseconds since the epoch.
80    pub at_ms: u64,
81}
82
83/// The first line of a record: a JSON object with a `type` key and no `role`.
84///
85/// Not a message: every reader of a record skips a line that is not a JSON object carrying a
86/// string `role`, so a header is invisible to `murmur:conversation/read`, to the `threaded`
87/// reload, and to the `total` a page reports. Not a sidecar: a sidecar needs a second write that
88/// a crash could desynchronise from the rename, where a header makes the whole truncation
89/// exactly one atomic rename of one file.
90///
91/// [`Self::capsule`] is what makes automatic record pruning store-safe: two capsules pointed at
92/// one `context.record_store` each own only the records their own name is on.
93#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
94pub struct RecordHeader {
95    /// Always [`RECORD_HEADER_TYPE`]. A first line carrying anything else is not a header.
96    #[serde(rename = "type")]
97    pub kind: String,
98    /// The capsule that owns this record — `name:` from its manifest, never the record store.
99    pub capsule: String,
100    /// When the header was written, in milliseconds since the epoch. On an adopted record, that
101    /// is the adoption, not the conversation's first message.
102    pub created_ms: u64,
103    /// Absent until this record has been truncated.
104    #[serde(default, skip_serializing_if = "Option::is_none")]
105    pub truncated: Option<TruncationMarker>,
106}
107
108/// `header` as the one line it occupies, newline included.
109pub(crate) fn header_line(header: &RecordHeader) -> Result<String, String> {
110    let mut line = serde_json::to_string(header)
111        .map_err(|err| format!("failed to encode the record header: {err}"))?;
112    line.push('\n');
113    Ok(line)
114}
115
116/// The header `path` carries, or `None` for a record whose first line is not one — an unowned
117/// record, and every file that does not exist.
118pub(crate) fn read_header(path: &Path) -> Option<RecordHeader> {
119    let raw = std::fs::read_to_string(path).ok()?;
120    parse_header(raw.lines().next()?)
121}
122
123/// One line as a record header, or `None` when it is not one.
124pub(crate) fn parse_header(line: &str) -> Option<RecordHeader> {
125    let header: RecordHeader = serde_json::from_str(line).ok()?;
126    (header.kind == RECORD_HEADER_TYPE).then_some(header)
127}
128
129/// The millisecond timestamp inside a `msg_` id, or `None` when the id is not one the runtime
130/// minted. Message ids are uuid v7 in simple form, so the first 12 hex characters are the mint
131/// time — which is what lets a dropped id be placed against a truncation marker without the
132/// messages themselves.
133pub(crate) fn message_id_timestamp_ms(id: &str) -> Option<u64> {
134    crate::retention::uuid_v7_timestamp_ms(id, "msg_")
135}
136
137/// Whether `value` is usable as one directory segment of a record path.
138///
139/// `field` names where the value came from (`context.record_store`, `--context`), because that is
140/// what an operator has to go and change. Refused at staging, before anything is created.
141pub(crate) fn validate_record_segment(field: &str, value: &str) -> Result<(), RuntimeError> {
142    crate::state_store::validate_store_name(value).map_err(|_| {
143        RuntimeError::InvalidConversationRecord {
144            field: field.to_string(),
145            value: value.to_string(),
146            message: SEGMENT_RULE.to_string(),
147        }
148    })
149}
150
151/// The record name a capsule's `context:` block resolves to: what it declared, or the capsule
152/// name when it declared nothing. `None` when `context.record: off` — the switch wins over a
153/// `record_store` declared beside it.
154pub(crate) fn resolve_record_name(
155    context: Option<&murmur_artifact::ContextConfig>,
156    capsule_name: &str,
157) -> Option<String> {
158    match context {
159        Some(context) if !context.record => None,
160        Some(context) => Some(
161            context
162                .record_store
163                .clone()
164                .unwrap_or_else(|| capsule_name.to_string()),
165        ),
166        None => Some(capsule_name.to_string()),
167    }
168}
169
170/// Where `record`'s conversations live. Resolves only — nothing is created and nothing is checked
171/// for existence — so a launch that never appends a message leaves no directory behind.
172pub(crate) fn record_root(record: &str) -> Result<PathBuf, String> {
173    validate_record_segment("context.record_store", record).map_err(|err| err.to_string())?;
174    Ok(crate::state_store::murmur_home_dir()?
175        .join(CONVERSATION_ROOT_DIR)
176        .join(record))
177}
178
179/// The record file one context writes under `root`. Resolves only, like [`record_root`]: nothing
180/// is created and nothing is checked for existence.
181///
182/// `mur run --resume` asks whether this path exists before staging goes any further, which is the
183/// one question that separates "continue this conversation" from "start fresh and say nothing".
184pub(crate) fn record_file(root: &Path, context_id: &str) -> PathBuf {
185    root.join(context_id).join(RECORD_FILE_NAME)
186}
187
188/// One context's durable record, and the ids it already holds.
189///
190/// Held by the agent loop for one attempt. Every message the loop puts in the context goes through
191/// [`Self::append`], which is what makes the record the loop's own message list rather than a
192/// summary of it.
193pub(crate) struct ConversationRecord {
194    /// `<home>/.murmur/conversations/<record>/<context-id>`.
195    dir: PathBuf,
196    /// Session directory the failures are reported into, beside stderr.
197    workdir: WorkdirRoot,
198    /// The capsule this record's header names, or `None` for a capsule that declares no
199    /// `context.retain` — which writes no header at all. See [`Self::ensure_header`].
200    capsule: Option<String>,
201    /// Whether the header has been settled for this instance. One check per attempt, not one per
202    /// message: the header is written or adopted at most once and then never looked at again.
203    header_ensured: bool,
204    /// Ids already on disk: those loaded from the record and those this attempt appended. A
205    /// compaction hook that hands a message back verbatim returns its id with it, and a threaded
206    /// reload starts from messages that are already lines — neither may be written twice.
207    recorded: HashSet<String>,
208    /// A write failure is reported once, then survived silently: one full disk must not produce
209    /// one stderr line per message.
210    failure_reported: bool,
211    /// A malformed line is reported once for this record, however many times it is read.
212    malformed_reported: bool,
213}
214
215impl ConversationRecord {
216    /// The record for `context_id` under `root`, or `None` when the context id is not a usable
217    /// path segment.
218    ///
219    /// A context id arrives from an A2A client as well as from `mur run --context`, so it is
220    /// checked here rather than trusted: a rejected id costs that conversation its record and
221    /// nothing else.
222    pub(crate) fn open(
223        root: &Path,
224        context_id: &str,
225        workdir: &WorkdirRoot,
226        capsule: Option<&str>,
227    ) -> Option<Self> {
228        if crate::state_store::validate_store_name(context_id).is_err() {
229            report(
230                workdir,
231                &format!(
232                    "[conversation] context id '{context_id}' {SEGMENT_RULE}; this conversation \
233                     is not recorded"
234                ),
235            );
236            return None;
237        }
238        Some(Self {
239            dir: root.join(context_id),
240            workdir: workdir.clone(),
241            capsule: capsule.map(str::to_string),
242            header_ensured: false,
243            recorded: HashSet::new(),
244            failure_reported: false,
245            malformed_reported: false,
246        })
247    }
248
249    pub(crate) fn path(&self) -> PathBuf {
250        self.dir.join(RECORD_FILE_NAME)
251    }
252
253    /// Every message the record already holds, oldest first, and never appended again.
254    ///
255    /// The agent loop starts a `threaded` task's message list from this. Each message keeps the
256    /// `id` its line carries, byte for byte.
257    pub(crate) fn load(&mut self) -> Vec<Value> {
258        let path = self.path();
259        let messages = match read_record(&path, &self.workdir, &mut self.malformed_reported) {
260            Ok(messages) => messages,
261            Err(reason) => {
262                report(
263                    &self.workdir,
264                    &format!(
265                        "[conversation] {} could not be read ({reason}); this task starts with no \
266                         prior messages",
267                        path.display()
268                    ),
269                );
270                Vec::new()
271            }
272        };
273        for message in &messages {
274            if let Some(id) = crate::agent::message_id(message) {
275                self.recorded.insert(id.to_string());
276            }
277        }
278        messages
279    }
280
281    /// Append every message not already in this record, in the order given.
282    ///
283    /// Called at each point a message enters the agent loop's context. A message whose id is
284    /// already recorded is skipped, which is what a compaction hook returning part of the context
285    /// verbatim produces.
286    ///
287    /// An id joins `recorded` only once its line is on disk. A failure the caller survives — a
288    /// disk that fills and is freed again, a descriptor limit that passes — is followed by an
289    /// append carrying the same messages, and a message marked as written before its write
290    /// succeeded would be skipped by that retry and lost from the record for good.
291    pub(crate) fn append(&mut self, messages: &[Value]) {
292        for message in messages {
293            let id = crate::agent::message_id(message).map(str::to_string);
294            if let Some(id) = id.as_deref() {
295                if self.recorded.contains(id) {
296                    continue;
297                }
298            }
299            if let Err(reason) = self.append_line(message) {
300                if !self.failure_reported {
301                    self.failure_reported = true;
302                    report(
303                        &self.workdir,
304                        &format!(
305                            "[conversation] {} could not be written ({reason}); this task \
306                             continues unrecorded",
307                            self.path().display()
308                        ),
309                    );
310                }
311                return;
312            }
313            if let Some(id) = id {
314                self.recorded.insert(id);
315            }
316        }
317    }
318
319    /// Create the directory chain if it is missing and append one line.
320    ///
321    /// Creation happens here rather than at staging so `context.record: off`, and a launch that
322    /// never reaches a message, both leave nothing behind.
323    fn append_line(&mut self, message: &Value) -> Result<(), String> {
324        let mut line = serde_json::to_string(message)
325            .map_err(|err| format!("failed to encode the message: {err}"))?;
326        line.push('\n');
327
328        self.ensure_dirs()?;
329        self.ensure_header()?;
330
331        let mut file = open_record_file(&self.path(), false)?;
332        file.write_all(line.as_bytes())
333            .map_err(|err| err.to_string())
334    }
335
336    fn ensure_dirs(&self) -> Result<(), String> {
337        ensure_context_dirs(&self.dir)
338    }
339
340    /// Put a [`RecordHeader`] at the front of this record, once, before its first new line.
341    ///
342    /// A capsule that declares no `context.retain` writes no header: there is nothing to own a
343    /// record for, and a record that gains a line on upgrade is a behaviour change an operator
344    /// did not ask for. Everything below is therefore conditional on [`Self::capsule`].
345    ///
346    /// Three cases, and the third is the whole reason this is not just a write:
347    ///
348    /// * A record that does not exist yet gets its header as its first line.
349    /// * A record that already carries one is left exactly as it is.
350    /// * A record written before the header existed is **adopted**: the header is prepended
351    ///   through the same atomic temp-and-rename rewrite a truncation uses, so a crash leaves the
352    ///   original whole. Until that adoption the record is unowned and automatic pruning skips
353    ///   it, however old it is.
354    ///
355    /// Adoption is what brings a headerless record under a policy without ever guessing at
356    /// ownership: the capsule that writes to a record is the capsule that owns it.
357    fn ensure_header(&mut self) -> Result<(), String> {
358        if self.header_ensured {
359            return Ok(());
360        }
361        let Some(capsule) = self.capsule.clone() else {
362            self.header_ensured = true;
363            return Ok(());
364        };
365        let path = self.path();
366        let existing = match std::fs::read_to_string(&path) {
367            Ok(raw) => raw,
368            Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
369            Err(err) => return Err(format!("{}: {err}", path.display())),
370        };
371        if existing.lines().next().and_then(parse_header).is_some() {
372            self.header_ensured = true;
373            return Ok(());
374        }
375
376        let header = RecordHeader {
377            kind: RECORD_HEADER_TYPE.to_string(),
378            capsule,
379            created_ms: crate::retention::now_ms(),
380            truncated: None,
381        };
382        let line = header_line(&header)?;
383        // The flag is set only once the header is on disk: a failure the caller survives is
384        // followed by an append carrying the same messages, and a record marked as headered
385        // before its header was written would take those messages headerless and stay unowned.
386        if existing.is_empty() {
387            open_record_file(&path, true)?
388                .write_all(line.as_bytes())
389                .map_err(|err| format!("{}: {err}", path.display()))?;
390        } else {
391            let mut contents = line;
392            contents.push_str(&existing);
393            crate::retention::StagedRewrite::stage_with_mode(
394                &path,
395                contents.as_bytes(),
396                Some(RECORD_FILE_MODE),
397            )?
398            .commit()?;
399        }
400        self.header_ensured = true;
401        Ok(())
402    }
403}
404
405/// Open a record file for appending, or for replacing when `truncate`, held at
406/// [`RECORD_FILE_MODE`] whether or not this call created it.
407fn open_record_file(path: &Path, truncate: bool) -> Result<std::fs::File, String> {
408    use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
409
410    let mut options = std::fs::OpenOptions::new();
411    options.create(true).mode(RECORD_FILE_MODE);
412    if truncate {
413        options.write(true).truncate(true);
414    } else {
415        options.append(true);
416    }
417    let file = options
418        .open(path)
419        .map_err(|err| format!("{}: {err}", path.display()))?;
420    file.set_permissions(std::fs::Permissions::from_mode(RECORD_FILE_MODE))
421        .map_err(|err| format!("{}: {err}", path.display()))?;
422    Ok(file)
423}
424
425/// The parsed record one reader holds between pages.
426///
427/// A hook pages a record it does not shrink and cannot write, so re-reading and re-parsing the
428/// whole file for each page makes a walk quadratic in a file designed to grow without bound. One
429/// cache serves a whole launch: [`crate::conversation_import::ConversationState`] holds it across
430/// every task, which is why the key carries the path as well as the length.
431#[derive(Default)]
432pub(crate) struct RecordCache {
433    /// Record the messages below were parsed from, and `None` while no parse is trusted.
434    path: Option<PathBuf>,
435    /// Length of that file at the moment it was stat'd for the parse below.
436    len: u64,
437    /// The record, oldest first, as [`read_record`] returned it.
438    messages: Vec<Value>,
439    /// A malformed line is reported once for this reader, however many pages it walks.
440    malformed_reported: bool,
441    /// Reads of the record file that reached the filesystem. What pins the cost of a paging loop.
442    #[cfg(test)]
443    reads: u32,
444}
445
446impl RecordCache {
447    /// The whole record, oldest first, parsed at most once per length the file has held.
448    ///
449    /// The runtime is the record's only writer and only ever appends, so a file whose length is
450    /// unchanged holds the same bytes and the parse above it still stands. A missing file is an
451    /// empty record of length 0, exactly as a read of one is.
452    fn messages(&mut self, path: &Path, workdir: &WorkdirRoot) -> Result<&[Value], String> {
453        let len = match std::fs::metadata(path) {
454            Ok(metadata) => metadata.len(),
455            Err(err) if err.kind() == std::io::ErrorKind::NotFound => 0,
456            Err(err) => return Err(err.to_string()),
457        };
458        if self.path.as_deref() == Some(path) && self.len == len {
459            return Ok(&self.messages);
460        }
461
462        // Cleared before the read and set again only after the messages are in place, so a read
463        // that fails leaves a cache that re-reads rather than one record's messages under
464        // another's key.
465        self.path = None;
466        #[cfg(test)]
467        {
468            self.reads += 1;
469        }
470        self.messages = read_record(path, workdir, &mut self.malformed_reported)?;
471        self.len = len;
472        self.path = Some(path.to_path_buf());
473        Ok(&self.messages)
474    }
475
476    #[cfg(test)]
477    pub(crate) fn reads(&self) -> u32 {
478        self.reads
479    }
480}
481
482/// One page of a record, newest first, as `murmur:conversation/read` serves it.
483///
484/// `path` is `None` for a session that writes no record — `context.record: off`, a
485/// `process`-transport capsule, or a task whose context id was refused — and reads as an empty
486/// page rather than an error: there is nothing wrong with a conversation that has said nothing
487/// yet. That case leaves `cache` alone, so a hook dispatched between two tasks does not cost the
488/// next one its parse.
489pub(crate) fn page(
490    cache: &mut RecordCache,
491    path: Option<&Path>,
492    workdir: &WorkdirRoot,
493    cursor: Option<&str>,
494    limit: u32,
495) -> Result<MessagePage, String> {
496    let Some(path) = path else {
497        return Ok(empty_page());
498    };
499    let messages = cache
500        .messages(path, workdir)
501        .map_err(|reason| format!("unavailable: {reason}"))?;
502
503    // A cursor is the exclusive upper bound of the next page, counted from the oldest message, so
504    // it stays valid while the runtime appends to the record underneath a paging hook.
505    let end = match cursor {
506        None => messages.len(),
507        Some(cursor) => decode_cursor(cursor)
508            .filter(|end| *end <= messages.len())
509            .ok_or_else(|| "invalid-cursor".to_string())?,
510    };
511    let limit = limit.clamp(1, MAX_PAGE_LIMIT) as usize;
512    let start = end.saturating_sub(limit);
513
514    let mut window: Vec<Value> = messages[start..end].to_vec();
515    window.reverse();
516    // `murmur:conversation/read` re-uses `murmur:hook/lifecycle`'s `message`, but bindgen
517    // generates the imported and exported views of it as two Rust types. The lowering itself is
518    // `to_wit_messages`, so a "tool" message reaches a reader folded exactly as it reaches a
519    // compaction hook.
520    Ok(MessagePage {
521        messages: crate::agent::to_wit_messages(&window)
522            .into_iter()
523            .map(|message| Message {
524                role: message.role,
525                content: message.content,
526                id: message.id,
527                source_id: message.source_id,
528                inserted_by: message.inserted_by.map(|mark| {
529                    crate::agent::ContextInsertion::from_lowered(mark).to_imported_wit()
530                }),
531            })
532            .collect(),
533        next_cursor: (start > 0).then(|| encode_cursor(start)),
534        total: u32::try_from(messages.len()).unwrap_or(u32::MAX),
535    })
536}
537
538/// What a record with nothing in it reads as.
539fn empty_page() -> MessagePage {
540    MessagePage {
541        messages: Vec::new(),
542        next_cursor: None,
543        total: 0,
544    }
545}
546
547/// Every message in `path`, oldest first.
548///
549/// The single parser both readers go through — the `murmur:conversation/read` page and the
550/// `threaded` reload — so a line the hook can see is a line the next task will load. A line that
551/// is not a JSON object with a string `role` is skipped; a truncated final line is exactly that
552/// case. A missing file is an empty record, not an error.
553///
554/// `reported` is the reader's own once-only flag: the first skipped line it meets is reported
555/// naming the path and the 1-based line number, and every line after it, in this read and in the
556/// reader's later ones, is silent. One such line is worth one log entry however many times the
557/// reader goes back to the file.
558fn read_record(
559    path: &Path,
560    workdir: &WorkdirRoot,
561    reported: &mut bool,
562) -> Result<Vec<Value>, String> {
563    let raw = match std::fs::read_to_string(path) {
564        Ok(raw) => raw,
565        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
566        Err(err) => return Err(err.to_string()),
567    };
568
569    let mut messages = Vec::new();
570    for (index, line) in raw.lines().enumerate() {
571        if line.trim().is_empty() {
572            continue;
573        }
574        match parse_message_line(line) {
575            Some(message) => messages.push(message),
576            None => {
577                if !*reported {
578                    *reported = true;
579                    report(
580                        workdir,
581                        &format!(
582                            "[conversation] {}: line {} is not a message and was skipped",
583                            path.display(),
584                            index + 1
585                        ),
586                    );
587                }
588            }
589        }
590    }
591    Ok(messages)
592}
593
594/// One record line as a message, or `None` when it is not one. A message is a JSON object
595/// carrying a string `role`: everything the runtime writes is, and nothing else can be lowered
596/// into the WIT `message` record. The header line is the deliberate non-message — it is how a
597/// truncation marker rides in the file without being visible to anything that reads messages.
598pub(crate) fn parse_message_line(line: &str) -> Option<Value> {
599    let value: Value = serde_json::from_str(line).ok()?;
600    value
601        .get("role")
602        .and_then(Value::as_str)
603        .is_some()
604        .then_some(value)
605}
606
607fn encode_cursor(position: usize) -> String {
608    format!("{CURSOR_PREFIX}{position}")
609}
610
611fn decode_cursor(cursor: &str) -> Option<usize> {
612    cursor.strip_prefix(CURSOR_PREFIX)?.parse().ok()
613}
614
615/// Create the conversation root, the record directory and `dir` itself if they are missing,
616/// oldest ancestor first, so a context directory is never reachable through a parent a wider mode
617/// left open.
618///
619/// When the conversation root sits in `~/.murmur`, the home itself is held at `0700` first.
620///
621/// `dir` is `<conversation root>/<record>/<context-id>` — the directory holding this context's
622/// record, and the one [`crate::harness_session`] puts its map file in for a capsule whose harness
623/// owns the conversation instead.
624pub(crate) fn ensure_context_dirs(dir: &Path) -> Result<(), String> {
625    let mut chain = Vec::new();
626    let mut ancestor = Some(dir);
627    for _ in 0..3 {
628        if let Some(current) = ancestor {
629            chain.push(current);
630            ancestor = current.parent();
631        }
632    }
633    if let (Some(owner), Ok(home)) = (ancestor, crate::state_store::murmur_home_dir()) {
634        if owner == home {
635            crate::murmur_home::ensure_murmur_home()?;
636        }
637    }
638    for path in chain.into_iter().rev() {
639        crate::state_store::ensure_private_dir(path, RECORD_DIR_MODE)
640            .map_err(|reason| format!("{}: {reason}", path.display()))?;
641    }
642    Ok(())
643}
644
645/// Both places an operator looks for what the runtime did with a record. Never fatal.
646pub(crate) fn report(workdir: &WorkdirRoot, message: &str) {
647    crate::runtime_err!("[capsule-runtime] {message}");
648    crate::agent::append_bootstrap_log(workdir, message);
649}
650
651#[cfg(test)]
652mod tests {
653    use super::*;
654    use serde_json::json;
655
656    /// The capsule every record below is opened for, and therefore the name its header carries.
657    const CAPSULE: &str = "capsule";
658
659    fn message(id: &str, role: &str, text: &str) -> Value {
660        json!({"role": role, "content": [{"type": "text", "text": text}], "id": id})
661    }
662
663    fn id_at(id: usize) -> String {
664        format!("msg_{:032x}", id)
665    }
666
667    #[test]
668    fn record_file_is_held_owner_only_on_append() {
669        let home = tempfile::tempdir().unwrap();
670        let workdir = tempfile::tempdir().unwrap();
671        let root = home.path().join("conversations/shared");
672        crate::murmur_home::wide_dir(&root.join("ctx_1"), 0o755);
673        let path = root.join("ctx_1").join(RECORD_FILE_NAME);
674        crate::murmur_home::wide_file(&path, "", 0o644);
675
676        let mut record = ConversationRecord::open(
677            &root,
678            "ctx_1",
679            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
680            None,
681        )
682        .unwrap();
683        record.append(&[message(&id_at(1), "user", "hello")]);
684
685        assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
686        assert_eq!(crate::murmur_home::mode_of(&root.join("ctx_1")), 0o700);
687        assert_eq!(std::fs::read_to_string(&path).unwrap().lines().count(), 1);
688    }
689
690    #[test]
691    fn record_file_is_created_owner_only_with_its_header() {
692        let home = tempfile::tempdir().unwrap();
693        let workdir = tempfile::tempdir().unwrap();
694        let root = home.path().join("conversations/shared");
695
696        let mut record = ConversationRecord::open(
697            &root,
698            "ctx_1",
699            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
700            Some(CAPSULE),
701        )
702        .unwrap();
703        record.append(&[message(&id_at(1), "user", "hello")]);
704
705        assert_eq!(crate::murmur_home::mode_of(&record.path()), 0o600);
706    }
707
708    /// Adopting a headerless record rewrites it by rename; the replacement is owner-only too.
709    #[test]
710    fn record_rewrite_keeps_the_file_owner_only() {
711        let home = tempfile::tempdir().unwrap();
712        let workdir = tempfile::tempdir().unwrap();
713        let root = home.path().join("conversations/shared");
714        crate::murmur_home::wide_dir(&root.join("ctx_1"), 0o755);
715        let path = root.join("ctx_1").join(RECORD_FILE_NAME);
716        let old = serde_json::to_string(&message(&id_at(1), "user", "old")).unwrap();
717        crate::murmur_home::wide_file(&path, &format!("{old}\n"), 0o644);
718
719        let mut record = ConversationRecord::open(
720            &root,
721            "ctx_1",
722            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
723            Some(CAPSULE),
724        )
725        .unwrap();
726        record.append(&[message(&id_at(2), "user", "new")]);
727
728        assert!(read_header(&path).is_some(), "the record was adopted");
729        assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
730    }
731
732    #[test]
733    fn record_root_under_a_wide_murmur_home_holds_the_home_owner_only() {
734        let home = tempfile::tempdir().unwrap();
735        crate::murmur_home::wide_dir(&home.path().join(".murmur"), 0o755);
736        crate::murmur_home::run_with_home(
737            "conversation::tests::inner_append_under_scratch_home",
738            home.path(),
739        );
740        assert_eq!(
741            crate::murmur_home::mode_of(&home.path().join(".murmur")),
742            0o700
743        );
744        let record = home
745            .path()
746            .join(".murmur/conversations/shared/ctx_1")
747            .join(RECORD_FILE_NAME);
748        assert_eq!(crate::murmur_home::mode_of(&record), 0o600);
749    }
750
751    #[test]
752    #[ignore = "run by record_root_under_a_wide_murmur_home_holds_the_home_owner_only"]
753    fn inner_append_under_scratch_home() {
754        if !crate::murmur_home::in_scratch_home() {
755            return;
756        }
757        let workdir = tempfile::tempdir().unwrap();
758        let root = record_root("shared").unwrap();
759        let mut record = ConversationRecord::open(
760            &root,
761            "ctx_1",
762            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
763            None,
764        )
765        .unwrap();
766        record.append(&[message(&id_at(1), "user", "hello")]);
767        assert!(record.path().exists());
768    }
769
770    /// The header is the first line a new record gets, and it names the capsule that wrote it —
771    /// which is what makes automatic pruning store-safe when two capsules share one record store.
772    #[test]
773    fn the_first_append_writes_a_header_naming_the_capsule() {
774        let home = tempfile::tempdir().unwrap();
775        let workdir = tempfile::tempdir().unwrap();
776        let root = home.path().join("conversations/shared");
777
778        let mut record = ConversationRecord::open(
779            &root,
780            "ctx_1",
781            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
782            Some(CAPSULE),
783        )
784        .unwrap();
785        record.append(&[message(&id_at(1), "user", "hello")]);
786
787        let raw = std::fs::read_to_string(record.path()).unwrap();
788        let lines: Vec<&str> = raw.lines().collect();
789        assert_eq!(lines.len(), 2, "a header and one message: {lines:?}");
790        let header = parse_header(lines[0]).expect("the first line is a header");
791        assert_eq!(header.kind, RECORD_HEADER_TYPE);
792        assert_eq!(header.capsule, CAPSULE);
793        assert!(header.truncated.is_none(), "nothing has been dropped yet");
794        assert!(header.created_ms > 0);
795    }
796
797    /// The header is invisible to everything that reads messages: `read_record` skips it, so the
798    /// `total` a `murmur:conversation/read` page reports counts messages and nothing else.
799    #[test]
800    fn a_header_line_changes_neither_the_loaded_messages_nor_total() {
801        let home = tempfile::tempdir().unwrap();
802        let workdir = tempfile::tempdir().unwrap();
803        let root = home.path().join("conversations/c");
804
805        let mut record = ConversationRecord::open(
806            &root,
807            "ctx_1",
808            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
809            Some(CAPSULE),
810        )
811        .unwrap();
812        record.append(&[
813            message(&id_at(1), "user", "one"),
814            message(&id_at(2), "assistant", "two"),
815        ]);
816        let path = record.path();
817
818        let mut reader = ConversationRecord::open(
819            &root,
820            "ctx_1",
821            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
822            Some(CAPSULE),
823        )
824        .unwrap();
825        let loaded = reader.load();
826        assert_eq!(loaded.len(), 2, "the header is not a message: {loaded:?}");
827
828        let mut cache = RecordCache::default();
829        let page = page(
830            &mut cache,
831            Some(&path),
832            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
833            None,
834            100,
835        )
836        .unwrap();
837        assert_eq!(page.total, 2);
838        assert_eq!(page.messages.len(), 2);
839    }
840
841    /// A record written before the header existed is unowned, and is adopted on the next append
842    /// by the capsule that writes to it — through the same atomic rewrite a truncation uses, so
843    /// every line it already held survives byte for byte.
844    #[test]
845    fn a_headerless_record_is_adopted_on_the_next_append() {
846        let home = tempfile::tempdir().unwrap();
847        let workdir = tempfile::tempdir().unwrap();
848        let root = home.path().join("conversations/c");
849        let dir = root.join("ctx_1");
850        std::fs::create_dir_all(&dir).unwrap();
851        let path = dir.join(RECORD_FILE_NAME);
852        let headerless = format!(
853            "{}
854{}
855",
856            serde_json::to_string(&message(&id_at(1), "user", "one")).unwrap(),
857            serde_json::to_string(&message(&id_at(2), "assistant", "two")).unwrap()
858        );
859        std::fs::write(&path, &headerless).unwrap();
860        assert!(read_header(&path).is_none(), "unowned before the append");
861
862        let mut record = ConversationRecord::open(
863            &root,
864            "ctx_1",
865            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
866            Some(CAPSULE),
867        )
868        .unwrap();
869        record.append(&[message(&id_at(3), "user", "three")]);
870
871        let raw = std::fs::read_to_string(&path).unwrap();
872        assert_eq!(read_header(&path).unwrap().capsule, CAPSULE);
873        assert!(
874            raw.contains(headerless.trim_end()),
875            "every line it already held survives byte for byte: {raw}"
876        );
877        assert_eq!(raw.lines().count(), 4, "a header and three messages");
878    }
879
880    /// A record already carrying a header is left exactly as it is — the header is written once,
881    /// not once per launch, and a truncation marker on it is never cleared by an append.
882    #[test]
883    fn an_owned_record_keeps_the_header_it_already_has() {
884        let home = tempfile::tempdir().unwrap();
885        let workdir = tempfile::tempdir().unwrap();
886        let root = home.path().join("conversations/c");
887        let dir = root.join("ctx_1");
888        std::fs::create_dir_all(&dir).unwrap();
889        let path = dir.join(RECORD_FILE_NAME);
890        let header = RecordHeader {
891            kind: RECORD_HEADER_TYPE.to_string(),
892            capsule: "someone-else".to_string(),
893            created_ms: 1_756_400_000_000,
894            truncated: Some(TruncationMarker {
895                dropped: 5,
896                oldest_surviving_id: id_at(6),
897                last_dropped_id: id_at(5),
898                at_ms: 1_756_400_000_001,
899            }),
900        };
901        std::fs::write(&path, header_line(&header).unwrap()).unwrap();
902
903        let mut record = ConversationRecord::open(
904            &root,
905            "ctx_1",
906            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
907            Some(CAPSULE),
908        )
909        .unwrap();
910        record.append(&[message(&id_at(7), "user", "seven")]);
911
912        assert_eq!(read_header(&path).unwrap(), header);
913    }
914
915    /// A record written by one session and read back by the next: the ids are the same bytes, and
916    /// nothing loaded is appended a second time.
917    #[test]
918    fn a_reloaded_message_keeps_its_id_and_is_not_written_twice() {
919        let home = tempfile::tempdir().unwrap();
920        let workdir = tempfile::tempdir().unwrap();
921        let root = home.path().join("conversations/capsule");
922
923        let mut first = ConversationRecord::open(
924            &root,
925            "ctx_fixed",
926            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
927            Some(CAPSULE),
928        )
929        .unwrap();
930        first.append(&[message(&id_at(1), "user", "hello")]);
931
932        let mut second = ConversationRecord::open(
933            &root,
934            "ctx_fixed",
935            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
936            Some(CAPSULE),
937        )
938        .unwrap();
939        let loaded = second.load();
940        assert_eq!(loaded.len(), 1);
941        assert_eq!(
942            crate::agent::message_id(&loaded[0]),
943            Some(id_at(1).as_str())
944        );
945
946        second.append(&loaded);
947        second.append(&[message(&id_at(2), "assistant", "hi")]);
948        assert_eq!(
949            second.load().len(),
950            2,
951            "a reloaded message is never re-appended"
952        );
953    }
954
955    /// Every directory on the record path is owner-only, and none of them exists before the first
956    /// message is appended.
957    #[test]
958    fn the_directory_chain_is_created_at_0700_on_first_write() {
959        use std::os::unix::fs::PermissionsExt;
960
961        let home = tempfile::tempdir().unwrap();
962        let workdir = tempfile::tempdir().unwrap();
963        let conversations = home.path().join("conversations");
964        let root = conversations.join("capsule");
965
966        let mut record = ConversationRecord::open(
967            &root,
968            "ctx_1",
969            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
970            Some(CAPSULE),
971        )
972        .unwrap();
973        assert!(!conversations.exists(), "opening a record creates nothing");
974
975        record.append(&[message(&id_at(1), "user", "hello")]);
976        for dir in [&conversations, &root, &root.join("ctx_1")] {
977            let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777;
978            assert_eq!(
979                mode,
980                0o700,
981                "{} must be 0700, got {mode:04o}",
982                dir.display()
983            );
984        }
985    }
986
987    /// A context id that is not one path segment costs that conversation its record, and refuses
988    /// before any path is built from it.
989    #[test]
990    fn a_context_id_that_is_not_a_segment_records_nothing() {
991        let home = tempfile::tempdir().unwrap();
992        let workdir = tempfile::tempdir().unwrap();
993        for context_id in ["../escape", "a/b", "", ".hidden"] {
994            assert!(
995                ConversationRecord::open(
996                    &home.path().join("conversations/c"),
997                    context_id,
998                    &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
999                    Some(CAPSULE)
1000                )
1001                .is_none(),
1002                "'{context_id}' must not open a record"
1003            );
1004        }
1005        assert!(!home.path().join("conversations").exists());
1006    }
1007
1008    /// Newest first, paged, with a cursor that neither repeats nor skips a message.
1009    #[test]
1010    fn paging_walks_the_record_newest_first_without_repeating_a_message() {
1011        let home = tempfile::tempdir().unwrap();
1012        let workdir = tempfile::tempdir().unwrap();
1013        let root = home.path().join("conversations/capsule");
1014        let mut record = ConversationRecord::open(
1015            &root,
1016            "ctx_1",
1017            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1018            Some(CAPSULE),
1019        )
1020        .unwrap();
1021        let written: Vec<Value> = (1..=5)
1022            .map(|n| message(&id_at(n), "user", &format!("m{n}")))
1023            .collect();
1024        record.append(&written);
1025        let path = record.path();
1026
1027        let mut seen = Vec::new();
1028        let mut cursor: Option<String> = None;
1029        loop {
1030            let page = page(
1031                &mut RecordCache::default(),
1032                Some(&path),
1033                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1034                cursor.as_deref(),
1035                2,
1036            )
1037            .unwrap();
1038            assert_eq!(page.total, 5);
1039            seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1040            match page.next_cursor {
1041                Some(next) => cursor = Some(next),
1042                None => break,
1043            }
1044        }
1045        assert_eq!(
1046            seen,
1047            (1..=5).rev().map(id_at).collect::<Vec<_>>(),
1048            "the whole record, newest first, exactly once each"
1049        );
1050    }
1051
1052    /// `limit` is the host's to bound: `0` still makes progress and a huge value cannot ask for
1053    /// more than the ceiling.
1054    #[test]
1055    fn limit_is_clamped_at_both_ends() {
1056        let home = tempfile::tempdir().unwrap();
1057        let workdir = tempfile::tempdir().unwrap();
1058        let root = home.path().join("conversations/capsule");
1059        let mut record = ConversationRecord::open(
1060            &root,
1061            "ctx_1",
1062            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1063            Some(CAPSULE),
1064        )
1065        .unwrap();
1066        record.append(
1067            &(1..=120)
1068                .map(|n| message(&id_at(n), "user", "m"))
1069                .collect::<Vec<_>>(),
1070        );
1071        let path = record.path();
1072
1073        assert_eq!(
1074            page(
1075                &mut RecordCache::default(),
1076                Some(&path),
1077                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1078                None,
1079                0
1080            )
1081            .unwrap()
1082            .messages
1083            .len(),
1084            1
1085        );
1086        assert_eq!(
1087            page(
1088                &mut RecordCache::default(),
1089                Some(&path),
1090                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1091                None,
1092                u32::MAX
1093            )
1094            .unwrap()
1095            .messages
1096            .len(),
1097            MAX_PAGE_LIMIT as usize
1098        );
1099    }
1100
1101    /// A value that is not `mc_<number>`, and a position past the end of the record, are both
1102    /// `invalid-cursor` rather than a page.
1103    #[test]
1104    fn a_foreign_cursor_is_refused() {
1105        let home = tempfile::tempdir().unwrap();
1106        let workdir = tempfile::tempdir().unwrap();
1107        let root = home.path().join("conversations/capsule");
1108        let mut record = ConversationRecord::open(
1109            &root,
1110            "ctx_1",
1111            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1112            Some(CAPSULE),
1113        )
1114        .unwrap();
1115        record.append(&[message(&id_at(1), "user", "m")]);
1116        let path = record.path();
1117
1118        for cursor in ["7", "mc_", "mc_nine", "mc_9", "cur_1"] {
1119            assert_eq!(
1120                page(
1121                    &mut RecordCache::default(),
1122                    Some(&path),
1123                    &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1124                    Some(cursor),
1125                    10
1126                )
1127                .err(),
1128                Some("invalid-cursor".to_string()),
1129                "cursor '{cursor}'"
1130            );
1131        }
1132    }
1133
1134    /// A record that was never written, and a session with no record at all, are both an empty
1135    /// page — never an error.
1136    #[test]
1137    fn an_absent_record_is_an_empty_page() {
1138        let home = tempfile::tempdir().unwrap();
1139        let workdir = tempfile::tempdir().unwrap();
1140        let missing = home
1141            .path()
1142            .join("conversations/capsule/ctx_1/conversation.jsonl");
1143
1144        for path in [None, Some(missing.as_path())] {
1145            let page = page(
1146                &mut RecordCache::default(),
1147                path,
1148                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1149                None,
1150                10,
1151            )
1152            .unwrap();
1153            assert!(page.messages.is_empty());
1154            assert_eq!(page.next_cursor, None);
1155            assert_eq!(page.total, 0);
1156        }
1157    }
1158
1159    /// A read for a session that keeps no record leaves the cache as it found it, so a hook
1160    /// dispatched between two tasks does not cost the next one its parse.
1161    #[test]
1162    fn a_read_with_no_record_leaves_the_cache_alone() {
1163        let home = tempfile::tempdir().unwrap();
1164        let workdir = tempfile::tempdir().unwrap();
1165        let root = home.path().join("conversations/capsule");
1166        let mut record = ConversationRecord::open(
1167            &root,
1168            "ctx_1",
1169            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1170            Some(CAPSULE),
1171        )
1172        .unwrap();
1173        record.append(&[message(&id_at(1), "user", "a")]);
1174        let path = record.path();
1175
1176        let mut cache = RecordCache::default();
1177        assert_eq!(
1178            page(
1179                &mut cache,
1180                Some(&path),
1181                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1182                None,
1183                10
1184            )
1185            .unwrap()
1186            .total,
1187            1
1188        );
1189        assert!(page(
1190            &mut cache,
1191            None,
1192            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1193            None,
1194            10
1195        )
1196        .unwrap()
1197        .messages
1198        .is_empty());
1199        assert_eq!(
1200            page(
1201                &mut cache,
1202                Some(&path),
1203                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1204                None,
1205                10
1206            )
1207            .unwrap()
1208            .total,
1209            1
1210        );
1211        assert_eq!(cache.reads(), 1, "the parse survived the empty read");
1212    }
1213
1214    /// A malformed line is skipped, reported once naming the path and its 1-based number, and
1215    /// costs the read nothing else.
1216    #[test]
1217    fn a_malformed_line_is_skipped_and_reported() {
1218        let home = tempfile::tempdir().unwrap();
1219        let workdir = tempfile::tempdir().unwrap();
1220        let dir = home.path().join("conversations/capsule/ctx_1");
1221        std::fs::create_dir_all(&dir).unwrap();
1222        let path = dir.join(RECORD_FILE_NAME);
1223        let mut raw = String::new();
1224        for n in 1..=3 {
1225            raw.push_str(&serde_json::to_string(&message(&id_at(n), "user", "m")).unwrap());
1226            raw.push('\n');
1227        }
1228        raw.push_str("{\"role\":\"user\"");
1229        std::fs::write(&path, raw).unwrap();
1230
1231        let page = page(
1232            &mut RecordCache::default(),
1233            Some(&path),
1234            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1235            None,
1236            10,
1237        )
1238        .unwrap();
1239        assert_eq!(page.total, 3);
1240        assert_eq!(page.messages.len(), 3);
1241
1242        let log =
1243            std::fs::read_to_string(workdir.path().join("logs/bootstrap.log")).unwrap_or_default();
1244        assert_eq!(
1245            log.lines()
1246                .filter(|line| line.contains("line 4 is not a message"))
1247                .count(),
1248            1,
1249            "reported once, naming the line: {log}"
1250        );
1251    }
1252
1253    /// A write that fails and then can succeed leaves the record whole: the message the failure
1254    /// cost is written by the next append that carries it, in its own order.
1255    ///
1256    /// The failure is a regular file sitting where the context directory has to go, which no
1257    /// permission bit is involved in — the suite reads the same as root and as anyone else.
1258    #[test]
1259    fn a_failed_write_is_retried_and_leaves_no_hole() {
1260        let home = tempfile::tempdir().unwrap();
1261        let workdir = tempfile::tempdir().unwrap();
1262        let root = home.path().join("conversations/capsule");
1263        let blocker = root.join("ctx_1");
1264        std::fs::create_dir_all(&root).unwrap();
1265        std::fs::write(&blocker, "").unwrap();
1266
1267        let mut record = ConversationRecord::open(
1268            &root,
1269            "ctx_1",
1270            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1271            Some(CAPSULE),
1272        )
1273        .unwrap();
1274        let first = message(&id_at(1), "user", "a");
1275        let second = message(&id_at(2), "assistant", "b");
1276        record.append(std::slice::from_ref(&first));
1277        assert!(
1278            blocker.is_file(),
1279            "the directory chain could not be created"
1280        );
1281
1282        std::fs::remove_file(&blocker).unwrap();
1283        record.append(&[first, second]);
1284
1285        let written: Vec<Value> = std::fs::read_to_string(record.path())
1286            .unwrap()
1287            .lines()
1288            .filter_map(parse_message_line)
1289            .collect();
1290        assert_eq!(
1291            written
1292                .iter()
1293                .map(|m| crate::agent::message_id(m).unwrap())
1294                .collect::<Vec<_>>(),
1295            vec![id_at(1), id_at(2)],
1296            "the message the failure cost is on disk, before the one that followed it"
1297        );
1298    }
1299
1300    /// A whole paging walk costs one read of the record and one parse of it.
1301    #[test]
1302    fn a_paging_loop_reads_the_record_once() {
1303        let home = tempfile::tempdir().unwrap();
1304        let workdir = tempfile::tempdir().unwrap();
1305        let root = home.path().join("conversations/capsule");
1306        let mut record = ConversationRecord::open(
1307            &root,
1308            "ctx_1",
1309            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1310            Some(CAPSULE),
1311        )
1312        .unwrap();
1313        record.append(
1314            &(1..=250)
1315                .map(|n| message(&id_at(n), "user", "m"))
1316                .collect::<Vec<_>>(),
1317        );
1318        let path = record.path();
1319        let mut cache = RecordCache::default();
1320
1321        let mut pages = 0;
1322        let mut seen = Vec::new();
1323        let mut cursor: Option<String> = None;
1324        loop {
1325            let page = page(
1326                &mut cache,
1327                Some(&path),
1328                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1329                cursor.as_deref(),
1330                25,
1331            )
1332            .unwrap();
1333            pages += 1;
1334            seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1335            match page.next_cursor {
1336                Some(next) => cursor = Some(next),
1337                None => break,
1338            }
1339        }
1340        assert_eq!(pages, 10);
1341        assert_eq!(seen, (1..=250).rev().map(id_at).collect::<Vec<_>>());
1342        assert_eq!(cache.reads(), 1, "one read for the whole walk");
1343
1344        // The record grows underneath the same reader: the next walk sees the new messages and
1345        // pays for exactly one more read.
1346        record.append(
1347            &(251..=260)
1348                .map(|n| message(&id_at(n), "user", "m"))
1349                .collect::<Vec<_>>(),
1350        );
1351        let mut seen = Vec::new();
1352        let mut cursor: Option<String> = None;
1353        let mut first_total = None;
1354        loop {
1355            let page = page(
1356                &mut cache,
1357                Some(&path),
1358                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1359                cursor.as_deref(),
1360                25,
1361            )
1362            .unwrap();
1363            first_total.get_or_insert(page.total);
1364            seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1365            match page.next_cursor {
1366                Some(next) => cursor = Some(next),
1367                None => break,
1368            }
1369        }
1370        assert_eq!(first_total, Some(260));
1371        assert_eq!(seen, (1..=260).rev().map(id_at).collect::<Vec<_>>());
1372        assert_eq!(cache.reads(), 2, "one further read for the second walk");
1373    }
1374
1375    /// One truncated line is one log entry, however many pages are walked over it.
1376    #[test]
1377    fn a_malformed_line_is_reported_once_for_a_whole_walk() {
1378        let home = tempfile::tempdir().unwrap();
1379        let workdir = tempfile::tempdir().unwrap();
1380        let dir = home.path().join("conversations/capsule/ctx_1");
1381        std::fs::create_dir_all(&dir).unwrap();
1382        let path = dir.join(RECORD_FILE_NAME);
1383        let mut raw = String::new();
1384        for n in 1..=10 {
1385            raw.push_str(&serde_json::to_string(&message(&id_at(n), "user", "m")).unwrap());
1386            raw.push('\n');
1387        }
1388        raw.push_str("{\"role\":\"user\"");
1389        std::fs::write(&path, raw).unwrap();
1390
1391        let mut cache = RecordCache::default();
1392        let mut cursor: Option<String> = None;
1393        let mut pages = 0;
1394        loop {
1395            let page = page(
1396                &mut cache,
1397                Some(&path),
1398                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1399                cursor.as_deref(),
1400                2,
1401            )
1402            .unwrap();
1403            pages += 1;
1404            assert_eq!(page.total, 10);
1405            match page.next_cursor {
1406                Some(next) => cursor = Some(next),
1407                None => break,
1408            }
1409        }
1410        assert_eq!(pages, 5);
1411
1412        let log =
1413            std::fs::read_to_string(workdir.path().join("logs/bootstrap.log")).unwrap_or_default();
1414        assert_eq!(
1415            log.lines()
1416                .filter(|line| line.contains("line 11 is not a message"))
1417                .count(),
1418            1,
1419            "one entry for the whole walk: {log}"
1420        );
1421    }
1422
1423    /// `context.record: off` resolves to no record name at all, whatever `record_store` says
1424    /// beside it.
1425    #[test]
1426    fn record_off_wins_over_a_declared_store() {
1427        let off = murmur_artifact::ContextConfig {
1428            max_tokens: None,
1429            record: false,
1430            record_store: Some("shey".to_string()),
1431            seed_budget: murmur_artifact::DEFAULT_SEED_BUDGET,
1432            seed_overflow_margin: murmur_artifact::DEFAULT_SEED_OVERFLOW_MARGIN,
1433            retain: None,
1434        };
1435        assert_eq!(resolve_record_name(Some(&off), "capsule"), None);
1436
1437        let on = murmur_artifact::ContextConfig {
1438            record: true,
1439            ..off.clone()
1440        };
1441        assert_eq!(
1442            resolve_record_name(Some(&on), "capsule"),
1443            Some("shey".to_string())
1444        );
1445        assert_eq!(
1446            resolve_record_name(None, "capsule"),
1447            Some("capsule".to_string()),
1448            "a capsule declaring no context block still records"
1449        );
1450    }
1451
1452    /// Every refusal names the value the operator wrote and the rule it broke.
1453    #[test]
1454    fn a_malformed_record_name_is_refused_by_name() {
1455        for value in ["", "../escape", "/abs", "a/b", ".hidden"] {
1456            let err = validate_record_segment("context.record_store", value)
1457                .expect_err("'{value}' must be refused");
1458            let message = err.to_string();
1459            assert!(message.contains(&format!("'{value}'")), "{message}");
1460            assert!(message.contains("single path segment"), "{message}");
1461            assert!(message.contains("context.record_store"), "{message}");
1462        }
1463        validate_record_segment("--context", "ctx_fixed").unwrap();
1464    }
1465}