1use crate::store::temporary_sibling;
22use crate::{SessionEntry, SessionError, SessionInfo, SessionMetadata};
23use chrono::Utc;
24use std::fs::{self, OpenOptions};
25use std::io::{Read, Seek, SeekFrom, Write};
26use std::path::Path;
27use uuid::Uuid;
28
29const TLOG_MAGIC: &str = "TALOS";
31const TLOG_VERSION: u8 = 1;
33const KIND_ENTRY: char = 'E';
35
36#[derive(Debug, Clone, Copy, Default)]
41pub struct CompactTextSessionStore;
42
43impl crate::store::SessionStore for CompactTextSessionStore {
44 fn read_entries(&self, file_path: &Path) -> Result<Vec<SessionEntry>, SessionError> {
45 read_tlog_entries(file_path)
46 }
47
48 fn append_entry(&self, file_path: &Path, entry: &SessionEntry) -> Result<(), SessionError> {
49 if !file_path.exists() {
50 if let Some(parent) = file_path.parent() {
52 fs::create_dir_all(parent)?;
53 }
54 let mut file = OpenOptions::new()
55 .create(true)
56 .write(true)
57 .truncate(true)
58 .open(file_path)?;
59 writeln!(
60 file,
61 "{TLOG_MAGIC}\t{TLOG_VERSION}\t{}",
62 Utc::now().timestamp_millis()
63 )?;
64 file.flush()?;
65 }
66
67 let line = encode_entry(entry);
68 let mut file = OpenOptions::new().append(true).open(file_path)?;
69 file.write_all(line.as_bytes())?;
70 file.flush()?;
71 Ok(())
72 }
73
74 fn replace_entries_atomically(
75 &self,
76 file_path: &Path,
77 entries: &[SessionEntry],
78 ) -> Result<(), SessionError> {
79 let parent = file_path.parent().ok_or_else(|| {
80 SessionError::ParseError("session file has no parent directory".into())
81 })?;
82 fs::create_dir_all(parent)?;
83 let temporary = temporary_sibling(file_path);
84 let mut file = OpenOptions::new()
85 .create_new(true)
86 .write(true)
87 .open(&temporary)?;
88 writeln!(
89 file,
90 "{TLOG_MAGIC}\t{TLOG_VERSION}\t{}",
91 Utc::now().timestamp_millis()
92 )?;
93 for entry in entries {
94 file.write_all(encode_entry(entry).as_bytes())?;
95 }
96 file.sync_all()?;
97 drop(file);
98 fs::rename(&temporary, file_path)?;
99 Ok(())
100 }
101
102 fn read_last_entry_id(&self, file_path: &Path) -> Option<String> {
103 read_last_entry_id_tlog(file_path)
104 }
105
106 fn scan_file(&self, file_path: &Path) -> Result<SessionInfo, SessionError> {
107 scan_tlog_file(file_path)
108 }
109
110 fn read_bytes(&self, file_path: &Path) -> Result<Vec<u8>, SessionError> {
111 std::fs::read(file_path).map_err(SessionError::IoError)
112 }
113
114 fn file_extension(&self) -> &'static str {
115 "tlog"
116 }
117}
118
119fn encode_entry(entry: &SessionEntry) -> String {
127 let role_num = role_to_num(&entry.role);
128 let parent = entry.parent_id.as_deref().unwrap_or("-");
129 let meta_json = if entry.metadata.is_empty() {
130 String::from("{}")
131 } else {
132 serde_json::to_string(&entry.metadata).unwrap_or_else(|_| String::from("{}"))
133 };
134 let ts_ms = entry.timestamp.timestamp_millis();
135
136 format!(
140 "{KIND_ENTRY}\t{role_num}\t{ts_ms}\t{}\t{parent}\t{}:{}\t{}:{}\n",
141 entry.id,
142 entry.content.len(),
143 entry.content,
144 meta_json.len(),
145 meta_json,
146 )
147}
148
149fn role_to_num(role: &str) -> u8 {
151 match role {
152 "user" => 0,
153 "assistant" => 1,
154 "system" => 2,
155 _ => 3, }
157}
158
159fn num_to_role(num: u8) -> String {
161 match num {
162 0 => "user".into(),
163 1 => "assistant".into(),
164 2 => "system".into(),
165 _ => format!("unknown-{num}"),
166 }
167}
168
169fn read_tlog_entries(path: &Path) -> Result<Vec<SessionEntry>, SessionError> {
178 if !path.exists() {
179 return Ok(Vec::new());
180 }
181
182 let data = std::fs::read(path)?;
183 parse_tlog_bytes(&data)
184}
185
186fn parse_tlog_bytes(data: &[u8]) -> Result<Vec<SessionEntry>, SessionError> {
187 let text =
188 std::str::from_utf8(data).map_err(|_| SessionError::ParseError("invalid UTF-8".into()))?;
189 let mut entries = Vec::new();
190 let mut pos = 0;
191 let mut first_line = true;
192
193 while pos < text.len() {
194 let remaining = &text[pos..];
195
196 let newline_pos = match remaining.find('\n') {
200 Some(p) => p,
201 None => {
202 break;
204 }
205 };
206
207 let line = &remaining[..newline_pos];
208
209 if first_line {
210 first_line = false;
211 if line.starts_with(TLOG_MAGIC) {
212 pos += newline_pos + 1;
213 continue;
214 }
215 }
216
217 if line.is_empty() {
218 pos += newline_pos + 1;
219 continue;
220 }
221
222 match try_parse_record(remaining) {
226 Ok((entry, consumed)) => {
227 entries.push(entry);
228 pos += consumed;
229 }
230 Err(DecodeError::Skip) => {
231 let mut search_pos = pos + newline_pos + 1;
232 while search_pos < text.len() {
233 let search_remaining = &text[search_pos..];
234 if let Some(nl) = search_remaining.find('\n') {
235 let candidate = &search_remaining[..nl];
236 if candidate.starts_with("E\t")
237 || candidate.is_empty()
238 || candidate.starts_with(TLOG_MAGIC)
239 {
240 pos = search_pos;
241 break;
242 }
243 search_pos += nl + 1;
244 } else {
245 pos = text.len();
246 break;
247 }
248 }
249 if search_pos >= text.len() {
250 pos = text.len();
251 }
252 }
253 Err(DecodeError::Fatal(e)) => return Err(e),
254 }
255 }
256
257 Ok(entries)
258}
259
260fn try_parse_record(text: &str) -> Result<(SessionEntry, usize), DecodeError> {
261 let bytes = text.as_bytes();
262 let mut tab_positions = Vec::new();
263 for (i, &b) in bytes.iter().enumerate() {
264 if b == b'\t' {
265 tab_positions.push(i);
266 if tab_positions.len() == 5 {
267 break;
268 }
269 }
270 }
271
272 if tab_positions.len() < 5 {
273 return Err(DecodeError::Skip);
274 }
275
276 let kind = &text[..tab_positions[0]];
277 if kind != "E" {
278 return Err(DecodeError::Skip);
279 }
280
281 let role_num: u8 = text[tab_positions[0] + 1..tab_positions[1]]
282 .parse()
283 .map_err(|_| DecodeError::Skip)?;
284
285 let ts_ms_str = &text[tab_positions[1] + 1..tab_positions[2]];
286 let ts_ms: i64 = ts_ms_str.parse().map_err(|_| DecodeError::Skip)?;
287
288 let id = text[tab_positions[2] + 1..tab_positions[3]].to_string();
289
290 let parent_field = &text[tab_positions[3] + 1..tab_positions[4]];
291 let parent_id = if parent_field == "-" {
292 None
293 } else {
294 Some(parent_field.to_string())
295 };
296
297 let rest_start = tab_positions[4] + 1;
298 let rest = &text[rest_start..];
299
300 let colon_pos = rest.find(':').ok_or(DecodeError::Skip)?;
301 let content_len: usize = rest[..colon_pos].parse().map_err(|_| DecodeError::Skip)?;
302
303 let content_start = colon_pos + 1;
304 let content_end = content_start + content_len;
305 if content_end > rest.len() {
306 return Err(DecodeError::Skip);
307 }
308
309 let content = rest[content_start..content_end].to_string();
310
311 let after_content = &rest[content_end..];
312 if !after_content.starts_with('\t') {
313 return Err(DecodeError::Skip);
314 }
315
316 let meta_part = &after_content[1..];
317
318 let meta_colon = meta_part.find(':').ok_or(DecodeError::Skip)?;
319 let meta_len: usize = meta_part[..meta_colon]
320 .parse()
321 .map_err(|_| DecodeError::Skip)?;
322
323 let meta_start = meta_colon + 1;
324 let meta_end = meta_start + meta_len;
325 if meta_end > meta_part.len() {
326 return Err(DecodeError::Skip);
327 }
328
329 let meta_str = &meta_part[meta_start..meta_end];
330 let metadata: SessionMetadata = if meta_str == "{}" || meta_str.is_empty() {
331 SessionMetadata::default()
332 } else {
333 serde_json::from_str(meta_str).map_err(|_| DecodeError::Skip)?
334 };
335
336 let timestamp = chrono::DateTime::from_timestamp_millis(ts_ms).unwrap_or_else(Utc::now);
337
338 let entry = SessionEntry {
339 id,
340 parent_id,
341 timestamp,
342 role: num_to_role(role_num),
343 content,
344 metadata,
345 };
346
347 let total_consumed = rest_start + content_end + 1 + meta_colon + 1 + meta_len + 1; Ok((entry, total_consumed))
350}
351
352#[allow(dead_code)]
354enum DecodeError {
355 Skip,
357 Fatal(SessionError),
359}
360
361fn read_last_entry_id_tlog(path: &Path) -> Option<String> {
366 let mut file = fs::File::open(path).ok()?;
367 let file_size = file.metadata().ok()?.len();
368 if file_size == 0 {
369 return None;
370 }
371
372 let read_size = std::cmp::min(file_size, 16384) as usize;
374 let seek_pos = file_size.saturating_sub(read_size as u64);
375 file.seek(SeekFrom::Start(seek_pos)).ok()?;
376 let mut buf = vec![0u8; read_size];
377 file.read_exact(&mut buf).ok()?;
378 let text = String::from_utf8_lossy(&buf);
379
380 let mut last_id = None;
382 let mut first_line = true;
383 for line in text.lines() {
384 if first_line {
385 first_line = false;
386 if seek_pos == 0 && line.starts_with(TLOG_MAGIC) {
388 continue;
389 }
390 }
391 if line.is_empty() {
392 continue;
393 }
394 if let Some(id) = extract_id_from_line(line) {
396 last_id = Some(id);
397 }
398 }
399 last_id
400}
401
402fn extract_id_from_line(line: &str) -> Option<String> {
404 let parts: Vec<&str> = line.splitn(6, '\t').collect();
405 if parts.len() < 4 {
406 return None;
407 }
408 if parts[0] != "E" {
410 return None;
411 }
412 Some(parts[3].to_string())
413}
414
415fn scan_tlog_file(path: &Path) -> Result<SessionInfo, SessionError> {
420 let file = fs::File::open(path)?;
421 let metadata = file.metadata()?;
422 let timestamp = metadata
423 .modified()
424 .ok()
425 .map(chrono::DateTime::<Utc>::from)
426 .unwrap_or_else(Utc::now);
427
428 let id = path
429 .file_stem()
430 .and_then(|s| s.to_str())
431 .and_then(|s| Uuid::parse_str(s).ok())
432 .unwrap_or_else(Uuid::nil);
433
434 let data = std::fs::read(path)?;
435 let entries = parse_tlog_bytes(&data)?;
436
437 let count = entries.len();
438 let last_preview = entries
439 .last()
440 .map(|e| crate::jsonl::preview_text(&e.content))
441 .unwrap_or_default();
442
443 Ok(SessionInfo {
444 id,
445 project: String::new(),
446 workspace_root: String::new(),
447 last_message_preview: last_preview,
448 timestamp,
449 message_count: count,
450 })
451}
452
453#[cfg(test)]
458#[allow(warnings)]
459mod tests {
460 use super::*;
461 use crate::JsonlSessionStore;
462 use crate::store::SessionStore;
463 use chrono::TimeZone;
464 use std::io::Read;
465
466 fn make_entry(role: &str, content: &str) -> SessionEntry {
467 SessionEntry {
468 id: Uuid::new_v4().to_string(),
469 parent_id: None,
470 timestamp: Utc::now(),
471 role: role.into(),
472 content: content.into(),
473 metadata: SessionMetadata::default(),
474 }
475 }
476
477 fn make_entry_with_meta(role: &str, content: &str, meta: SessionMetadata) -> SessionEntry {
478 SessionEntry {
479 id: Uuid::new_v4().to_string(),
480 parent_id: None,
481 timestamp: Utc::now(),
482 role: role.into(),
483 content: content.into(),
484 metadata: meta,
485 }
486 }
487
488 #[test]
489 fn round_trip_basic() {
490 let store = CompactTextSessionStore;
491 let dir = std::env::temp_dir().join("tlog_test_roundtrip_basic");
492 let _ = std::fs::remove_dir_all(&dir);
493 std::fs::create_dir_all(&dir).expect("operation should succeed");
494 let path = dir.join("test.tlog");
495
496 let entry = make_entry("user", "Hello, world!");
497 store
498 .append_entry(&path, &entry)
499 .expect("operation should succeed");
500
501 let entries = store.read_entries(&path).expect("operation should succeed");
502 assert_eq!(entries.len(), 1);
503 assert_eq!(entries[0].role, "user");
504 assert_eq!(entries[0].content, "Hello, world!");
505 assert_eq!(entries[0].id, entry.id);
506
507 std::fs::remove_dir_all(&dir).ok();
508 }
509
510 #[test]
511 fn round_trip_multiple_entries() {
512 let store = CompactTextSessionStore;
513 let dir = std::env::temp_dir().join("tlog_test_roundtrip_multi");
514 let _ = std::fs::remove_dir_all(&dir);
515 std::fs::create_dir_all(&dir).expect("operation should succeed");
516 let path = dir.join("multi.tlog");
517
518 let entries = vec![
519 make_entry("user", "What is 2+2?"),
520 make_entry("assistant", "The answer is 4."),
521 make_entry("user", "Thanks!"),
522 ];
523
524 for e in &entries {
525 store
526 .append_entry(&path, e)
527 .expect("operation should succeed");
528 }
529
530 let read = store.read_entries(&path).expect("operation should succeed");
531 assert_eq!(read.len(), 3);
532 assert_eq!(read[0].content, "What is 2+2?");
533 assert_eq!(read[1].content, "The answer is 4.");
534 assert_eq!(read[2].content, "Thanks!");
535
536 std::fs::remove_dir_all(&dir).ok();
537 }
538
539 #[test]
540 fn round_trip_with_parent_id() {
541 let store = CompactTextSessionStore;
542 let dir = std::env::temp_dir().join("tlog_test_parent");
543 let _ = std::fs::remove_dir_all(&dir);
544 std::fs::create_dir_all(&dir).expect("operation should succeed");
545 let path = dir.join("parent.tlog");
546
547 let parent = make_entry("user", "parent message");
548 let mut child = make_entry("assistant", "child response");
549 child.parent_id = Some(parent.id.clone());
550
551 store
552 .append_entry(&path, &parent)
553 .expect("operation should succeed");
554 store
555 .append_entry(&path, &child)
556 .expect("operation should succeed");
557
558 let read = store.read_entries(&path).expect("operation should succeed");
559 assert_eq!(read.len(), 2);
560 assert!(read[0].parent_id.is_none());
561 assert_eq!(read[1].parent_id, Some(parent.id.clone()));
562
563 std::fs::remove_dir_all(&dir).ok();
564 }
565
566 #[test]
567 fn round_trip_with_metadata() {
568 let store = CompactTextSessionStore;
569 let dir = std::env::temp_dir().join("tlog_test_meta");
570 let _ = std::fs::remove_dir_all(&dir);
571 std::fs::create_dir_all(&dir).expect("operation should succeed");
572 let path = dir.join("meta.tlog");
573
574 let meta = SessionMetadata {
575 turn_id: None,
576 provider: Some("anthropic".into()),
577 model: Some("claude-sonnet-4".into()),
578 token_count: Some(42),
579 working_directory: None,
580 reasoning: None,
581 raw_content: None,
582 };
583 let entry = make_entry_with_meta("assistant", "Response with metadata", meta);
584
585 store
586 .append_entry(&path, &entry)
587 .expect("operation should succeed");
588
589 let read = store.read_entries(&path).expect("operation should succeed");
590 assert_eq!(read.len(), 1);
591 assert_eq!(read[0].metadata.provider, Some("anthropic".into()));
592 assert_eq!(read[0].metadata.model, Some("claude-sonnet-4".into()));
593 assert_eq!(read[0].metadata.token_count, Some(42));
594
595 std::fs::remove_dir_all(&dir).ok();
596 }
597
598 #[test]
599 fn content_with_tabs_and_newlines() {
600 let store = CompactTextSessionStore;
601 let dir = std::env::temp_dir().join("tlog_test_special");
602 let _ = std::fs::remove_dir_all(&dir);
603 std::fs::create_dir_all(&dir).expect("operation should succeed");
604 let path = dir.join("special.tlog");
605
606 let tricky = "line1\nline2\twith\ttabs\nline3\twith more";
608 let entry = make_entry("assistant", tricky);
609
610 store
611 .append_entry(&path, &entry)
612 .expect("operation should succeed");
613
614 let read = store.read_entries(&path).expect("operation should succeed");
615 assert_eq!(read.len(), 1);
616 assert_eq!(read[0].content, tricky);
617
618 std::fs::remove_dir_all(&dir).ok();
619 }
620
621 #[test]
622 fn content_with_unicode() {
623 let store = CompactTextSessionStore;
624 let dir = std::env::temp_dir().join("tlog_test_unicode");
625 let _ = std::fs::remove_dir_all(&dir);
626 std::fs::create_dir_all(&dir).expect("operation should succeed");
627 let path = dir.join("unicode.tlog");
628
629 let entry = make_entry("user", "你好世界 🌍 Привет мир");
630
631 store
632 .append_entry(&path, &entry)
633 .expect("operation should succeed");
634
635 let read = store.read_entries(&path).expect("operation should succeed");
636 assert_eq!(read.len(), 1);
637 assert_eq!(read[0].content, "你好世界 🌍 Привет мир");
638
639 std::fs::remove_dir_all(&dir).ok();
640 }
641
642 #[test]
643 fn corrupt_final_record_skipped() {
644 let dir = std::env::temp_dir().join("tlog_test_corrupt");
645 let _ = std::fs::remove_dir_all(&dir);
646 std::fs::create_dir_all(&dir).expect("operation should succeed");
647 let path = dir.join("corrupt.tlog");
648
649 let store = CompactTextSessionStore;
650
651 store
653 .append_entry(&path, &make_entry("user", "first"))
654 .expect("operation should succeed");
655 store
656 .append_entry(&path, &make_entry("assistant", "second"))
657 .expect("operation should succeed");
658
659 std::fs::OpenOptions::new()
661 .append(true)
662 .open(&path)
663 .expect("operation should succeed")
664 .write_all(b"E\t0\t999\tincomplete\t-\t")
665 .expect("operation should succeed");
666
667 let read = store.read_entries(&path).expect("operation should succeed");
668 assert_eq!(read.len(), 2, "corrupt final record should be skipped");
669 assert_eq!(read[0].content, "first");
670 assert_eq!(read[1].content, "second");
671
672 std::fs::remove_dir_all(&dir).ok();
673 }
674
675 #[test]
676 fn read_last_entry_id_returns_last_valid() {
677 let dir = std::env::temp_dir().join("tlog_test_last_id");
678 let _ = std::fs::remove_dir_all(&dir);
679 std::fs::create_dir_all(&dir).expect("operation should succeed");
680 let path = dir.join("last_id.tlog");
681
682 let store = CompactTextSessionStore;
683 let e1 = make_entry("user", "first");
684 let e2 = make_entry("assistant", "second");
685 let e3 = make_entry("user", "third");
686
687 store
688 .append_entry(&path, &e1)
689 .expect("operation should succeed");
690 store
691 .append_entry(&path, &e2)
692 .expect("operation should succeed");
693 store
694 .append_entry(&path, &e3)
695 .expect("operation should succeed");
696
697 let last_id = store.read_last_entry_id(&path);
698 assert_eq!(last_id, Some(e3.id));
699
700 std::fs::remove_dir_all(&dir).ok();
701 }
702
703 #[test]
704 fn read_empty_file_returns_empty() {
705 let dir = std::env::temp_dir().join("tlog_test_empty");
706 let _ = std::fs::remove_dir_all(&dir);
707 std::fs::create_dir_all(&dir).expect("operation should succeed");
708 let path = dir.join("nonexistent.tlog");
709
710 let store = CompactTextSessionStore;
711 let entries = store.read_entries(&path).expect("operation should succeed");
712 assert!(entries.is_empty());
713
714 std::fs::remove_dir_all(&dir).ok();
715 }
716
717 #[test]
718 fn scan_file_counts_and_previews() {
719 let dir = std::env::temp_dir().join("tlog_test_scan");
720 let _ = std::fs::remove_dir_all(&dir);
721 std::fs::create_dir_all(&dir).expect("operation should succeed");
722 let path = dir.join("scan.tlog");
723
724 let store = CompactTextSessionStore;
725 store
726 .append_entry(&path, &make_entry("user", "first message"))
727 .expect("operation should succeed");
728 store
729 .append_entry(&path, &make_entry("assistant", "second message"))
730 .expect("operation should succeed");
731
732 let info = store.scan_file(&path).expect("operation should succeed");
733 assert_eq!(info.message_count, 2);
734
735 std::fs::remove_dir_all(&dir).ok();
736 }
737
738 #[test]
739 fn file_extension_is_tlog() {
740 let store = CompactTextSessionStore;
741 assert_eq!(store.file_extension(), "tlog");
742 }
743
744 #[test]
745 fn density_comparison() {
746 let dir = std::env::temp_dir().join("tlog_test_density");
748 let _ = std::fs::remove_dir_all(&dir);
749 std::fs::create_dir_all(&dir).expect("operation should succeed");
750
751 let jsonl_path = dir.join("session.jsonl");
752 let tlog_path = dir.join("session.tlog");
753
754 let entries: Vec<SessionEntry> = (0..50)
755 .map(|i| SessionEntry {
756 id: Uuid::new_v4().to_string(),
757 parent_id: if i > 0 {
758 Some(format!("prev-{i}"))
759 } else {
760 None
761 },
762 timestamp: Utc::now(),
763 role: if i % 2 == 0 {
764 "user".to_string()
765 } else {
766 "assistant".to_string()
767 },
768 content: format!(
769 "This is message number {i} with some content to simulate a real conversation."
770 ),
771 metadata: SessionMetadata {
772 turn_id: None,
773 provider: Some("anthropic".into()),
774 model: Some("claude-sonnet-4-20250514".into()),
775 token_count: Some(100 + i as u32),
776 working_directory: None,
777 reasoning: None,
778 raw_content: None,
779 },
780 })
781 .collect();
782
783 let jsonl_store = JsonlSessionStore;
785 for e in &entries {
786 jsonl_store
787 .append_entry(&jsonl_path, e)
788 .expect("operation should succeed");
789 }
790
791 let tlog_store = CompactTextSessionStore;
793 for e in &entries {
794 tlog_store
795 .append_entry(&tlog_path, e)
796 .expect("operation should succeed");
797 }
798
799 let jsonl_size = std::fs::metadata(&jsonl_path)
800 .expect("operation should succeed")
801 .len();
802 let tlog_size = std::fs::metadata(&tlog_path)
803 .expect("operation should succeed")
804 .len();
805
806 println!("JSONL: {jsonl_size} bytes");
807 println!("TLOG: {tlog_size} bytes");
808 println!(
809 "Ratio: {:.1}%",
810 (tlog_size as f64 / jsonl_size as f64) * 100.0
811 );
812 println!(
813 "Saving: {:.1}%",
814 (1.0 - tlog_size as f64 / jsonl_size as f64) * 100.0
815 );
816
817 assert!(
819 tlog_size < jsonl_size,
820 ".tlog ({tlog_size}) should be smaller than JSONL ({jsonl_size})"
821 );
822
823 let jsonl_entries = jsonl_store
825 .read_entries(&jsonl_path)
826 .expect("operation should succeed");
827 let tlog_entries = tlog_store
828 .read_entries(&tlog_path)
829 .expect("operation should succeed");
830 assert_eq!(jsonl_entries.len(), tlog_entries.len());
831 assert_eq!(jsonl_entries.len(), 50);
832
833 std::fs::remove_dir_all(&dir).ok();
834 }
835}