use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use std::collections::{HashMap, HashSet};
use std::fs::File;
use std::io::{BufRead, BufReader};
use std::path::Path;
use super::models::{DataQuality, GlobalDataQuality, SessionData};
use super::parser::parse_session_file;
use super::scanner::{resolve_agent_parents, scan_claude_home, scan_projects_dir};
fn extract_version(path: &Path) -> Option<String> {
let file = File::open(path).ok()?;
let reader = BufReader::new(file);
let first_line = reader.lines().next()?.ok()?;
let val: serde_json::Value = serde_json::from_str(&first_line).ok()?;
val.get("version")
.and_then(|v| v.as_str())
.map(|s| s.to_string())
}
fn time_range<'a, I>(timestamps: I) -> (Option<DateTime<Utc>>, Option<DateTime<Utc>>)
where
I: Iterator<Item = &'a DateTime<Utc>>,
{
let mut min: Option<DateTime<Utc>> = None;
let mut max: Option<DateTime<Utc>> = None;
for ts in timestamps {
min = Some(min.map_or(*ts, |m: DateTime<Utc>| m.min(*ts)));
max = Some(max.map_or(*ts, |m: DateTime<Utc>| m.max(*ts)));
}
(min, max)
}
fn request_id_set(turns: &[super::models::ValidatedTurn]) -> HashSet<String> {
turns
.iter()
.filter_map(|t| t.request_id.as_ref())
.cloned()
.collect()
}
fn merge_agent_turns(parent: &mut SessionData, agent_turns: Vec<super::models::ValidatedTurn>, quality: &DataQuality) {
let existing_rids = request_id_set(&parent.turns);
let before = parent.agent_turns.len();
for turn in agent_turns {
let dominated = turn
.request_id
.as_ref()
.is_some_and(|rid| existing_rids.contains(rid));
if !dominated {
parent.agent_turns.push(turn);
}
}
let added = parent.agent_turns.len() - before;
let deduped = quality.valid_turns.saturating_sub(added);
parent.quality.total_lines += quality.total_lines;
parent.quality.valid_turns += added;
parent.quality.skipped_synthetic += quality.skipped_synthetic;
parent.quality.skipped_sidechain += quality.skipped_sidechain;
parent.quality.skipped_invalid += quality.skipped_invalid;
parent.quality.skipped_parse_error += quality.skipped_parse_error;
parent.quality.duplicate_turns += quality.duplicate_turns + deduped;
}
pub fn load_all(claude_home: &Path) -> Result<(Vec<SessionData>, GlobalDataQuality)> {
let mut files = scan_claude_home(claude_home)
.context("failed to scan claude home for session files")?;
resolve_agent_parents(&mut files)
.context("failed to resolve agent parent sessions")?;
load_from_files(files)
}
pub fn load_from_projects_dir(projects_dir: &Path) -> Result<(Vec<SessionData>, GlobalDataQuality)> {
let mut files = scan_projects_dir(projects_dir)
.context("failed to scan projects dir for session files")?;
resolve_agent_parents(&mut files)
.context("failed to resolve agent parent sessions")?;
load_from_files(files)
}
fn load_from_files(files: Vec<super::models::SessionFile>) -> Result<(Vec<SessionData>, GlobalDataQuality)> {
let (main_files, agent_files): (Vec<_>, Vec<_>) =
files.into_iter().partition(|f| !f.is_agent);
let mut global_quality = GlobalDataQuality {
total_session_files: main_files.len(),
total_agent_files: agent_files.len(),
..Default::default()
};
let mut sessions: HashMap<String, SessionData> = HashMap::new();
for sf in &main_files {
let (turns, quality) = parse_session_file(&sf.file_path, false)
.with_context(|| format!("failed to parse session: {}", sf.file_path.display()))?;
let version = extract_version(&sf.file_path);
let (first_ts, last_ts) = time_range(turns.iter().map(|t| &t.timestamp));
global_quality.total_valid_turns += quality.valid_turns;
global_quality.total_skipped +=
quality.skipped_synthetic + quality.skipped_sidechain + quality.skipped_invalid + quality.skipped_parse_error;
sessions.insert(sf.session_id.clone(), SessionData {
session_id: sf.session_id.clone(),
project: sf.project.clone(),
turns,
agent_turns: Vec::new(),
first_timestamp: first_ts,
last_timestamp: last_ts,
version,
quality,
});
}
for sf in &agent_files {
let (agent_turns, quality) = parse_session_file(&sf.file_path, true)
.with_context(|| format!("failed to parse agent file: {}", sf.file_path.display()))?;
global_quality.total_valid_turns += quality.valid_turns;
global_quality.total_skipped +=
quality.skipped_synthetic + quality.skipped_sidechain + quality.skipped_invalid + quality.skipped_parse_error;
let target_id = sf.parent_session_id.clone().unwrap_or_else(|| sf.session_id.clone());
if !sessions.contains_key(&target_id) {
let project = sf.project.clone().or_else(|| Some("(orphan)".to_string()));
sessions.insert(target_id.clone(), SessionData {
session_id: target_id.clone(),
project,
turns: Vec::new(),
agent_turns: Vec::new(),
first_timestamp: None,
last_timestamp: None,
version: None,
quality: DataQuality::default(),
});
global_quality.orphan_agents += 1;
}
let parent = sessions.get_mut(&target_id).unwrap();
merge_agent_turns(parent, agent_turns, &quality);
}
let mut result: Vec<SessionData> = sessions.into_values().collect();
let mut global_min: Option<DateTime<Utc>> = None;
let mut global_max: Option<DateTime<Utc>> = None;
for session in &mut result {
let all_timestamps = session.all_responses();
let (first_ts, last_ts) = time_range(all_timestamps.iter().map(|t| &t.timestamp));
session.first_timestamp = first_ts;
session.last_timestamp = last_ts;
if let Some(ts) = first_ts {
global_min = Some(global_min.map_or(ts, |m: DateTime<Utc>| m.min(ts)));
}
if let Some(ts) = last_ts {
global_max = Some(global_max.map_or(ts, |m: DateTime<Utc>| m.max(ts)));
}
}
global_quality.time_range = match (global_min, global_max) {
(Some(min), Some(max)) => Some((min, max)),
_ => None,
};
Ok((result, global_quality))
}