1use 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
31pub(crate) const CONVERSATION_ROOT_DIR: &str = "conversations";
33
34pub(crate) const RECORD_FILE_NAME: &str = "conversation.jsonl";
36
37pub(crate) const RECORD_HEADER_TYPE: &str = "murmur.record";
39
40pub(crate) const UNRESOLVED_CONTEXT: &str = "<unresolved>";
43
44const RECORD_DIR_MODE: u32 = 0o700;
47
48pub(crate) const RECORD_FILE_MODE: u32 = 0o600;
50
51const CURSOR_PREFIX: &str = "mc_";
54
55const MAX_PAGE_LIMIT: u32 = 100;
58
59const SEGMENT_RULE: &str = "must be a single path segment: no '/', no '.' or '..', not absolute, \
64 and not starting with a dot";
65
66#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
71pub struct TruncationMarker {
72 pub dropped: u64,
73 pub oldest_surviving_id: String,
75 pub last_dropped_id: String,
79 pub at_ms: u64,
81}
82
83#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
94pub struct RecordHeader {
95 #[serde(rename = "type")]
97 pub kind: String,
98 pub capsule: String,
100 pub created_ms: u64,
103 #[serde(default, skip_serializing_if = "Option::is_none")]
105 pub truncated: Option<TruncationMarker>,
106}
107
108pub(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
116pub(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
123pub(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
129pub(crate) fn message_id_timestamp_ms(id: &str) -> Option<u64> {
134 crate::retention::uuid_v7_timestamp_ms(id, "msg_")
135}
136
137pub(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
151pub(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
170pub(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
179pub(crate) fn record_file(root: &Path, context_id: &str) -> PathBuf {
185 root.join(context_id).join(RECORD_FILE_NAME)
186}
187
188pub(crate) struct ConversationRecord {
194 dir: PathBuf,
196 workdir: WorkdirRoot,
198 capsule: Option<String>,
201 header_ensured: bool,
204 recorded: HashSet<String>,
208 failure_reported: bool,
211 malformed_reported: bool,
213}
214
215impl ConversationRecord {
216 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 pub(crate) fn dir(&self) -> &Path {
256 &self.dir
257 }
258
259 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 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 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 crate::record_seal::note_conversation_append(self.workdir.path(), &self.dir);
342 Ok(())
343 }
344
345 fn ensure_dirs(&self) -> Result<(), String> {
346 ensure_context_dirs(&self.dir)
347 }
348
349 fn ensure_header(&mut self) -> Result<(), String> {
367 if self.header_ensured {
368 return Ok(());
369 }
370 let Some(capsule) = self.capsule.clone() else {
371 self.header_ensured = true;
372 return Ok(());
373 };
374 let path = self.path();
375 let existing = match std::fs::read_to_string(&path) {
376 Ok(raw) => raw,
377 Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
378 Err(err) => return Err(format!("{}: {err}", path.display())),
379 };
380 if existing.lines().next().and_then(parse_header).is_some() {
381 self.header_ensured = true;
382 return Ok(());
383 }
384
385 let header = RecordHeader {
386 kind: RECORD_HEADER_TYPE.to_string(),
387 capsule,
388 created_ms: crate::retention::now_ms(),
389 truncated: None,
390 };
391 let line = header_line(&header)?;
392 if existing.is_empty() {
396 open_record_file(&path, true)?
397 .write_all(line.as_bytes())
398 .map_err(|err| format!("{}: {err}", path.display()))?;
399 } else {
400 let mut contents = line;
401 contents.push_str(&existing);
402 crate::retention::StagedRewrite::stage_with_mode(
403 &path,
404 contents.as_bytes(),
405 Some(RECORD_FILE_MODE),
406 )?
407 .commit()?;
408 let sealed_by = self
411 .workdir
412 .path()
413 .file_name()
414 .and_then(|name| name.to_str())
415 .unwrap_or(crate::record_seal::CLI_SEALER)
416 .to_string();
417 crate::record_seal::seal_conversation(&self.dir, &sealed_by);
418 }
419 self.header_ensured = true;
420 Ok(())
421 }
422}
423
424fn open_record_file(path: &Path, truncate: bool) -> Result<std::fs::File, String> {
427 use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
428
429 let mut options = std::fs::OpenOptions::new();
430 options.create(true).mode(RECORD_FILE_MODE);
431 if truncate {
432 options.write(true).truncate(true);
433 } else {
434 options.append(true);
435 }
436 let file = options
437 .open(path)
438 .map_err(|err| format!("{}: {err}", path.display()))?;
439 file.set_permissions(std::fs::Permissions::from_mode(RECORD_FILE_MODE))
440 .map_err(|err| format!("{}: {err}", path.display()))?;
441 Ok(file)
442}
443
444#[derive(Default)]
451pub(crate) struct RecordCache {
452 path: Option<PathBuf>,
454 len: u64,
456 messages: Vec<Value>,
458 malformed_reported: bool,
460 #[cfg(test)]
462 reads: u32,
463}
464
465impl RecordCache {
466 fn messages(&mut self, path: &Path, workdir: &WorkdirRoot) -> Result<&[Value], String> {
472 let len = match std::fs::metadata(path) {
473 Ok(metadata) => metadata.len(),
474 Err(err) if err.kind() == std::io::ErrorKind::NotFound => 0,
475 Err(err) => return Err(err.to_string()),
476 };
477 if self.path.as_deref() == Some(path) && self.len == len {
478 return Ok(&self.messages);
479 }
480
481 self.path = None;
485 #[cfg(test)]
486 {
487 self.reads += 1;
488 }
489 self.messages = read_record(path, workdir, &mut self.malformed_reported)?;
490 self.len = len;
491 self.path = Some(path.to_path_buf());
492 Ok(&self.messages)
493 }
494
495 #[cfg(test)]
496 pub(crate) fn reads(&self) -> u32 {
497 self.reads
498 }
499}
500
501pub(crate) fn page(
509 cache: &mut RecordCache,
510 path: Option<&Path>,
511 workdir: &WorkdirRoot,
512 cursor: Option<&str>,
513 limit: u32,
514) -> Result<MessagePage, String> {
515 let Some(path) = path else {
516 return Ok(empty_page());
517 };
518 let messages = cache
519 .messages(path, workdir)
520 .map_err(|reason| format!("unavailable: {reason}"))?;
521
522 let end = match cursor {
525 None => messages.len(),
526 Some(cursor) => decode_cursor(cursor)
527 .filter(|end| *end <= messages.len())
528 .ok_or_else(|| "invalid-cursor".to_string())?,
529 };
530 let limit = limit.clamp(1, MAX_PAGE_LIMIT) as usize;
531 let start = end.saturating_sub(limit);
532
533 let mut window: Vec<Value> = messages[start..end].to_vec();
534 window.reverse();
535 Ok(MessagePage {
540 messages: crate::agent::to_wit_messages(&window)
541 .into_iter()
542 .map(|message| Message {
543 role: message.role,
544 content: message.content,
545 id: message.id,
546 source_id: message.source_id,
547 inserted_by: message.inserted_by.map(|mark| {
548 crate::agent::ContextInsertion::from_lowered(mark).to_imported_wit()
549 }),
550 })
551 .collect(),
552 next_cursor: (start > 0).then(|| encode_cursor(start)),
553 total: u32::try_from(messages.len()).unwrap_or(u32::MAX),
554 })
555}
556
557fn empty_page() -> MessagePage {
559 MessagePage {
560 messages: Vec::new(),
561 next_cursor: None,
562 total: 0,
563 }
564}
565
566fn read_record(
578 path: &Path,
579 workdir: &WorkdirRoot,
580 reported: &mut bool,
581) -> Result<Vec<Value>, String> {
582 let raw = match std::fs::read_to_string(path) {
583 Ok(raw) => raw,
584 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
585 Err(err) => return Err(err.to_string()),
586 };
587
588 Ok(record_messages(&raw, |line_number| {
589 if !*reported {
590 *reported = true;
591 report(
592 workdir,
593 &format!(
594 "[conversation] {}: line {line_number} is not a message and was skipped",
595 path.display()
596 ),
597 );
598 }
599 }))
600}
601
602pub(crate) fn record_messages(raw: &str, mut skipped: impl FnMut(usize)) -> Vec<Value> {
606 let mut messages = Vec::new();
607 for (index, line) in raw.lines().enumerate() {
608 if line.trim().is_empty() {
609 continue;
610 }
611 match parse_message_line(line) {
612 Some(message) => messages.push(message),
613 None => skipped(index + 1),
614 }
615 }
616 messages
617}
618
619pub(crate) fn parse_message_line(line: &str) -> Option<Value> {
624 let value: Value = serde_json::from_str(line).ok()?;
625 value
626 .get("role")
627 .and_then(Value::as_str)
628 .is_some()
629 .then_some(value)
630}
631
632fn encode_cursor(position: usize) -> String {
633 format!("{CURSOR_PREFIX}{position}")
634}
635
636fn decode_cursor(cursor: &str) -> Option<usize> {
637 cursor.strip_prefix(CURSOR_PREFIX)?.parse().ok()
638}
639
640pub(crate) fn ensure_context_dirs(dir: &Path) -> Result<(), String> {
650 let mut chain = Vec::new();
651 let mut ancestor = Some(dir);
652 for _ in 0..3 {
653 if let Some(current) = ancestor {
654 chain.push(current);
655 ancestor = current.parent();
656 }
657 }
658 if let (Some(owner), Ok(home)) = (ancestor, crate::state_store::murmur_home_dir()) {
659 if owner == home {
660 crate::murmur_home::ensure_murmur_home()?;
661 }
662 }
663 for path in chain.into_iter().rev() {
664 crate::state_store::ensure_private_dir(path, RECORD_DIR_MODE)
665 .map_err(|reason| format!("{}: {reason}", path.display()))?;
666 }
667 Ok(())
668}
669
670pub(crate) fn report(workdir: &WorkdirRoot, message: &str) {
672 crate::runtime_err!("[capsule-runtime] {message}");
673 crate::agent::append_bootstrap_log(workdir, message);
674}
675
676#[cfg(test)]
677mod tests {
678 use super::*;
679 use serde_json::json;
680
681 const CAPSULE: &str = "capsule";
683
684 fn message(id: &str, role: &str, text: &str) -> Value {
685 json!({"role": role, "content": [{"type": "text", "text": text}], "id": id})
686 }
687
688 fn id_at(id: usize) -> String {
689 format!("msg_{:032x}", id)
690 }
691
692 #[test]
693 fn record_file_is_held_owner_only_on_append() {
694 let home = tempfile::tempdir().unwrap();
695 let workdir = tempfile::tempdir().unwrap();
696 let root = home.path().join("conversations/shared");
697 crate::murmur_home::wide_dir(&root.join("ctx_1"), 0o755);
698 let path = root.join("ctx_1").join(RECORD_FILE_NAME);
699 crate::murmur_home::wide_file(&path, "", 0o644);
700
701 let mut record = ConversationRecord::open(
702 &root,
703 "ctx_1",
704 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
705 None,
706 )
707 .unwrap();
708 record.append(&[message(&id_at(1), "user", "hello")]);
709
710 assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
711 assert_eq!(crate::murmur_home::mode_of(&root.join("ctx_1")), 0o700);
712 assert_eq!(std::fs::read_to_string(&path).unwrap().lines().count(), 1);
713 }
714
715 #[test]
716 fn record_file_is_created_owner_only_with_its_header() {
717 let home = tempfile::tempdir().unwrap();
718 let workdir = tempfile::tempdir().unwrap();
719 let root = home.path().join("conversations/shared");
720
721 let mut record = ConversationRecord::open(
722 &root,
723 "ctx_1",
724 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
725 Some(CAPSULE),
726 )
727 .unwrap();
728 record.append(&[message(&id_at(1), "user", "hello")]);
729
730 assert_eq!(crate::murmur_home::mode_of(&record.path()), 0o600);
731 }
732
733 #[test]
735 fn record_rewrite_keeps_the_file_owner_only() {
736 let home = tempfile::tempdir().unwrap();
737 let workdir = tempfile::tempdir().unwrap();
738 let root = home.path().join("conversations/shared");
739 crate::murmur_home::wide_dir(&root.join("ctx_1"), 0o755);
740 let path = root.join("ctx_1").join(RECORD_FILE_NAME);
741 let old = serde_json::to_string(&message(&id_at(1), "user", "old")).unwrap();
742 crate::murmur_home::wide_file(&path, &format!("{old}\n"), 0o644);
743
744 let mut record = ConversationRecord::open(
745 &root,
746 "ctx_1",
747 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
748 Some(CAPSULE),
749 )
750 .unwrap();
751 record.append(&[message(&id_at(2), "user", "new")]);
752
753 assert!(read_header(&path).is_some(), "the record was adopted");
754 assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
755 }
756
757 #[test]
758 fn record_root_under_a_wide_murmur_home_holds_the_home_owner_only() {
759 let home = tempfile::tempdir().unwrap();
760 crate::murmur_home::wide_dir(&home.path().join(".murmur"), 0o755);
761 crate::murmur_home::run_with_home(
762 "conversation::tests::inner_append_under_scratch_home",
763 home.path(),
764 );
765 assert_eq!(
766 crate::murmur_home::mode_of(&home.path().join(".murmur")),
767 0o700
768 );
769 let record = home
770 .path()
771 .join(".murmur/conversations/shared/ctx_1")
772 .join(RECORD_FILE_NAME);
773 assert_eq!(crate::murmur_home::mode_of(&record), 0o600);
774 }
775
776 #[test]
777 #[ignore = "run by record_root_under_a_wide_murmur_home_holds_the_home_owner_only"]
778 fn inner_append_under_scratch_home() {
779 if !crate::murmur_home::in_scratch_home() {
780 return;
781 }
782 let workdir = tempfile::tempdir().unwrap();
783 let root = record_root("shared").unwrap();
784 let mut record = ConversationRecord::open(
785 &root,
786 "ctx_1",
787 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
788 None,
789 )
790 .unwrap();
791 record.append(&[message(&id_at(1), "user", "hello")]);
792 assert!(record.path().exists());
793 }
794
795 #[test]
798 fn the_first_append_writes_a_header_naming_the_capsule() {
799 let home = tempfile::tempdir().unwrap();
800 let workdir = tempfile::tempdir().unwrap();
801 let root = home.path().join("conversations/shared");
802
803 let mut record = ConversationRecord::open(
804 &root,
805 "ctx_1",
806 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
807 Some(CAPSULE),
808 )
809 .unwrap();
810 record.append(&[message(&id_at(1), "user", "hello")]);
811
812 let raw = std::fs::read_to_string(record.path()).unwrap();
813 let lines: Vec<&str> = raw.lines().collect();
814 assert_eq!(lines.len(), 2, "a header and one message: {lines:?}");
815 let header = parse_header(lines[0]).expect("the first line is a header");
816 assert_eq!(header.kind, RECORD_HEADER_TYPE);
817 assert_eq!(header.capsule, CAPSULE);
818 assert!(header.truncated.is_none(), "nothing has been dropped yet");
819 assert!(header.created_ms > 0);
820 }
821
822 #[test]
825 fn a_header_line_changes_neither_the_loaded_messages_nor_total() {
826 let home = tempfile::tempdir().unwrap();
827 let workdir = tempfile::tempdir().unwrap();
828 let root = home.path().join("conversations/c");
829
830 let mut record = ConversationRecord::open(
831 &root,
832 "ctx_1",
833 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
834 Some(CAPSULE),
835 )
836 .unwrap();
837 record.append(&[
838 message(&id_at(1), "user", "one"),
839 message(&id_at(2), "assistant", "two"),
840 ]);
841 let path = record.path();
842
843 let mut reader = ConversationRecord::open(
844 &root,
845 "ctx_1",
846 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
847 Some(CAPSULE),
848 )
849 .unwrap();
850 let loaded = reader.load();
851 assert_eq!(loaded.len(), 2, "the header is not a message: {loaded:?}");
852
853 let mut cache = RecordCache::default();
854 let page = page(
855 &mut cache,
856 Some(&path),
857 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
858 None,
859 100,
860 )
861 .unwrap();
862 assert_eq!(page.total, 2);
863 assert_eq!(page.messages.len(), 2);
864 }
865
866 #[test]
870 fn a_headerless_record_is_adopted_on_the_next_append() {
871 let home = tempfile::tempdir().unwrap();
872 let workdir = tempfile::tempdir().unwrap();
873 let root = home.path().join("conversations/c");
874 let dir = root.join("ctx_1");
875 std::fs::create_dir_all(&dir).unwrap();
876 let path = dir.join(RECORD_FILE_NAME);
877 let headerless = format!(
878 "{}
879{}
880",
881 serde_json::to_string(&message(&id_at(1), "user", "one")).unwrap(),
882 serde_json::to_string(&message(&id_at(2), "assistant", "two")).unwrap()
883 );
884 std::fs::write(&path, &headerless).unwrap();
885 assert!(read_header(&path).is_none(), "unowned before the append");
886
887 let mut record = ConversationRecord::open(
888 &root,
889 "ctx_1",
890 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
891 Some(CAPSULE),
892 )
893 .unwrap();
894 record.append(&[message(&id_at(3), "user", "three")]);
895
896 let raw = std::fs::read_to_string(&path).unwrap();
897 assert_eq!(read_header(&path).unwrap().capsule, CAPSULE);
898 assert!(
899 raw.contains(headerless.trim_end()),
900 "every line it already held survives byte for byte: {raw}"
901 );
902 assert_eq!(raw.lines().count(), 4, "a header and three messages");
903 }
904
905 #[test]
908 fn an_owned_record_keeps_the_header_it_already_has() {
909 let home = tempfile::tempdir().unwrap();
910 let workdir = tempfile::tempdir().unwrap();
911 let root = home.path().join("conversations/c");
912 let dir = root.join("ctx_1");
913 std::fs::create_dir_all(&dir).unwrap();
914 let path = dir.join(RECORD_FILE_NAME);
915 let header = RecordHeader {
916 kind: RECORD_HEADER_TYPE.to_string(),
917 capsule: "someone-else".to_string(),
918 created_ms: 1_756_400_000_000,
919 truncated: Some(TruncationMarker {
920 dropped: 5,
921 oldest_surviving_id: id_at(6),
922 last_dropped_id: id_at(5),
923 at_ms: 1_756_400_000_001,
924 }),
925 };
926 std::fs::write(&path, header_line(&header).unwrap()).unwrap();
927
928 let mut record = ConversationRecord::open(
929 &root,
930 "ctx_1",
931 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
932 Some(CAPSULE),
933 )
934 .unwrap();
935 record.append(&[message(&id_at(7), "user", "seven")]);
936
937 assert_eq!(read_header(&path).unwrap(), header);
938 }
939
940 #[test]
943 fn a_reloaded_message_keeps_its_id_and_is_not_written_twice() {
944 let home = tempfile::tempdir().unwrap();
945 let workdir = tempfile::tempdir().unwrap();
946 let root = home.path().join("conversations/capsule");
947
948 let mut first = ConversationRecord::open(
949 &root,
950 "ctx_fixed",
951 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
952 Some(CAPSULE),
953 )
954 .unwrap();
955 first.append(&[message(&id_at(1), "user", "hello")]);
956
957 let mut second = ConversationRecord::open(
958 &root,
959 "ctx_fixed",
960 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
961 Some(CAPSULE),
962 )
963 .unwrap();
964 let loaded = second.load();
965 assert_eq!(loaded.len(), 1);
966 assert_eq!(
967 crate::agent::message_id(&loaded[0]),
968 Some(id_at(1).as_str())
969 );
970
971 second.append(&loaded);
972 second.append(&[message(&id_at(2), "assistant", "hi")]);
973 assert_eq!(
974 second.load().len(),
975 2,
976 "a reloaded message is never re-appended"
977 );
978 }
979
980 #[test]
983 fn the_directory_chain_is_created_at_0700_on_first_write() {
984 use std::os::unix::fs::PermissionsExt;
985
986 let home = tempfile::tempdir().unwrap();
987 let workdir = tempfile::tempdir().unwrap();
988 let conversations = home.path().join("conversations");
989 let root = conversations.join("capsule");
990
991 let mut record = ConversationRecord::open(
992 &root,
993 "ctx_1",
994 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
995 Some(CAPSULE),
996 )
997 .unwrap();
998 assert!(!conversations.exists(), "opening a record creates nothing");
999
1000 record.append(&[message(&id_at(1), "user", "hello")]);
1001 for dir in [&conversations, &root, &root.join("ctx_1")] {
1002 let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777;
1003 assert_eq!(
1004 mode,
1005 0o700,
1006 "{} must be 0700, got {mode:04o}",
1007 dir.display()
1008 );
1009 }
1010 }
1011
1012 #[test]
1015 fn a_context_id_that_is_not_a_segment_records_nothing() {
1016 let home = tempfile::tempdir().unwrap();
1017 let workdir = tempfile::tempdir().unwrap();
1018 for context_id in ["../escape", "a/b", "", ".hidden"] {
1019 assert!(
1020 ConversationRecord::open(
1021 &home.path().join("conversations/c"),
1022 context_id,
1023 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1024 Some(CAPSULE)
1025 )
1026 .is_none(),
1027 "'{context_id}' must not open a record"
1028 );
1029 }
1030 assert!(!home.path().join("conversations").exists());
1031 }
1032
1033 #[test]
1035 fn paging_walks_the_record_newest_first_without_repeating_a_message() {
1036 let home = tempfile::tempdir().unwrap();
1037 let workdir = tempfile::tempdir().unwrap();
1038 let root = home.path().join("conversations/capsule");
1039 let mut record = ConversationRecord::open(
1040 &root,
1041 "ctx_1",
1042 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1043 Some(CAPSULE),
1044 )
1045 .unwrap();
1046 let written: Vec<Value> = (1..=5)
1047 .map(|n| message(&id_at(n), "user", &format!("m{n}")))
1048 .collect();
1049 record.append(&written);
1050 let path = record.path();
1051
1052 let mut seen = Vec::new();
1053 let mut cursor: Option<String> = None;
1054 loop {
1055 let page = page(
1056 &mut RecordCache::default(),
1057 Some(&path),
1058 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1059 cursor.as_deref(),
1060 2,
1061 )
1062 .unwrap();
1063 assert_eq!(page.total, 5);
1064 seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1065 match page.next_cursor {
1066 Some(next) => cursor = Some(next),
1067 None => break,
1068 }
1069 }
1070 assert_eq!(
1071 seen,
1072 (1..=5).rev().map(id_at).collect::<Vec<_>>(),
1073 "the whole record, newest first, exactly once each"
1074 );
1075 }
1076
1077 #[test]
1080 fn limit_is_clamped_at_both_ends() {
1081 let home = tempfile::tempdir().unwrap();
1082 let workdir = tempfile::tempdir().unwrap();
1083 let root = home.path().join("conversations/capsule");
1084 let mut record = ConversationRecord::open(
1085 &root,
1086 "ctx_1",
1087 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1088 Some(CAPSULE),
1089 )
1090 .unwrap();
1091 record.append(
1092 &(1..=120)
1093 .map(|n| message(&id_at(n), "user", "m"))
1094 .collect::<Vec<_>>(),
1095 );
1096 let path = record.path();
1097
1098 assert_eq!(
1099 page(
1100 &mut RecordCache::default(),
1101 Some(&path),
1102 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1103 None,
1104 0
1105 )
1106 .unwrap()
1107 .messages
1108 .len(),
1109 1
1110 );
1111 assert_eq!(
1112 page(
1113 &mut RecordCache::default(),
1114 Some(&path),
1115 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1116 None,
1117 u32::MAX
1118 )
1119 .unwrap()
1120 .messages
1121 .len(),
1122 MAX_PAGE_LIMIT as usize
1123 );
1124 }
1125
1126 #[test]
1129 fn a_foreign_cursor_is_refused() {
1130 let home = tempfile::tempdir().unwrap();
1131 let workdir = tempfile::tempdir().unwrap();
1132 let root = home.path().join("conversations/capsule");
1133 let mut record = ConversationRecord::open(
1134 &root,
1135 "ctx_1",
1136 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1137 Some(CAPSULE),
1138 )
1139 .unwrap();
1140 record.append(&[message(&id_at(1), "user", "m")]);
1141 let path = record.path();
1142
1143 for cursor in ["7", "mc_", "mc_nine", "mc_9", "cur_1"] {
1144 assert_eq!(
1145 page(
1146 &mut RecordCache::default(),
1147 Some(&path),
1148 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1149 Some(cursor),
1150 10
1151 )
1152 .err(),
1153 Some("invalid-cursor".to_string()),
1154 "cursor '{cursor}'"
1155 );
1156 }
1157 }
1158
1159 #[test]
1162 fn an_absent_record_is_an_empty_page() {
1163 let home = tempfile::tempdir().unwrap();
1164 let workdir = tempfile::tempdir().unwrap();
1165 let missing = home
1166 .path()
1167 .join("conversations/capsule/ctx_1/conversation.jsonl");
1168
1169 for path in [None, Some(missing.as_path())] {
1170 let page = page(
1171 &mut RecordCache::default(),
1172 path,
1173 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1174 None,
1175 10,
1176 )
1177 .unwrap();
1178 assert!(page.messages.is_empty());
1179 assert_eq!(page.next_cursor, None);
1180 assert_eq!(page.total, 0);
1181 }
1182 }
1183
1184 #[test]
1187 fn a_read_with_no_record_leaves_the_cache_alone() {
1188 let home = tempfile::tempdir().unwrap();
1189 let workdir = tempfile::tempdir().unwrap();
1190 let root = home.path().join("conversations/capsule");
1191 let mut record = ConversationRecord::open(
1192 &root,
1193 "ctx_1",
1194 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1195 Some(CAPSULE),
1196 )
1197 .unwrap();
1198 record.append(&[message(&id_at(1), "user", "a")]);
1199 let path = record.path();
1200
1201 let mut cache = RecordCache::default();
1202 assert_eq!(
1203 page(
1204 &mut cache,
1205 Some(&path),
1206 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1207 None,
1208 10
1209 )
1210 .unwrap()
1211 .total,
1212 1
1213 );
1214 assert!(page(
1215 &mut cache,
1216 None,
1217 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1218 None,
1219 10
1220 )
1221 .unwrap()
1222 .messages
1223 .is_empty());
1224 assert_eq!(
1225 page(
1226 &mut cache,
1227 Some(&path),
1228 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1229 None,
1230 10
1231 )
1232 .unwrap()
1233 .total,
1234 1
1235 );
1236 assert_eq!(cache.reads(), 1, "the parse survived the empty read");
1237 }
1238
1239 #[test]
1242 fn a_malformed_line_is_skipped_and_reported() {
1243 let home = tempfile::tempdir().unwrap();
1244 let workdir = tempfile::tempdir().unwrap();
1245 let dir = home.path().join("conversations/capsule/ctx_1");
1246 std::fs::create_dir_all(&dir).unwrap();
1247 let path = dir.join(RECORD_FILE_NAME);
1248 let mut raw = String::new();
1249 for n in 1..=3 {
1250 raw.push_str(&serde_json::to_string(&message(&id_at(n), "user", "m")).unwrap());
1251 raw.push('\n');
1252 }
1253 raw.push_str("{\"role\":\"user\"");
1254 std::fs::write(&path, raw).unwrap();
1255
1256 let page = page(
1257 &mut RecordCache::default(),
1258 Some(&path),
1259 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1260 None,
1261 10,
1262 )
1263 .unwrap();
1264 assert_eq!(page.total, 3);
1265 assert_eq!(page.messages.len(), 3);
1266
1267 let log =
1268 std::fs::read_to_string(workdir.path().join("logs/bootstrap.log")).unwrap_or_default();
1269 assert_eq!(
1270 log.lines()
1271 .filter(|line| line.contains("line 4 is not a message"))
1272 .count(),
1273 1,
1274 "reported once, naming the line: {log}"
1275 );
1276 }
1277
1278 #[test]
1284 fn a_failed_write_is_retried_and_leaves_no_hole() {
1285 let home = tempfile::tempdir().unwrap();
1286 let workdir = tempfile::tempdir().unwrap();
1287 let root = home.path().join("conversations/capsule");
1288 let blocker = root.join("ctx_1");
1289 std::fs::create_dir_all(&root).unwrap();
1290 std::fs::write(&blocker, "").unwrap();
1291
1292 let mut record = ConversationRecord::open(
1293 &root,
1294 "ctx_1",
1295 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1296 Some(CAPSULE),
1297 )
1298 .unwrap();
1299 let first = message(&id_at(1), "user", "a");
1300 let second = message(&id_at(2), "assistant", "b");
1301 record.append(std::slice::from_ref(&first));
1302 assert!(
1303 blocker.is_file(),
1304 "the directory chain could not be created"
1305 );
1306
1307 std::fs::remove_file(&blocker).unwrap();
1308 record.append(&[first, second]);
1309
1310 let written: Vec<Value> = std::fs::read_to_string(record.path())
1311 .unwrap()
1312 .lines()
1313 .filter_map(parse_message_line)
1314 .collect();
1315 assert_eq!(
1316 written
1317 .iter()
1318 .map(|m| crate::agent::message_id(m).unwrap())
1319 .collect::<Vec<_>>(),
1320 vec![id_at(1), id_at(2)],
1321 "the message the failure cost is on disk, before the one that followed it"
1322 );
1323 }
1324
1325 #[test]
1327 fn a_paging_loop_reads_the_record_once() {
1328 let home = tempfile::tempdir().unwrap();
1329 let workdir = tempfile::tempdir().unwrap();
1330 let root = home.path().join("conversations/capsule");
1331 let mut record = ConversationRecord::open(
1332 &root,
1333 "ctx_1",
1334 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1335 Some(CAPSULE),
1336 )
1337 .unwrap();
1338 record.append(
1339 &(1..=250)
1340 .map(|n| message(&id_at(n), "user", "m"))
1341 .collect::<Vec<_>>(),
1342 );
1343 let path = record.path();
1344 let mut cache = RecordCache::default();
1345
1346 let mut pages = 0;
1347 let mut seen = Vec::new();
1348 let mut cursor: Option<String> = None;
1349 loop {
1350 let page = page(
1351 &mut cache,
1352 Some(&path),
1353 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1354 cursor.as_deref(),
1355 25,
1356 )
1357 .unwrap();
1358 pages += 1;
1359 seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1360 match page.next_cursor {
1361 Some(next) => cursor = Some(next),
1362 None => break,
1363 }
1364 }
1365 assert_eq!(pages, 10);
1366 assert_eq!(seen, (1..=250).rev().map(id_at).collect::<Vec<_>>());
1367 assert_eq!(cache.reads(), 1, "one read for the whole walk");
1368
1369 record.append(
1372 &(251..=260)
1373 .map(|n| message(&id_at(n), "user", "m"))
1374 .collect::<Vec<_>>(),
1375 );
1376 let mut seen = Vec::new();
1377 let mut cursor: Option<String> = None;
1378 let mut first_total = None;
1379 loop {
1380 let page = page(
1381 &mut cache,
1382 Some(&path),
1383 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1384 cursor.as_deref(),
1385 25,
1386 )
1387 .unwrap();
1388 first_total.get_or_insert(page.total);
1389 seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1390 match page.next_cursor {
1391 Some(next) => cursor = Some(next),
1392 None => break,
1393 }
1394 }
1395 assert_eq!(first_total, Some(260));
1396 assert_eq!(seen, (1..=260).rev().map(id_at).collect::<Vec<_>>());
1397 assert_eq!(cache.reads(), 2, "one further read for the second walk");
1398 }
1399
1400 #[test]
1402 fn a_malformed_line_is_reported_once_for_a_whole_walk() {
1403 let home = tempfile::tempdir().unwrap();
1404 let workdir = tempfile::tempdir().unwrap();
1405 let dir = home.path().join("conversations/capsule/ctx_1");
1406 std::fs::create_dir_all(&dir).unwrap();
1407 let path = dir.join(RECORD_FILE_NAME);
1408 let mut raw = String::new();
1409 for n in 1..=10 {
1410 raw.push_str(&serde_json::to_string(&message(&id_at(n), "user", "m")).unwrap());
1411 raw.push('\n');
1412 }
1413 raw.push_str("{\"role\":\"user\"");
1414 std::fs::write(&path, raw).unwrap();
1415
1416 let mut cache = RecordCache::default();
1417 let mut cursor: Option<String> = None;
1418 let mut pages = 0;
1419 loop {
1420 let page = page(
1421 &mut cache,
1422 Some(&path),
1423 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1424 cursor.as_deref(),
1425 2,
1426 )
1427 .unwrap();
1428 pages += 1;
1429 assert_eq!(page.total, 10);
1430 match page.next_cursor {
1431 Some(next) => cursor = Some(next),
1432 None => break,
1433 }
1434 }
1435 assert_eq!(pages, 5);
1436
1437 let log =
1438 std::fs::read_to_string(workdir.path().join("logs/bootstrap.log")).unwrap_or_default();
1439 assert_eq!(
1440 log.lines()
1441 .filter(|line| line.contains("line 11 is not a message"))
1442 .count(),
1443 1,
1444 "one entry for the whole walk: {log}"
1445 );
1446 }
1447
1448 #[test]
1451 fn record_off_wins_over_a_declared_store() {
1452 let off = murmur_artifact::ContextConfig {
1453 max_tokens: None,
1454 record: false,
1455 record_store: Some("shey".to_string()),
1456 seed_budget: murmur_artifact::DEFAULT_SEED_BUDGET,
1457 seed_overflow_margin: murmur_artifact::DEFAULT_SEED_OVERFLOW_MARGIN,
1458 retain: None,
1459 images: murmur_artifact::ContextImagesConfig::default(),
1460 };
1461 assert_eq!(resolve_record_name(Some(&off), "capsule"), None);
1462
1463 let on = murmur_artifact::ContextConfig {
1464 record: true,
1465 ..off.clone()
1466 };
1467 assert_eq!(
1468 resolve_record_name(Some(&on), "capsule"),
1469 Some("shey".to_string())
1470 );
1471 assert_eq!(
1472 resolve_record_name(None, "capsule"),
1473 Some("capsule".to_string()),
1474 "a capsule declaring no context block still records"
1475 );
1476 }
1477
1478 #[test]
1480 fn a_malformed_record_name_is_refused_by_name() {
1481 for value in ["", "../escape", "/abs", "a/b", ".hidden"] {
1482 let err = validate_record_segment("context.record_store", value)
1483 .expect_err("'{value}' must be refused");
1484 let message = err.to_string();
1485 assert!(message.contains(&format!("'{value}'")), "{message}");
1486 assert!(message.contains("single path segment"), "{message}");
1487 assert!(message.contains("context.record_store"), "{message}");
1488 }
1489 validate_record_segment("--context", "ctx_fixed").unwrap();
1490 }
1491}