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