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