1use std::io::BufRead;
6use std::path::{Path, PathBuf};
7
8use super::orphan::OrphanToolCall;
9use super::serialize::{entry_to_message, CompactBoundaryEntry, LoadedEntry, TranscriptEntry};
10use super::SessionMeta;
11
12pub struct SessionReader;
14
15impl SessionReader {
16 pub fn load_transcript(session_dir: &Path) -> std::io::Result<Vec<TranscriptEntry>> {
23 let jsonl_path = session_dir.join("transcript.jsonl");
24 let file = std::fs::File::open(&jsonl_path)?;
25 let reader = std::io::BufReader::new(file);
26
27 let mut all_entries: Vec<TranscriptEntry> = Vec::new();
28 let mut boundary_after: usize = 0;
30
31 for line in reader.lines() {
32 let line = line?;
33 if line.trim().is_empty() {
34 continue;
35 }
36 if let Ok(sys) = serde_json::from_str::<CompactBoundaryEntry>(&line) {
39 if sys.entry_type == "system" && sys.subtype == "compact_boundary" {
40 boundary_after = all_entries.len();
42 continue;
43 }
44 }
45 match serde_json::from_str::<TranscriptEntry>(&line) {
46 Ok(entry) => all_entries.push(entry),
47 Err(e) => {
48 eprintln!(
50 "warning: skipping corrupt line in {}: {}",
51 jsonl_path.display(),
52 e
53 );
54 continue;
55 }
56 }
57 }
58 Ok(all_entries.split_off(boundary_after))
60 }
61
62 pub fn load_full_history(session_dir: &Path) -> std::io::Result<Vec<LoadedEntry>> {
77 let jsonl_path = session_dir.join("transcript.jsonl");
78 let file = std::fs::File::open(&jsonl_path)?;
79 let reader = std::io::BufReader::new(file);
80
81 let mut out: Vec<LoadedEntry> = Vec::new();
82 for line in reader.lines() {
83 let line = line?;
84 if line.trim().is_empty() {
85 continue;
86 }
87 if let Ok(sys) = serde_json::from_str::<CompactBoundaryEntry>(&line) {
90 if sys.entry_type == "system" && sys.subtype == "compact_boundary" {
91 out.push(LoadedEntry::CompactBoundary {
92 turn: sys.turn,
93 removed: sys.compacted_count.unwrap_or(0),
94 });
95 continue;
96 }
97 }
98 match serde_json::from_str::<TranscriptEntry>(&line) {
99 Ok(entry) => out.push(LoadedEntry::Message(Box::new(entry))),
100 Err(e) => {
101 eprintln!(
103 "warning: skipping corrupt line in {}: {}",
104 jsonl_path.display(),
105 e
106 );
107 continue;
108 }
109 }
110 }
111 Ok(out)
112 }
113
114 pub fn load_transcript_indexed(
117 session_dir: &Path,
118 ) -> std::io::Result<(
119 Vec<TranscriptEntry>,
120 std::collections::HashMap<String, TranscriptEntry>,
121 )> {
122 let entries = Self::load_transcript(session_dir)?;
123 let mut index = std::collections::HashMap::with_capacity(entries.len());
124 for entry in &entries {
125 if !entry.uuid.is_empty() {
126 index.insert(entry.uuid.clone(), entry.clone());
127 }
128 }
129 Ok((entries, index))
130 }
131
132 pub fn load_messages(session_dir: &Path) -> std::io::Result<Vec<crate::message::Message>> {
142 let entries = Self::load_transcript(session_dir)?;
143 Ok(entries.into_iter().map(entry_to_message).collect())
144 }
145
146 pub fn scan_orphan_tool_calls(
160 session_dir: &Path,
161 registry: &crate::tools::ToolRegistry,
162 ) -> std::io::Result<Vec<OrphanToolCall>> {
163 let entries = Self::load_transcript(session_dir)?;
164 if entries.is_empty() {
165 return Ok(Vec::new());
166 }
167
168 let last_assistant = entries
170 .iter()
171 .enumerate()
172 .rev()
173 .find(|(_, e)| e.role == "assistant" && !e.tool_calls.is_empty());
174
175 let Some((asst_idx, asst_entry)) = last_assistant else {
176 return Ok(Vec::new());
177 };
178
179 let answered: std::collections::HashSet<String> = entries[asst_idx + 1..]
182 .iter()
183 .filter(|e| e.role == "tool")
184 .filter_map(|e| e.tool_call_id.clone())
185 .collect();
186
187 let mut orphans = Vec::new();
188 for tc in &asst_entry.tool_calls {
189 if !answered.contains(&tc.id) {
190 let side_effect = registry
191 .get(&tc.name)
192 .map(|t| t.side_effect_class())
193 .unwrap_or(crate::tools::ToolSideEffect::External);
194 let args_hash = {
195 let canonical = tc.arguments.to_string();
196 let hash = blake3::hash(canonical.as_bytes());
197 hash.to_hex().to_string()
198 };
199 orphans.push(OrphanToolCall {
200 assistant_msg_id: asst_entry.id.clone(),
201 tool_call_id: tc.id.clone(),
202 tool_name: tc.name.clone(),
203 args_hash,
204 side_effect_at_call: side_effect,
205 });
206 }
207 }
208 Ok(orphans)
209 }
210
211 pub fn load_meta(session_dir: &Path) -> std::io::Result<SessionMeta> {
218 let meta_path = session_dir.join(".meta.json");
219 let bytes = std::fs::read(&meta_path)?;
220 let meta: SessionMeta = serde_json::from_slice(&bytes)
221 .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
222 if meta.schema_version > super::SUPPORTED_SESSION_SCHEMA_VERSION {
223 let err = crate::error::Error::SchemaTooNew {
224 session_id: meta.session_id.clone(),
225 found: meta.schema_version,
226 supported: super::SUPPORTED_SESSION_SCHEMA_VERSION,
227 };
228 tracing::warn!("{}", err);
229 return Err(std::io::Error::new(
230 std::io::ErrorKind::InvalidData,
231 err.to_string(),
232 ));
233 }
234 Ok(meta)
235 }
236
237 pub fn list_sessions(workspace: &Path) -> std::io::Result<Vec<PathBuf>> {
242 let base = match crate::paths::user_sessions_dir(workspace) {
243 Ok(d) => d,
244 Err(_) => workspace.join(".recursive").join("sessions"),
245 };
246 if !base.is_dir() {
247 return Ok(Vec::new());
248 }
249
250 let mut sessions = Vec::new();
251 for entry in std::fs::read_dir(&base)? {
253 let entry = entry?;
254 let slug_dir = entry.path();
255 if !slug_dir.is_dir() {
256 continue;
257 }
258 for session_entry in std::fs::read_dir(&slug_dir)? {
260 let session_entry = session_entry?;
261 let session_dir = session_entry.path();
262 if session_dir.is_dir() && session_dir.join(".meta.json").is_file() {
263 sessions.push(session_dir);
264 }
265 }
266 }
267 sessions.sort();
268 Ok(sessions)
269 }
270
271 pub fn list_sessions_sorted_by_updated_at(
283 workspace: &Path,
284 ) -> std::io::Result<Vec<(PathBuf, SessionMeta)>> {
285 let dirs = Self::list_sessions(workspace)?;
286
287 let mut entries: Vec<(PathBuf, SessionMeta, std::time::SystemTime)> = Vec::new();
288 for dir in dirs {
289 let meta = match Self::load_meta(&dir) {
290 Ok(m) => m,
291 Err(_) => continue,
292 };
293 let mtime = std::fs::metadata(dir.join("transcript.jsonl"))
296 .and_then(|m| m.modified())
297 .unwrap_or(std::time::SystemTime::UNIX_EPOCH);
298 entries.push((dir, meta, mtime));
299 }
300
301 entries.sort_by(|a, b| {
302 b.1.updated_at
305 .cmp(&a.1.updated_at)
306 .then_with(|| b.2.cmp(&a.2))
308 .then_with(|| a.1.session_id.cmp(&b.1.session_id))
310 });
311
312 Ok(entries.into_iter().map(|(p, m, _)| (p, m)).collect())
313 }
314
315 pub fn list_all_sessions(base: &Path) -> std::io::Result<Vec<PathBuf>> {
317 let sessions_dir = base.join(".recursive").join("sessions");
318 if !sessions_dir.is_dir() {
319 return Ok(Vec::new());
320 }
321
322 let mut sessions = Vec::new();
323 for entry in std::fs::read_dir(&sessions_dir)? {
324 let entry = entry?;
325 let slug_dir = entry.path();
326 if !slug_dir.is_dir() {
327 continue;
328 }
329 for session_entry in std::fs::read_dir(&slug_dir)? {
330 let session_entry = session_entry?;
331 let session_dir = session_entry.path();
332 if session_dir.is_dir() && session_dir.join(".meta.json").is_file() {
333 sessions.push(session_dir);
334 }
335 }
336 }
337 sessions.sort();
338 Ok(sessions)
339 }
340}
341
342#[cfg(test)]
343mod tests {
344 use super::*;
345 use crate::llm::ToolSpec;
346 use crate::message::{Message, Role};
347 use crate::session::{hash_tool_specs, SessionStatus, SessionWriter};
348
349 #[test]
350 fn session_reader_round_trips() {
351 let tmp = crate::test_util::IsolatedWorkspace::new();
352 let ws = tmp.path();
353
354 let mut writer = SessionWriter::create(ws, "round trip", "gpt-4o", "openai").unwrap();
355
356 writer
357 .append(&Message::system("You are a bot."), None, None)
358 .unwrap();
359 writer
360 .append(&Message::user("do something"), None, None)
361 .unwrap();
362 writer
363 .append(&Message::assistant("I will do it."), None, None)
364 .unwrap();
365
366 let session_dir = writer.session_dir().to_path_buf();
367 writer.finish(SessionStatus::Completed).unwrap();
368
369 let entries = SessionReader::load_transcript(&session_dir).unwrap();
370 assert_eq!(entries.len(), 3);
371 assert_eq!(entries[0].role, "system");
372 assert_eq!(entries[1].role, "user");
373 assert_eq!(entries[2].role, "assistant");
374 }
375
376 #[test]
377 fn crash_partial_line_skipped() {
378 let tmp = crate::test_util::IsolatedWorkspace::new();
379 let ws = tmp.path();
380
381 let mut writer = SessionWriter::create(ws, "crash test", "gpt-4o", "openai").unwrap();
382 writer
383 .append(&Message::user("good line"), None, None)
384 .unwrap();
385 let session_dir = writer.session_dir().to_path_buf();
386 writer.finish(SessionStatus::Crashed).unwrap();
387
388 use std::io::Write;
390 let jsonl_path = session_dir.join("transcript.jsonl");
391 let mut f = std::fs::OpenOptions::new()
392 .append(true)
393 .open(&jsonl_path)
394 .unwrap();
395 writeln!(f, "this is not json").unwrap();
396 drop(f);
397
398 let entries = SessionReader::load_transcript(&session_dir).unwrap();
400 assert_eq!(entries.len(), 1);
401 assert_eq!(entries[0].content, "good line");
402 }
403
404 #[test]
405 fn load_messages_drops_persistence_fields() {
406 let tmp = crate::test_util::IsolatedWorkspace::new();
407 let ws = tmp.path();
408 let mut writer = SessionWriter::create(ws, "g151 test", "model", "openai").unwrap();
409 let session_dir = writer.session_dir().to_path_buf();
410
411 writer
412 .append(&Message::user("hello".to_string()), None, None)
413 .unwrap();
414 writer
415 .append(&Message::assistant("hi back".to_string()), None, None)
416 .unwrap();
417 writer.finish(SessionStatus::Completed).unwrap();
418 drop(writer);
419
420 let msgs = SessionReader::load_messages(&session_dir).unwrap();
422 assert_eq!(msgs.len(), 2);
423 assert_eq!(msgs[0].role, Role::User);
424 assert_eq!(msgs[0].content, "hello");
425 assert_eq!(msgs[1].role, Role::Assistant);
426 assert_eq!(msgs[1].content, "hi back");
427
428 let entries = SessionReader::load_transcript(&session_dir).unwrap();
431 assert_eq!(entries.len(), 2);
432 assert_eq!(entries[0].id, "msg_001");
433 assert!(!entries[0].timestamp.is_empty());
434 }
435
436 #[test]
437 fn meta_round_trip_with_tool_registry_hash() {
438 let tmp = crate::test_util::IsolatedWorkspace::new();
439 let ws = tmp.path();
440
441 let specs = vec![ToolSpec {
442 name: "Read".into(),
443 description: "Read".into(),
444 parameters: serde_json::json!({"type":"object"}),
445 }];
446 let writer =
447 SessionWriter::create_with_tools(ws, "with hash", "model", "openai", &specs, None)
448 .unwrap();
449 let session_dir = writer.session_dir().to_path_buf();
450 drop(writer);
451
452 let meta = SessionReader::load_meta(&session_dir).unwrap();
453 let hash = meta
454 .tool_registry_hash
455 .as_ref()
456 .expect("expected hash to be Some(_)");
457 assert_eq!(*hash, hash_tool_specs(&specs));
458 }
459
460 #[test]
461 fn meta_round_trip_old_format_no_hash() {
462 let tmp = crate::test_util::IsolatedWorkspace::new();
466 let session_dir = tmp
467 .path()
468 .join(".recursive")
469 .join("sessions")
470 .join("legacy");
471 std::fs::create_dir_all(&session_dir).unwrap();
472
473 let raw = r#"{
474 "session_id": "legacy-id",
475 "goal": "old goal",
476 "model": "model",
477 "provider": "openai",
478 "created_at": "2020-01-01T00:00:00Z",
479 "updated_at": "2020-01-01T00:00:00Z",
480 "message_count": 0,
481 "status": "active"
482}"#;
483 std::fs::write(session_dir.join(".meta.json"), raw).unwrap();
484 std::fs::write(session_dir.join("transcript.jsonl"), "").unwrap();
485
486 let meta = SessionReader::load_meta(&session_dir).unwrap();
487 assert_eq!(meta.session_id, "legacy-id");
488 assert!(meta.tool_registry_hash.is_none());
489 }
490
491 #[test]
492 fn append_bumps_updated_at() {
493 let tmp = crate::test_util::IsolatedWorkspace::new();
494 let ws = tmp.path();
495 let mut writer = SessionWriter::create(ws, "bump", "model", "openai").unwrap();
496 let session_dir = writer.session_dir().to_path_buf();
497
498 let meta_before = SessionReader::load_meta(&session_dir).unwrap();
499 std::thread::sleep(std::time::Duration::from_millis(1100));
503
504 writer
505 .append(&Message::user("ping".to_string()), None, None)
506 .unwrap();
507
508 let meta_after = SessionReader::load_meta(&session_dir).unwrap();
509 assert_ne!(
510 meta_before.updated_at, meta_after.updated_at,
511 "expected updated_at to advance after append; before={} after={}",
512 meta_before.updated_at, meta_after.updated_at
513 );
514 }
515
516 #[test]
517 fn open_existing_continues_msg_numbering() {
518 let tmp = crate::test_util::IsolatedWorkspace::new();
519 let ws = tmp.path();
520 let mut writer = SessionWriter::create(ws, "resume-num", "model", "openai").unwrap();
521 let session_dir = writer.session_dir().to_path_buf();
522
523 writer
524 .append(&Message::user("u1".to_string()), None, None)
525 .unwrap();
526 writer
527 .append(&Message::assistant("a1".to_string()), None, None)
528 .unwrap();
529 writer
530 .append(&Message::user("u2".to_string()), None, None)
531 .unwrap();
532 drop(writer);
535
536 let mut writer2 = SessionWriter::open_existing(&session_dir).unwrap();
538 let id = writer2
539 .append(&Message::assistant("a2".to_string()), None, None)
540 .unwrap();
541 assert!(!id.is_empty(), "expected non-empty UUID from append");
543 drop(writer2);
544
545 let entries = SessionReader::load_transcript(&session_dir).unwrap();
546 assert_eq!(entries.len(), 4);
547 assert_eq!(entries[0].id, "msg_001");
548 assert_eq!(entries[1].id, "msg_002");
549 assert_eq!(entries[2].id, "msg_003");
550 assert_eq!(entries[3].id, "msg_004");
551 assert_eq!(entries[3].parent_id.as_deref(), Some("msg_003"));
552 }
553
554 #[test]
555 fn list_sessions_finds_sessions() {
556 let tmp = crate::test_util::IsolatedWorkspace::new();
557 let ws = tmp.path();
558
559 let sessions = SessionReader::list_sessions(ws).unwrap();
561 assert!(sessions.is_empty());
562
563 let writer = SessionWriter::create(ws, "session1", "gpt-4o", "openai").unwrap();
565 let dir1 = writer.session_dir().to_path_buf();
566 drop(writer);
567
568 let sessions = SessionReader::list_sessions(ws).unwrap();
569 assert_eq!(sessions.len(), 1);
570 assert_eq!(sessions[0], dir1);
571 }
572
573 #[test]
574 fn list_sessions_sorted_by_updated_at_returns_one_session_with_meta() {
575 let tmp = crate::test_util::IsolatedWorkspace::new();
577 let ws = tmp.path();
578
579 let w1 = SessionWriter::create(ws, "sort-test-goal", "gpt-4o", "openai").unwrap();
581 drop(w1);
582
583 let sorted = SessionReader::list_sessions_sorted_by_updated_at(ws).unwrap();
584 assert_eq!(sorted.len(), 1, "must find 1 session");
585 assert!(!sorted[0].1.session_id.is_empty());
587 }
588
589 #[test]
590 fn list_all_sessions_returns_empty_for_no_sessions() {
591 let tmp = crate::test_util::IsolatedWorkspace::new();
593 let ws = tmp.path();
594 let sessions = SessionReader::list_all_sessions(ws).unwrap();
595 assert!(sessions.is_empty(), "must be empty when no sessions exist");
596 }
597
598 #[test]
599 fn scan_orphan_tool_calls_returns_empty_for_no_assistant_with_tool_calls() {
600 use crate::message::Message;
602 use crate::session::{SessionStatus, SessionWriter};
603
604 let tmp = crate::test_util::IsolatedWorkspace::new();
605 let mut w = SessionWriter::create(tmp.path(), "g", "m", "p").unwrap();
606 w.append(&Message::user("hi".to_string()), None, None)
608 .unwrap();
609 w.append(&Message::assistant("hello".to_string()), None, None)
610 .unwrap();
611 w.finish(SessionStatus::Completed).unwrap();
612
613 let reg = crate::tools::ToolRegistry::local();
614 let orphans = SessionReader::scan_orphan_tool_calls(w.session_dir(), ®).unwrap();
615 assert!(
616 orphans.is_empty(),
617 "no assistant tool_calls → no orphans; got: {orphans:?}"
618 );
619 }
620
621 #[test]
622 fn scan_orphan_tool_calls_returns_empty_when_all_answered() {
623 use crate::message::Message;
625 use crate::session::{SessionStatus, SessionWriter};
626
627 let tmp = crate::test_util::IsolatedWorkspace::new();
628 let mut w = SessionWriter::create(tmp.path(), "g", "m", "p").unwrap();
629
630 let mut asst = Message::assistant("calling tool".to_string());
632 asst.tool_calls = vec![crate::llm::ToolCall {
633 id: "tc-001".to_string(),
634 name: "Read".to_string(),
635 arguments: serde_json::json!({"path": "foo.txt"}),
636 }];
637
638 let tool_result = Message {
639 role: crate::message::Role::Tool,
640 content: "file contents".to_string(),
641 tool_calls: vec![],
642 tool_call_id: Some("tc-001".to_string()),
643 reasoning_content: None,
644 is_compaction_summary: false,
645 };
646
647 w.append(&Message::user("do something".to_string()), None, None)
648 .unwrap();
649 w.append(&asst, None, None).unwrap();
650 w.append(&tool_result, None, None).unwrap();
651 w.finish(SessionStatus::Completed).unwrap();
652
653 let reg = crate::tools::ToolRegistry::local();
654 let orphans = SessionReader::scan_orphan_tool_calls(w.session_dir(), ®).unwrap();
655 assert!(
656 orphans.is_empty(),
657 "all tool calls answered → no orphans; got: {orphans:?}"
658 );
659 }
660
661 #[test]
662 fn scan_orphan_tool_calls_detects_unanswered_call() {
663 use crate::message::Message;
665 use crate::session::{SessionStatus, SessionWriter};
666
667 let tmp = crate::test_util::IsolatedWorkspace::new();
668 let mut w = SessionWriter::create(tmp.path(), "g", "m", "p").unwrap();
669
670 let mut asst = Message::assistant("calling tool".to_string());
672 asst.tool_calls = vec![crate::llm::ToolCall {
673 id: "tc-orphan".to_string(),
674 name: "Read".to_string(),
675 arguments: serde_json::json!({"path": "missing.txt"}),
676 }];
677
678 w.append(&Message::user("do something".to_string()), None, None)
679 .unwrap();
680 w.append(&asst, None, None).unwrap();
681 w.finish(SessionStatus::Completed).unwrap();
683
684 let reg = crate::tools::ToolRegistry::local();
685 let orphans = SessionReader::scan_orphan_tool_calls(w.session_dir(), ®).unwrap();
686 assert_eq!(
687 orphans.len(),
688 1,
689 "one unanswered tool call must be reported as orphan"
690 );
691 assert_eq!(orphans[0].tool_call_id, "tc-orphan");
692 assert_eq!(orphans[0].tool_name, "Read");
693 }
694
695 #[test]
700 fn load_full_history_includes_pre_compact_messages_and_boundaries() {
701 let tmp = crate::test_util::IsolatedWorkspace::new();
702 let ws = tmp.path();
703 let mut w = SessionWriter::create(ws, "full history", "model", "openai").unwrap();
704 let session_dir = w.session_dir().to_path_buf();
705
706 w.append(&Message::user("turn 1".to_string()), None, None)
708 .unwrap();
709 w.append(&Message::assistant("reply 1".to_string()), None, None)
710 .unwrap();
711 w.append(&Message::user("turn 2".to_string()), None, None)
712 .unwrap();
713 w.append(&Message::assistant("reply 2".to_string()), None, None)
714 .unwrap();
715 w.write_compact_boundary(3, 4, None).unwrap();
717 w.append(
719 &Message::system("summary of turn 1-2".to_string()),
720 None,
721 None,
722 )
723 .unwrap();
724 w.append(&Message::user("turn 3".to_string()), None, None)
725 .unwrap();
726 w.append(&Message::assistant("reply 3".to_string()), None, None)
727 .unwrap();
728 w.finish(SessionStatus::Completed).unwrap();
729
730 let loaded = SessionReader::load_full_history(&session_dir).unwrap();
732 assert_eq!(loaded.len(), 8, "all messages + the boundary marker");
733
734 let boundary_idx = loaded
736 .iter()
737 .position(
738 |e| matches!(e, LoadedEntry::CompactBoundary { removed, .. } if *removed == 4),
739 )
740 .expect("a CompactBoundary marker with removed=4");
741 assert_eq!(boundary_idx, 4, "boundary follows the 4 folded messages");
742
743 let pre = &loaded[..boundary_idx];
745 assert_eq!(pre.len(), 4);
746 assert!(pre.iter().all(|e| matches!(e, LoadedEntry::Message(_))));
747
748 let post = &loaded[boundary_idx + 1..];
749 assert_eq!(post.len(), 3);
750 assert!(post.iter().all(|e| matches!(e, LoadedEntry::Message(_))));
751
752 if let LoadedEntry::Message(m) = &post[0] {
754 assert_eq!(m.role, "system");
755 assert_eq!(m.content, "summary of turn 1-2");
756 } else {
757 panic!("expected a Message after the boundary");
758 }
759
760 let seed = SessionReader::load_messages(&session_dir).unwrap();
763 assert_eq!(
764 seed.len(),
765 3,
766 "load_messages keeps only post-compaction tail"
767 );
768 assert_eq!(seed[0].content, "summary of turn 1-2");
769 }
770
771 #[test]
772 fn load_full_history_no_boundary_returns_all_messages() {
773 let tmp = crate::test_util::IsolatedWorkspace::new();
774 let ws = tmp.path();
775 let mut w = SessionWriter::create(ws, "no compact", "model", "openai").unwrap();
776 let session_dir = w.session_dir().to_path_buf();
777 w.append(&Message::user("hi".to_string()), None, None)
778 .unwrap();
779 w.append(&Message::assistant("hello".to_string()), None, None)
780 .unwrap();
781 w.finish(SessionStatus::Completed).unwrap();
782
783 let loaded = SessionReader::load_full_history(&session_dir).unwrap();
784 assert_eq!(loaded.len(), 2);
785 assert!(
786 loaded.iter().all(|e| matches!(e, LoadedEntry::Message(_))),
787 "no CompactBoundary markers when the session never compacted"
788 );
789 }
790}