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