use std::io::BufRead;
use std::path::{Path, PathBuf};
use super::orphan::OrphanToolCall;
use super::serialize::{entry_to_message, CompactBoundaryEntry, LoadedEntry, TranscriptEntry};
use super::SessionMeta;
pub struct SessionReader;
impl SessionReader {
pub fn load_transcript(session_dir: &Path) -> std::io::Result<Vec<TranscriptEntry>> {
let jsonl_path = session_dir.join("transcript.jsonl");
let file = std::fs::File::open(&jsonl_path)?;
let reader = std::io::BufReader::new(file);
let mut all_entries: Vec<TranscriptEntry> = Vec::new();
let mut boundary_after: usize = 0;
for line in reader.lines() {
let line = line?;
if line.trim().is_empty() {
continue;
}
if let Ok(sys) = serde_json::from_str::<CompactBoundaryEntry>(&line) {
if sys.entry_type == "system" && sys.subtype == "compact_boundary" {
boundary_after = all_entries.len();
continue;
}
}
match serde_json::from_str::<TranscriptEntry>(&line) {
Ok(entry) => all_entries.push(entry),
Err(e) => {
eprintln!(
"warning: skipping corrupt line in {}: {}",
jsonl_path.display(),
e
);
continue;
}
}
}
Ok(all_entries.split_off(boundary_after))
}
pub fn load_full_history(session_dir: &Path) -> std::io::Result<Vec<LoadedEntry>> {
let jsonl_path = session_dir.join("transcript.jsonl");
let file = std::fs::File::open(&jsonl_path)?;
let reader = std::io::BufReader::new(file);
let mut out: Vec<LoadedEntry> = Vec::new();
for line in reader.lines() {
let line = line?;
if line.trim().is_empty() {
continue;
}
if let Ok(sys) = serde_json::from_str::<CompactBoundaryEntry>(&line) {
if sys.entry_type == "system" && sys.subtype == "compact_boundary" {
out.push(LoadedEntry::CompactBoundary {
turn: sys.turn,
removed: sys.compacted_count.unwrap_or(0),
});
continue;
}
}
match serde_json::from_str::<TranscriptEntry>(&line) {
Ok(entry) => out.push(LoadedEntry::Message(Box::new(entry))),
Err(e) => {
eprintln!(
"warning: skipping corrupt line in {}: {}",
jsonl_path.display(),
e
);
continue;
}
}
}
Ok(out)
}
pub fn load_transcript_indexed(
session_dir: &Path,
) -> std::io::Result<(
Vec<TranscriptEntry>,
std::collections::HashMap<String, TranscriptEntry>,
)> {
let entries = Self::load_transcript(session_dir)?;
let mut index = std::collections::HashMap::with_capacity(entries.len());
for entry in &entries {
if !entry.uuid.is_empty() {
index.insert(entry.uuid.clone(), entry.clone());
}
}
Ok((entries, index))
}
pub fn load_messages(session_dir: &Path) -> std::io::Result<Vec<crate::message::Message>> {
let entries = Self::load_transcript(session_dir)?;
Ok(entries.into_iter().map(entry_to_message).collect())
}
pub fn scan_orphan_tool_calls(
session_dir: &Path,
registry: &crate::tools::ToolRegistry,
) -> std::io::Result<Vec<OrphanToolCall>> {
let entries = Self::load_transcript(session_dir)?;
if entries.is_empty() {
return Ok(Vec::new());
}
let last_assistant = entries
.iter()
.enumerate()
.rev()
.find(|(_, e)| e.role == "assistant" && !e.tool_calls.is_empty());
let Some((asst_idx, asst_entry)) = last_assistant else {
return Ok(Vec::new());
};
let answered: std::collections::HashSet<String> = entries[asst_idx + 1..]
.iter()
.filter(|e| e.role == "tool")
.filter_map(|e| e.tool_call_id.clone())
.collect();
let mut orphans = Vec::new();
for tc in &asst_entry.tool_calls {
if !answered.contains(&tc.id) {
let side_effect = registry
.get(&tc.name)
.map(|t| t.side_effect_class())
.unwrap_or(crate::tools::ToolSideEffect::External);
let args_hash = {
let canonical = tc.arguments.to_string();
let hash = blake3::hash(canonical.as_bytes());
hash.to_hex().to_string()
};
orphans.push(OrphanToolCall {
assistant_msg_id: asst_entry.id.clone(),
tool_call_id: tc.id.clone(),
tool_name: tc.name.clone(),
args_hash,
side_effect_at_call: side_effect,
});
}
}
Ok(orphans)
}
pub fn load_meta(session_dir: &Path) -> std::io::Result<SessionMeta> {
let meta_path = session_dir.join(".meta.json");
let bytes = std::fs::read(&meta_path)?;
let meta: SessionMeta = serde_json::from_slice(&bytes)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
if meta.schema_version > super::SUPPORTED_SESSION_SCHEMA_VERSION {
let err = crate::error::Error::SchemaTooNew {
session_id: meta.session_id.clone(),
found: meta.schema_version,
supported: super::SUPPORTED_SESSION_SCHEMA_VERSION,
};
tracing::warn!("{}", err);
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
err.to_string(),
));
}
Ok(meta)
}
pub fn list_sessions(workspace: &Path) -> std::io::Result<Vec<PathBuf>> {
let base = match crate::paths::user_sessions_dir(workspace) {
Ok(d) => d,
Err(_) => workspace.join(".recursive").join("sessions"),
};
if !base.is_dir() {
return Ok(Vec::new());
}
let mut sessions = Vec::new();
for entry in std::fs::read_dir(&base)? {
let entry = entry?;
let slug_dir = entry.path();
if !slug_dir.is_dir() {
continue;
}
for session_entry in std::fs::read_dir(&slug_dir)? {
let session_entry = session_entry?;
let session_dir = session_entry.path();
if session_dir.is_dir() && session_dir.join(".meta.json").is_file() {
sessions.push(session_dir);
}
}
}
sessions.sort();
Ok(sessions)
}
pub fn list_sessions_sorted_by_updated_at(
workspace: &Path,
) -> std::io::Result<Vec<(PathBuf, SessionMeta)>> {
let dirs = Self::list_sessions(workspace)?;
let mut entries: Vec<(PathBuf, SessionMeta, std::time::SystemTime)> = Vec::new();
for dir in dirs {
let meta = match Self::load_meta(&dir) {
Ok(m) => m,
Err(_) => continue,
};
let mtime = std::fs::metadata(dir.join("transcript.jsonl"))
.and_then(|m| m.modified())
.unwrap_or(std::time::SystemTime::UNIX_EPOCH);
entries.push((dir, meta, mtime));
}
entries.sort_by(|a, b| {
b.1.updated_at
.cmp(&a.1.updated_at)
.then_with(|| b.2.cmp(&a.2))
.then_with(|| a.1.session_id.cmp(&b.1.session_id))
});
Ok(entries.into_iter().map(|(p, m, _)| (p, m)).collect())
}
pub fn list_all_sessions(base: &Path) -> std::io::Result<Vec<PathBuf>> {
let sessions_dir = base.join(".recursive").join("sessions");
if !sessions_dir.is_dir() {
return Ok(Vec::new());
}
let mut sessions = Vec::new();
for entry in std::fs::read_dir(&sessions_dir)? {
let entry = entry?;
let slug_dir = entry.path();
if !slug_dir.is_dir() {
continue;
}
for session_entry in std::fs::read_dir(&slug_dir)? {
let session_entry = session_entry?;
let session_dir = session_entry.path();
if session_dir.is_dir() && session_dir.join(".meta.json").is_file() {
sessions.push(session_dir);
}
}
}
sessions.sort();
Ok(sessions)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::llm::ToolSpec;
use crate::message::{Message, Role};
use crate::session::{hash_tool_specs, SessionStatus, SessionWriter};
#[test]
fn session_reader_round_trips() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let mut writer = SessionWriter::create(ws, "round trip", "gpt-4o", "openai").unwrap();
writer
.append(&Message::system("You are a bot."), None, None)
.unwrap();
writer
.append(&Message::user("do something"), None, None)
.unwrap();
writer
.append(&Message::assistant("I will do it."), None, None)
.unwrap();
let session_dir = writer.session_dir().to_path_buf();
writer.finish(SessionStatus::Completed).unwrap();
let entries = SessionReader::load_transcript(&session_dir).unwrap();
assert_eq!(entries.len(), 3);
assert_eq!(entries[0].role, "system");
assert_eq!(entries[1].role, "user");
assert_eq!(entries[2].role, "assistant");
}
#[test]
fn crash_partial_line_skipped() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let mut writer = SessionWriter::create(ws, "crash test", "gpt-4o", "openai").unwrap();
writer
.append(&Message::user("good line"), None, None)
.unwrap();
let session_dir = writer.session_dir().to_path_buf();
writer.finish(SessionStatus::Crashed).unwrap();
use std::io::Write;
let jsonl_path = session_dir.join("transcript.jsonl");
let mut f = std::fs::OpenOptions::new()
.append(true)
.open(&jsonl_path)
.unwrap();
writeln!(f, "this is not json").unwrap();
drop(f);
let entries = SessionReader::load_transcript(&session_dir).unwrap();
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].content, "good line");
}
#[test]
fn load_messages_drops_persistence_fields() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let mut writer = SessionWriter::create(ws, "g151 test", "model", "openai").unwrap();
let session_dir = writer.session_dir().to_path_buf();
writer
.append(&Message::user("hello".to_string()), None, None)
.unwrap();
writer
.append(&Message::assistant("hi back".to_string()), None, None)
.unwrap();
writer.finish(SessionStatus::Completed).unwrap();
drop(writer);
let msgs = SessionReader::load_messages(&session_dir).unwrap();
assert_eq!(msgs.len(), 2);
assert_eq!(msgs[0].role, Role::User);
assert_eq!(msgs[0].content, "hello");
assert_eq!(msgs[1].role, Role::Assistant);
assert_eq!(msgs[1].content, "hi back");
let entries = SessionReader::load_transcript(&session_dir).unwrap();
assert_eq!(entries.len(), 2);
assert_eq!(entries[0].id, "msg_001");
assert!(!entries[0].timestamp.is_empty());
}
#[test]
fn meta_round_trip_with_tool_registry_hash() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let specs = vec![ToolSpec {
name: "Read".into(),
description: "Read".into(),
parameters: serde_json::json!({"type":"object"}),
}];
let writer =
SessionWriter::create_with_tools(ws, "with hash", "model", "openai", &specs, None)
.unwrap();
let session_dir = writer.session_dir().to_path_buf();
drop(writer);
let meta = SessionReader::load_meta(&session_dir).unwrap();
let hash = meta
.tool_registry_hash
.as_ref()
.expect("expected hash to be Some(_)");
assert_eq!(*hash, hash_tool_specs(&specs));
}
#[test]
fn meta_round_trip_old_format_no_hash() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let session_dir = tmp
.path()
.join(".recursive")
.join("sessions")
.join("legacy");
std::fs::create_dir_all(&session_dir).unwrap();
let raw = r#"{
"session_id": "legacy-id",
"goal": "old goal",
"model": "model",
"provider": "openai",
"created_at": "2020-01-01T00:00:00Z",
"updated_at": "2020-01-01T00:00:00Z",
"message_count": 0,
"status": "active"
}"#;
std::fs::write(session_dir.join(".meta.json"), raw).unwrap();
std::fs::write(session_dir.join("transcript.jsonl"), "").unwrap();
let meta = SessionReader::load_meta(&session_dir).unwrap();
assert_eq!(meta.session_id, "legacy-id");
assert!(meta.tool_registry_hash.is_none());
}
#[test]
fn append_bumps_updated_at() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let mut writer = SessionWriter::create(ws, "bump", "model", "openai").unwrap();
let session_dir = writer.session_dir().to_path_buf();
let meta_before = SessionReader::load_meta(&session_dir).unwrap();
std::thread::sleep(std::time::Duration::from_millis(1100));
writer
.append(&Message::user("ping".to_string()), None, None)
.unwrap();
let meta_after = SessionReader::load_meta(&session_dir).unwrap();
assert_ne!(
meta_before.updated_at, meta_after.updated_at,
"expected updated_at to advance after append; before={} after={}",
meta_before.updated_at, meta_after.updated_at
);
}
#[test]
fn open_existing_continues_msg_numbering() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let mut writer = SessionWriter::create(ws, "resume-num", "model", "openai").unwrap();
let session_dir = writer.session_dir().to_path_buf();
writer
.append(&Message::user("u1".to_string()), None, None)
.unwrap();
writer
.append(&Message::assistant("a1".to_string()), None, None)
.unwrap();
writer
.append(&Message::user("u2".to_string()), None, None)
.unwrap();
drop(writer);
let mut writer2 = SessionWriter::open_existing(&session_dir).unwrap();
let id = writer2
.append(&Message::assistant("a2".to_string()), None, None)
.unwrap();
assert!(!id.is_empty(), "expected non-empty UUID from append");
drop(writer2);
let entries = SessionReader::load_transcript(&session_dir).unwrap();
assert_eq!(entries.len(), 4);
assert_eq!(entries[0].id, "msg_001");
assert_eq!(entries[1].id, "msg_002");
assert_eq!(entries[2].id, "msg_003");
assert_eq!(entries[3].id, "msg_004");
assert_eq!(entries[3].parent_id.as_deref(), Some("msg_003"));
}
#[test]
fn list_sessions_finds_sessions() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let sessions = SessionReader::list_sessions(ws).unwrap();
assert!(sessions.is_empty());
let writer = SessionWriter::create(ws, "session1", "gpt-4o", "openai").unwrap();
let dir1 = writer.session_dir().to_path_buf();
drop(writer);
let sessions = SessionReader::list_sessions(ws).unwrap();
assert_eq!(sessions.len(), 1);
assert_eq!(sessions[0], dir1);
}
#[test]
fn list_sessions_sorted_by_updated_at_returns_one_session_with_meta() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let w1 = SessionWriter::create(ws, "sort-test-goal", "gpt-4o", "openai").unwrap();
drop(w1);
let sorted = SessionReader::list_sessions_sorted_by_updated_at(ws).unwrap();
assert_eq!(sorted.len(), 1, "must find 1 session");
assert!(!sorted[0].1.session_id.is_empty());
}
#[test]
fn list_all_sessions_returns_empty_for_no_sessions() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let sessions = SessionReader::list_all_sessions(ws).unwrap();
assert!(sessions.is_empty(), "must be empty when no sessions exist");
}
#[test]
fn scan_orphan_tool_calls_returns_empty_for_no_assistant_with_tool_calls() {
use crate::message::Message;
use crate::session::{SessionStatus, SessionWriter};
let tmp = crate::test_util::IsolatedWorkspace::new();
let mut w = SessionWriter::create(tmp.path(), "g", "m", "p").unwrap();
w.append(&Message::user("hi".to_string()), None, None)
.unwrap();
w.append(&Message::assistant("hello".to_string()), None, None)
.unwrap();
w.finish(SessionStatus::Completed).unwrap();
let reg = crate::tools::ToolRegistry::local();
let orphans = SessionReader::scan_orphan_tool_calls(w.session_dir(), ®).unwrap();
assert!(
orphans.is_empty(),
"no assistant tool_calls → no orphans; got: {orphans:?}"
);
}
#[test]
fn scan_orphan_tool_calls_returns_empty_when_all_answered() {
use crate::message::Message;
use crate::session::{SessionStatus, SessionWriter};
let tmp = crate::test_util::IsolatedWorkspace::new();
let mut w = SessionWriter::create(tmp.path(), "g", "m", "p").unwrap();
let mut asst = Message::assistant("calling tool".to_string());
asst.tool_calls = vec![crate::llm::ToolCall {
id: "tc-001".to_string(),
name: "Read".to_string(),
arguments: serde_json::json!({"path": "foo.txt"}),
}];
let tool_result = Message {
role: crate::message::Role::Tool,
content: "file contents".to_string(),
tool_calls: vec![],
tool_call_id: Some("tc-001".to_string()),
reasoning_content: None,
is_compaction_summary: false,
};
w.append(&Message::user("do something".to_string()), None, None)
.unwrap();
w.append(&asst, None, None).unwrap();
w.append(&tool_result, None, None).unwrap();
w.finish(SessionStatus::Completed).unwrap();
let reg = crate::tools::ToolRegistry::local();
let orphans = SessionReader::scan_orphan_tool_calls(w.session_dir(), ®).unwrap();
assert!(
orphans.is_empty(),
"all tool calls answered → no orphans; got: {orphans:?}"
);
}
#[test]
fn scan_orphan_tool_calls_detects_unanswered_call() {
use crate::message::Message;
use crate::session::{SessionStatus, SessionWriter};
let tmp = crate::test_util::IsolatedWorkspace::new();
let mut w = SessionWriter::create(tmp.path(), "g", "m", "p").unwrap();
let mut asst = Message::assistant("calling tool".to_string());
asst.tool_calls = vec![crate::llm::ToolCall {
id: "tc-orphan".to_string(),
name: "Read".to_string(),
arguments: serde_json::json!({"path": "missing.txt"}),
}];
w.append(&Message::user("do something".to_string()), None, None)
.unwrap();
w.append(&asst, None, None).unwrap();
w.finish(SessionStatus::Completed).unwrap();
let reg = crate::tools::ToolRegistry::local();
let orphans = SessionReader::scan_orphan_tool_calls(w.session_dir(), ®).unwrap();
assert_eq!(
orphans.len(),
1,
"one unanswered tool call must be reported as orphan"
);
assert_eq!(orphans[0].tool_call_id, "tc-orphan");
assert_eq!(orphans[0].tool_name, "Read");
}
#[test]
fn load_full_history_includes_pre_compact_messages_and_boundaries() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let mut w = SessionWriter::create(ws, "full history", "model", "openai").unwrap();
let session_dir = w.session_dir().to_path_buf();
w.append(&Message::user("turn 1".to_string()), None, None)
.unwrap();
w.append(&Message::assistant("reply 1".to_string()), None, None)
.unwrap();
w.append(&Message::user("turn 2".to_string()), None, None)
.unwrap();
w.append(&Message::assistant("reply 2".to_string()), None, None)
.unwrap();
w.write_compact_boundary(3, 4, None).unwrap();
w.append(
&Message::system("summary of turn 1-2".to_string()),
None,
None,
)
.unwrap();
w.append(&Message::user("turn 3".to_string()), None, None)
.unwrap();
w.append(&Message::assistant("reply 3".to_string()), None, None)
.unwrap();
w.finish(SessionStatus::Completed).unwrap();
let loaded = SessionReader::load_full_history(&session_dir).unwrap();
assert_eq!(loaded.len(), 8, "all messages + the boundary marker");
let boundary_idx = loaded
.iter()
.position(
|e| matches!(e, LoadedEntry::CompactBoundary { removed, .. } if *removed == 4),
)
.expect("a CompactBoundary marker with removed=4");
assert_eq!(boundary_idx, 4, "boundary follows the 4 folded messages");
let pre = &loaded[..boundary_idx];
assert_eq!(pre.len(), 4);
assert!(pre.iter().all(|e| matches!(e, LoadedEntry::Message(_))));
let post = &loaded[boundary_idx + 1..];
assert_eq!(post.len(), 3);
assert!(post.iter().all(|e| matches!(e, LoadedEntry::Message(_))));
if let LoadedEntry::Message(m) = &post[0] {
assert_eq!(m.role, "system");
assert_eq!(m.content, "summary of turn 1-2");
} else {
panic!("expected a Message after the boundary");
}
let seed = SessionReader::load_messages(&session_dir).unwrap();
assert_eq!(
seed.len(),
3,
"load_messages keeps only post-compaction tail"
);
assert_eq!(seed[0].content, "summary of turn 1-2");
}
#[test]
fn load_full_history_no_boundary_returns_all_messages() {
let tmp = crate::test_util::IsolatedWorkspace::new();
let ws = tmp.path();
let mut w = SessionWriter::create(ws, "no compact", "model", "openai").unwrap();
let session_dir = w.session_dir().to_path_buf();
w.append(&Message::user("hi".to_string()), None, None)
.unwrap();
w.append(&Message::assistant("hello".to_string()), None, None)
.unwrap();
w.finish(SessionStatus::Completed).unwrap();
let loaded = SessionReader::load_full_history(&session_dir).unwrap();
assert_eq!(loaded.len(), 2);
assert!(
loaded.iter().all(|e| matches!(e, LoadedEntry::Message(_))),
"no CompactBoundary markers when the session never compacted"
);
}
}