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