Skip to main content

capsule_runtime/
conversation.rs

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