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 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 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 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 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 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
405fn 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#[derive(Default)]
432pub(crate) struct RecordCache {
433 path: Option<PathBuf>,
435 len: u64,
437 messages: Vec<Value>,
439 malformed_reported: bool,
441 #[cfg(test)]
443 reads: u32,
444}
445
446impl RecordCache {
447 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 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
482pub(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 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 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
538fn empty_page() -> MessagePage {
540 MessagePage {
541 messages: Vec::new(),
542 next_cursor: None,
543 total: 0,
544 }
545}
546
547fn read_record(
559 path: &Path,
560 workdir: &WorkdirRoot,
561 reported: &mut bool,
562) -> Result<Vec<Value>, String> {
563 let raw = match std::fs::read_to_string(path) {
564 Ok(raw) => raw,
565 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
566 Err(err) => return Err(err.to_string()),
567 };
568
569 let mut messages = Vec::new();
570 for (index, line) in raw.lines().enumerate() {
571 if line.trim().is_empty() {
572 continue;
573 }
574 match parse_message_line(line) {
575 Some(message) => messages.push(message),
576 None => {
577 if !*reported {
578 *reported = true;
579 report(
580 workdir,
581 &format!(
582 "[conversation] {}: line {} is not a message and was skipped",
583 path.display(),
584 index + 1
585 ),
586 );
587 }
588 }
589 }
590 }
591 Ok(messages)
592}
593
594pub(crate) fn parse_message_line(line: &str) -> Option<Value> {
599 let value: Value = serde_json::from_str(line).ok()?;
600 value
601 .get("role")
602 .and_then(Value::as_str)
603 .is_some()
604 .then_some(value)
605}
606
607fn encode_cursor(position: usize) -> String {
608 format!("{CURSOR_PREFIX}{position}")
609}
610
611fn decode_cursor(cursor: &str) -> Option<usize> {
612 cursor.strip_prefix(CURSOR_PREFIX)?.parse().ok()
613}
614
615pub(crate) fn ensure_context_dirs(dir: &Path) -> Result<(), String> {
625 let mut chain = Vec::new();
626 let mut ancestor = Some(dir);
627 for _ in 0..3 {
628 if let Some(current) = ancestor {
629 chain.push(current);
630 ancestor = current.parent();
631 }
632 }
633 if let (Some(owner), Ok(home)) = (ancestor, crate::state_store::murmur_home_dir()) {
634 if owner == home {
635 crate::murmur_home::ensure_murmur_home()?;
636 }
637 }
638 for path in chain.into_iter().rev() {
639 crate::state_store::ensure_private_dir(path, RECORD_DIR_MODE)
640 .map_err(|reason| format!("{}: {reason}", path.display()))?;
641 }
642 Ok(())
643}
644
645pub(crate) fn report(workdir: &WorkdirRoot, message: &str) {
647 crate::runtime_err!("[capsule-runtime] {message}");
648 crate::agent::append_bootstrap_log(workdir, message);
649}
650
651#[cfg(test)]
652mod tests {
653 use super::*;
654 use serde_json::json;
655
656 const CAPSULE: &str = "capsule";
658
659 fn message(id: &str, role: &str, text: &str) -> Value {
660 json!({"role": role, "content": [{"type": "text", "text": text}], "id": id})
661 }
662
663 fn id_at(id: usize) -> String {
664 format!("msg_{:032x}", id)
665 }
666
667 #[test]
668 fn record_file_is_held_owner_only_on_append() {
669 let home = tempfile::tempdir().unwrap();
670 let workdir = tempfile::tempdir().unwrap();
671 let root = home.path().join("conversations/shared");
672 crate::murmur_home::wide_dir(&root.join("ctx_1"), 0o755);
673 let path = root.join("ctx_1").join(RECORD_FILE_NAME);
674 crate::murmur_home::wide_file(&path, "", 0o644);
675
676 let mut record = ConversationRecord::open(
677 &root,
678 "ctx_1",
679 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
680 None,
681 )
682 .unwrap();
683 record.append(&[message(&id_at(1), "user", "hello")]);
684
685 assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
686 assert_eq!(crate::murmur_home::mode_of(&root.join("ctx_1")), 0o700);
687 assert_eq!(std::fs::read_to_string(&path).unwrap().lines().count(), 1);
688 }
689
690 #[test]
691 fn record_file_is_created_owner_only_with_its_header() {
692 let home = tempfile::tempdir().unwrap();
693 let workdir = tempfile::tempdir().unwrap();
694 let root = home.path().join("conversations/shared");
695
696 let mut record = ConversationRecord::open(
697 &root,
698 "ctx_1",
699 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
700 Some(CAPSULE),
701 )
702 .unwrap();
703 record.append(&[message(&id_at(1), "user", "hello")]);
704
705 assert_eq!(crate::murmur_home::mode_of(&record.path()), 0o600);
706 }
707
708 #[test]
710 fn record_rewrite_keeps_the_file_owner_only() {
711 let home = tempfile::tempdir().unwrap();
712 let workdir = tempfile::tempdir().unwrap();
713 let root = home.path().join("conversations/shared");
714 crate::murmur_home::wide_dir(&root.join("ctx_1"), 0o755);
715 let path = root.join("ctx_1").join(RECORD_FILE_NAME);
716 let old = serde_json::to_string(&message(&id_at(1), "user", "old")).unwrap();
717 crate::murmur_home::wide_file(&path, &format!("{old}\n"), 0o644);
718
719 let mut record = ConversationRecord::open(
720 &root,
721 "ctx_1",
722 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
723 Some(CAPSULE),
724 )
725 .unwrap();
726 record.append(&[message(&id_at(2), "user", "new")]);
727
728 assert!(read_header(&path).is_some(), "the record was adopted");
729 assert_eq!(crate::murmur_home::mode_of(&path), 0o600);
730 }
731
732 #[test]
733 fn record_root_under_a_wide_murmur_home_holds_the_home_owner_only() {
734 let home = tempfile::tempdir().unwrap();
735 crate::murmur_home::wide_dir(&home.path().join(".murmur"), 0o755);
736 crate::murmur_home::run_with_home(
737 "conversation::tests::inner_append_under_scratch_home",
738 home.path(),
739 );
740 assert_eq!(
741 crate::murmur_home::mode_of(&home.path().join(".murmur")),
742 0o700
743 );
744 let record = home
745 .path()
746 .join(".murmur/conversations/shared/ctx_1")
747 .join(RECORD_FILE_NAME);
748 assert_eq!(crate::murmur_home::mode_of(&record), 0o600);
749 }
750
751 #[test]
752 #[ignore = "run by record_root_under_a_wide_murmur_home_holds_the_home_owner_only"]
753 fn inner_append_under_scratch_home() {
754 if !crate::murmur_home::in_scratch_home() {
755 return;
756 }
757 let workdir = tempfile::tempdir().unwrap();
758 let root = record_root("shared").unwrap();
759 let mut record = ConversationRecord::open(
760 &root,
761 "ctx_1",
762 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
763 None,
764 )
765 .unwrap();
766 record.append(&[message(&id_at(1), "user", "hello")]);
767 assert!(record.path().exists());
768 }
769
770 #[test]
773 fn the_first_append_writes_a_header_naming_the_capsule() {
774 let home = tempfile::tempdir().unwrap();
775 let workdir = tempfile::tempdir().unwrap();
776 let root = home.path().join("conversations/shared");
777
778 let mut record = ConversationRecord::open(
779 &root,
780 "ctx_1",
781 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
782 Some(CAPSULE),
783 )
784 .unwrap();
785 record.append(&[message(&id_at(1), "user", "hello")]);
786
787 let raw = std::fs::read_to_string(record.path()).unwrap();
788 let lines: Vec<&str> = raw.lines().collect();
789 assert_eq!(lines.len(), 2, "a header and one message: {lines:?}");
790 let header = parse_header(lines[0]).expect("the first line is a header");
791 assert_eq!(header.kind, RECORD_HEADER_TYPE);
792 assert_eq!(header.capsule, CAPSULE);
793 assert!(header.truncated.is_none(), "nothing has been dropped yet");
794 assert!(header.created_ms > 0);
795 }
796
797 #[test]
800 fn a_header_line_changes_neither_the_loaded_messages_nor_total() {
801 let home = tempfile::tempdir().unwrap();
802 let workdir = tempfile::tempdir().unwrap();
803 let root = home.path().join("conversations/c");
804
805 let mut record = ConversationRecord::open(
806 &root,
807 "ctx_1",
808 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
809 Some(CAPSULE),
810 )
811 .unwrap();
812 record.append(&[
813 message(&id_at(1), "user", "one"),
814 message(&id_at(2), "assistant", "two"),
815 ]);
816 let path = record.path();
817
818 let mut reader = ConversationRecord::open(
819 &root,
820 "ctx_1",
821 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
822 Some(CAPSULE),
823 )
824 .unwrap();
825 let loaded = reader.load();
826 assert_eq!(loaded.len(), 2, "the header is not a message: {loaded:?}");
827
828 let mut cache = RecordCache::default();
829 let page = page(
830 &mut cache,
831 Some(&path),
832 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
833 None,
834 100,
835 )
836 .unwrap();
837 assert_eq!(page.total, 2);
838 assert_eq!(page.messages.len(), 2);
839 }
840
841 #[test]
845 fn a_headerless_record_is_adopted_on_the_next_append() {
846 let home = tempfile::tempdir().unwrap();
847 let workdir = tempfile::tempdir().unwrap();
848 let root = home.path().join("conversations/c");
849 let dir = root.join("ctx_1");
850 std::fs::create_dir_all(&dir).unwrap();
851 let path = dir.join(RECORD_FILE_NAME);
852 let headerless = format!(
853 "{}
854{}
855",
856 serde_json::to_string(&message(&id_at(1), "user", "one")).unwrap(),
857 serde_json::to_string(&message(&id_at(2), "assistant", "two")).unwrap()
858 );
859 std::fs::write(&path, &headerless).unwrap();
860 assert!(read_header(&path).is_none(), "unowned before the append");
861
862 let mut record = ConversationRecord::open(
863 &root,
864 "ctx_1",
865 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
866 Some(CAPSULE),
867 )
868 .unwrap();
869 record.append(&[message(&id_at(3), "user", "three")]);
870
871 let raw = std::fs::read_to_string(&path).unwrap();
872 assert_eq!(read_header(&path).unwrap().capsule, CAPSULE);
873 assert!(
874 raw.contains(headerless.trim_end()),
875 "every line it already held survives byte for byte: {raw}"
876 );
877 assert_eq!(raw.lines().count(), 4, "a header and three messages");
878 }
879
880 #[test]
883 fn an_owned_record_keeps_the_header_it_already_has() {
884 let home = tempfile::tempdir().unwrap();
885 let workdir = tempfile::tempdir().unwrap();
886 let root = home.path().join("conversations/c");
887 let dir = root.join("ctx_1");
888 std::fs::create_dir_all(&dir).unwrap();
889 let path = dir.join(RECORD_FILE_NAME);
890 let header = RecordHeader {
891 kind: RECORD_HEADER_TYPE.to_string(),
892 capsule: "someone-else".to_string(),
893 created_ms: 1_756_400_000_000,
894 truncated: Some(TruncationMarker {
895 dropped: 5,
896 oldest_surviving_id: id_at(6),
897 last_dropped_id: id_at(5),
898 at_ms: 1_756_400_000_001,
899 }),
900 };
901 std::fs::write(&path, header_line(&header).unwrap()).unwrap();
902
903 let mut record = ConversationRecord::open(
904 &root,
905 "ctx_1",
906 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
907 Some(CAPSULE),
908 )
909 .unwrap();
910 record.append(&[message(&id_at(7), "user", "seven")]);
911
912 assert_eq!(read_header(&path).unwrap(), header);
913 }
914
915 #[test]
918 fn a_reloaded_message_keeps_its_id_and_is_not_written_twice() {
919 let home = tempfile::tempdir().unwrap();
920 let workdir = tempfile::tempdir().unwrap();
921 let root = home.path().join("conversations/capsule");
922
923 let mut first = ConversationRecord::open(
924 &root,
925 "ctx_fixed",
926 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
927 Some(CAPSULE),
928 )
929 .unwrap();
930 first.append(&[message(&id_at(1), "user", "hello")]);
931
932 let mut second = ConversationRecord::open(
933 &root,
934 "ctx_fixed",
935 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
936 Some(CAPSULE),
937 )
938 .unwrap();
939 let loaded = second.load();
940 assert_eq!(loaded.len(), 1);
941 assert_eq!(
942 crate::agent::message_id(&loaded[0]),
943 Some(id_at(1).as_str())
944 );
945
946 second.append(&loaded);
947 second.append(&[message(&id_at(2), "assistant", "hi")]);
948 assert_eq!(
949 second.load().len(),
950 2,
951 "a reloaded message is never re-appended"
952 );
953 }
954
955 #[test]
958 fn the_directory_chain_is_created_at_0700_on_first_write() {
959 use std::os::unix::fs::PermissionsExt;
960
961 let home = tempfile::tempdir().unwrap();
962 let workdir = tempfile::tempdir().unwrap();
963 let conversations = home.path().join("conversations");
964 let root = conversations.join("capsule");
965
966 let mut record = ConversationRecord::open(
967 &root,
968 "ctx_1",
969 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
970 Some(CAPSULE),
971 )
972 .unwrap();
973 assert!(!conversations.exists(), "opening a record creates nothing");
974
975 record.append(&[message(&id_at(1), "user", "hello")]);
976 for dir in [&conversations, &root, &root.join("ctx_1")] {
977 let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777;
978 assert_eq!(
979 mode,
980 0o700,
981 "{} must be 0700, got {mode:04o}",
982 dir.display()
983 );
984 }
985 }
986
987 #[test]
990 fn a_context_id_that_is_not_a_segment_records_nothing() {
991 let home = tempfile::tempdir().unwrap();
992 let workdir = tempfile::tempdir().unwrap();
993 for context_id in ["../escape", "a/b", "", ".hidden"] {
994 assert!(
995 ConversationRecord::open(
996 &home.path().join("conversations/c"),
997 context_id,
998 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
999 Some(CAPSULE)
1000 )
1001 .is_none(),
1002 "'{context_id}' must not open a record"
1003 );
1004 }
1005 assert!(!home.path().join("conversations").exists());
1006 }
1007
1008 #[test]
1010 fn paging_walks_the_record_newest_first_without_repeating_a_message() {
1011 let home = tempfile::tempdir().unwrap();
1012 let workdir = tempfile::tempdir().unwrap();
1013 let root = home.path().join("conversations/capsule");
1014 let mut record = ConversationRecord::open(
1015 &root,
1016 "ctx_1",
1017 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1018 Some(CAPSULE),
1019 )
1020 .unwrap();
1021 let written: Vec<Value> = (1..=5)
1022 .map(|n| message(&id_at(n), "user", &format!("m{n}")))
1023 .collect();
1024 record.append(&written);
1025 let path = record.path();
1026
1027 let mut seen = Vec::new();
1028 let mut cursor: Option<String> = None;
1029 loop {
1030 let page = page(
1031 &mut RecordCache::default(),
1032 Some(&path),
1033 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1034 cursor.as_deref(),
1035 2,
1036 )
1037 .unwrap();
1038 assert_eq!(page.total, 5);
1039 seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1040 match page.next_cursor {
1041 Some(next) => cursor = Some(next),
1042 None => break,
1043 }
1044 }
1045 assert_eq!(
1046 seen,
1047 (1..=5).rev().map(id_at).collect::<Vec<_>>(),
1048 "the whole record, newest first, exactly once each"
1049 );
1050 }
1051
1052 #[test]
1055 fn limit_is_clamped_at_both_ends() {
1056 let home = tempfile::tempdir().unwrap();
1057 let workdir = tempfile::tempdir().unwrap();
1058 let root = home.path().join("conversations/capsule");
1059 let mut record = ConversationRecord::open(
1060 &root,
1061 "ctx_1",
1062 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1063 Some(CAPSULE),
1064 )
1065 .unwrap();
1066 record.append(
1067 &(1..=120)
1068 .map(|n| message(&id_at(n), "user", "m"))
1069 .collect::<Vec<_>>(),
1070 );
1071 let path = record.path();
1072
1073 assert_eq!(
1074 page(
1075 &mut RecordCache::default(),
1076 Some(&path),
1077 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1078 None,
1079 0
1080 )
1081 .unwrap()
1082 .messages
1083 .len(),
1084 1
1085 );
1086 assert_eq!(
1087 page(
1088 &mut RecordCache::default(),
1089 Some(&path),
1090 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1091 None,
1092 u32::MAX
1093 )
1094 .unwrap()
1095 .messages
1096 .len(),
1097 MAX_PAGE_LIMIT as usize
1098 );
1099 }
1100
1101 #[test]
1104 fn a_foreign_cursor_is_refused() {
1105 let home = tempfile::tempdir().unwrap();
1106 let workdir = tempfile::tempdir().unwrap();
1107 let root = home.path().join("conversations/capsule");
1108 let mut record = ConversationRecord::open(
1109 &root,
1110 "ctx_1",
1111 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1112 Some(CAPSULE),
1113 )
1114 .unwrap();
1115 record.append(&[message(&id_at(1), "user", "m")]);
1116 let path = record.path();
1117
1118 for cursor in ["7", "mc_", "mc_nine", "mc_9", "cur_1"] {
1119 assert_eq!(
1120 page(
1121 &mut RecordCache::default(),
1122 Some(&path),
1123 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1124 Some(cursor),
1125 10
1126 )
1127 .err(),
1128 Some("invalid-cursor".to_string()),
1129 "cursor '{cursor}'"
1130 );
1131 }
1132 }
1133
1134 #[test]
1137 fn an_absent_record_is_an_empty_page() {
1138 let home = tempfile::tempdir().unwrap();
1139 let workdir = tempfile::tempdir().unwrap();
1140 let missing = home
1141 .path()
1142 .join("conversations/capsule/ctx_1/conversation.jsonl");
1143
1144 for path in [None, Some(missing.as_path())] {
1145 let page = page(
1146 &mut RecordCache::default(),
1147 path,
1148 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1149 None,
1150 10,
1151 )
1152 .unwrap();
1153 assert!(page.messages.is_empty());
1154 assert_eq!(page.next_cursor, None);
1155 assert_eq!(page.total, 0);
1156 }
1157 }
1158
1159 #[test]
1162 fn a_read_with_no_record_leaves_the_cache_alone() {
1163 let home = tempfile::tempdir().unwrap();
1164 let workdir = tempfile::tempdir().unwrap();
1165 let root = home.path().join("conversations/capsule");
1166 let mut record = ConversationRecord::open(
1167 &root,
1168 "ctx_1",
1169 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1170 Some(CAPSULE),
1171 )
1172 .unwrap();
1173 record.append(&[message(&id_at(1), "user", "a")]);
1174 let path = record.path();
1175
1176 let mut cache = RecordCache::default();
1177 assert_eq!(
1178 page(
1179 &mut cache,
1180 Some(&path),
1181 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1182 None,
1183 10
1184 )
1185 .unwrap()
1186 .total,
1187 1
1188 );
1189 assert!(page(
1190 &mut cache,
1191 None,
1192 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1193 None,
1194 10
1195 )
1196 .unwrap()
1197 .messages
1198 .is_empty());
1199 assert_eq!(
1200 page(
1201 &mut cache,
1202 Some(&path),
1203 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1204 None,
1205 10
1206 )
1207 .unwrap()
1208 .total,
1209 1
1210 );
1211 assert_eq!(cache.reads(), 1, "the parse survived the empty read");
1212 }
1213
1214 #[test]
1217 fn a_malformed_line_is_skipped_and_reported() {
1218 let home = tempfile::tempdir().unwrap();
1219 let workdir = tempfile::tempdir().unwrap();
1220 let dir = home.path().join("conversations/capsule/ctx_1");
1221 std::fs::create_dir_all(&dir).unwrap();
1222 let path = dir.join(RECORD_FILE_NAME);
1223 let mut raw = String::new();
1224 for n in 1..=3 {
1225 raw.push_str(&serde_json::to_string(&message(&id_at(n), "user", "m")).unwrap());
1226 raw.push('\n');
1227 }
1228 raw.push_str("{\"role\":\"user\"");
1229 std::fs::write(&path, raw).unwrap();
1230
1231 let page = page(
1232 &mut RecordCache::default(),
1233 Some(&path),
1234 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1235 None,
1236 10,
1237 )
1238 .unwrap();
1239 assert_eq!(page.total, 3);
1240 assert_eq!(page.messages.len(), 3);
1241
1242 let log =
1243 std::fs::read_to_string(workdir.path().join("logs/bootstrap.log")).unwrap_or_default();
1244 assert_eq!(
1245 log.lines()
1246 .filter(|line| line.contains("line 4 is not a message"))
1247 .count(),
1248 1,
1249 "reported once, naming the line: {log}"
1250 );
1251 }
1252
1253 #[test]
1259 fn a_failed_write_is_retried_and_leaves_no_hole() {
1260 let home = tempfile::tempdir().unwrap();
1261 let workdir = tempfile::tempdir().unwrap();
1262 let root = home.path().join("conversations/capsule");
1263 let blocker = root.join("ctx_1");
1264 std::fs::create_dir_all(&root).unwrap();
1265 std::fs::write(&blocker, "").unwrap();
1266
1267 let mut record = ConversationRecord::open(
1268 &root,
1269 "ctx_1",
1270 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1271 Some(CAPSULE),
1272 )
1273 .unwrap();
1274 let first = message(&id_at(1), "user", "a");
1275 let second = message(&id_at(2), "assistant", "b");
1276 record.append(std::slice::from_ref(&first));
1277 assert!(
1278 blocker.is_file(),
1279 "the directory chain could not be created"
1280 );
1281
1282 std::fs::remove_file(&blocker).unwrap();
1283 record.append(&[first, second]);
1284
1285 let written: Vec<Value> = std::fs::read_to_string(record.path())
1286 .unwrap()
1287 .lines()
1288 .filter_map(parse_message_line)
1289 .collect();
1290 assert_eq!(
1291 written
1292 .iter()
1293 .map(|m| crate::agent::message_id(m).unwrap())
1294 .collect::<Vec<_>>(),
1295 vec![id_at(1), id_at(2)],
1296 "the message the failure cost is on disk, before the one that followed it"
1297 );
1298 }
1299
1300 #[test]
1302 fn a_paging_loop_reads_the_record_once() {
1303 let home = tempfile::tempdir().unwrap();
1304 let workdir = tempfile::tempdir().unwrap();
1305 let root = home.path().join("conversations/capsule");
1306 let mut record = ConversationRecord::open(
1307 &root,
1308 "ctx_1",
1309 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1310 Some(CAPSULE),
1311 )
1312 .unwrap();
1313 record.append(
1314 &(1..=250)
1315 .map(|n| message(&id_at(n), "user", "m"))
1316 .collect::<Vec<_>>(),
1317 );
1318 let path = record.path();
1319 let mut cache = RecordCache::default();
1320
1321 let mut pages = 0;
1322 let mut seen = Vec::new();
1323 let mut cursor: Option<String> = None;
1324 loop {
1325 let page = page(
1326 &mut cache,
1327 Some(&path),
1328 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1329 cursor.as_deref(),
1330 25,
1331 )
1332 .unwrap();
1333 pages += 1;
1334 seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1335 match page.next_cursor {
1336 Some(next) => cursor = Some(next),
1337 None => break,
1338 }
1339 }
1340 assert_eq!(pages, 10);
1341 assert_eq!(seen, (1..=250).rev().map(id_at).collect::<Vec<_>>());
1342 assert_eq!(cache.reads(), 1, "one read for the whole walk");
1343
1344 record.append(
1347 &(251..=260)
1348 .map(|n| message(&id_at(n), "user", "m"))
1349 .collect::<Vec<_>>(),
1350 );
1351 let mut seen = Vec::new();
1352 let mut cursor: Option<String> = None;
1353 let mut first_total = None;
1354 loop {
1355 let page = page(
1356 &mut cache,
1357 Some(&path),
1358 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1359 cursor.as_deref(),
1360 25,
1361 )
1362 .unwrap();
1363 first_total.get_or_insert(page.total);
1364 seen.extend(page.messages.iter().map(|m| m.id.clone().unwrap()));
1365 match page.next_cursor {
1366 Some(next) => cursor = Some(next),
1367 None => break,
1368 }
1369 }
1370 assert_eq!(first_total, Some(260));
1371 assert_eq!(seen, (1..=260).rev().map(id_at).collect::<Vec<_>>());
1372 assert_eq!(cache.reads(), 2, "one further read for the second walk");
1373 }
1374
1375 #[test]
1377 fn a_malformed_line_is_reported_once_for_a_whole_walk() {
1378 let home = tempfile::tempdir().unwrap();
1379 let workdir = tempfile::tempdir().unwrap();
1380 let dir = home.path().join("conversations/capsule/ctx_1");
1381 std::fs::create_dir_all(&dir).unwrap();
1382 let path = dir.join(RECORD_FILE_NAME);
1383 let mut raw = String::new();
1384 for n in 1..=10 {
1385 raw.push_str(&serde_json::to_string(&message(&id_at(n), "user", "m")).unwrap());
1386 raw.push('\n');
1387 }
1388 raw.push_str("{\"role\":\"user\"");
1389 std::fs::write(&path, raw).unwrap();
1390
1391 let mut cache = RecordCache::default();
1392 let mut cursor: Option<String> = None;
1393 let mut pages = 0;
1394 loop {
1395 let page = page(
1396 &mut cache,
1397 Some(&path),
1398 &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
1399 cursor.as_deref(),
1400 2,
1401 )
1402 .unwrap();
1403 pages += 1;
1404 assert_eq!(page.total, 10);
1405 match page.next_cursor {
1406 Some(next) => cursor = Some(next),
1407 None => break,
1408 }
1409 }
1410 assert_eq!(pages, 5);
1411
1412 let log =
1413 std::fs::read_to_string(workdir.path().join("logs/bootstrap.log")).unwrap_or_default();
1414 assert_eq!(
1415 log.lines()
1416 .filter(|line| line.contains("line 11 is not a message"))
1417 .count(),
1418 1,
1419 "one entry for the whole walk: {log}"
1420 );
1421 }
1422
1423 #[test]
1426 fn record_off_wins_over_a_declared_store() {
1427 let off = murmur_artifact::ContextConfig {
1428 max_tokens: None,
1429 record: false,
1430 record_store: Some("shey".to_string()),
1431 seed_budget: murmur_artifact::DEFAULT_SEED_BUDGET,
1432 seed_overflow_margin: murmur_artifact::DEFAULT_SEED_OVERFLOW_MARGIN,
1433 retain: None,
1434 };
1435 assert_eq!(resolve_record_name(Some(&off), "capsule"), None);
1436
1437 let on = murmur_artifact::ContextConfig {
1438 record: true,
1439 ..off.clone()
1440 };
1441 assert_eq!(
1442 resolve_record_name(Some(&on), "capsule"),
1443 Some("shey".to_string())
1444 );
1445 assert_eq!(
1446 resolve_record_name(None, "capsule"),
1447 Some("capsule".to_string()),
1448 "a capsule declaring no context block still records"
1449 );
1450 }
1451
1452 #[test]
1454 fn a_malformed_record_name_is_refused_by_name() {
1455 for value in ["", "../escape", "/abs", "a/b", ".hidden"] {
1456 let err = validate_record_segment("context.record_store", value)
1457 .expect_err("'{value}' must be refused");
1458 let message = err.to_string();
1459 assert!(message.contains(&format!("'{value}'")), "{message}");
1460 assert!(message.contains("single path segment"), "{message}");
1461 assert!(message.contains("context.record_store"), "{message}");
1462 }
1463 validate_record_segment("--context", "ctx_fixed").unwrap();
1464 }
1465}