tokensave 7.11.1

//! JSONL session parser for Claude Code transcripts.
//!
//! Reads `~/.claude/projects/**/*.jsonl`, extracts assistant turns with
//! model/usage/tool data, and inserts them into the `turns` table via
//! `GlobalDb`. Uses offset tracking for incremental re-parsing.

use std::fs;
use std::io::{BufRead, BufReader, Seek, SeekFrom};
use std::path::{Path, PathBuf};

use serde_json::Value;

use crate::accounting::classifier;
use crate::accounting::pricing;
use crate::context::read_modes::estimate_tokens;
use crate::global_db::GlobalDb;
use crate::types::CostTurn;

/// Whether a data source contributes complete, partial, or no coverage.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "lowercase")]
pub enum CoverageState {
    Complete,
    Partial,
    Absent,
}

/// Coverage summary for one accounting data source.
#[derive(Debug, Clone, serde::Serialize)]
pub struct SourceCoverage {
    pub agent: &'static str,
    pub state: CoverageState,
    pub sessions: u64,
}

/// Find all JSONL session files under `<home>/.claude/projects/`, sorted.
///
/// The boolean indicates whether discovery encountered an I/O error.
pub(crate) fn find_session_files(home: &Path) -> (Vec<PathBuf>, bool) {
    let projects_dir = home.join(".claude").join("projects");
    match fs::metadata(&projects_dir) {
        Ok(metadata) if metadata.is_dir() => {}
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            return (Vec::new(), false);
        }
        Ok(_) | Err(_) => return (Vec::new(), true),
    }

    let mut files = Vec::new();
    let had_errors = collect_jsonl_files(&projects_dir, &mut files, 0);
    files.sort();
    (files, had_errors)
}

/// Recursively collect .jsonl files, with a depth limit to avoid runaway traversal.
fn collect_jsonl_files(dir: &Path, out: &mut Vec<PathBuf>, depth: u8) -> bool {
    if depth > 5 {
        return false;
    }
    let Ok(entries) = fs::read_dir(dir) else {
        return true;
    };
    let mut had_errors = false;
    for entry in entries {
        let Ok(entry) = entry else {
            had_errors = true;
            continue;
        };
        let path = entry.path();
        if path.is_dir() {
            had_errors |= collect_jsonl_files(&path, out, depth + 1);
        } else if path.extension().and_then(|e| e.to_str()) == Some("jsonl") {
            out.push(path);
        }
    }
    had_errors
}

/// Extract project hash and session ID from a JSONL file path.
/// Path pattern: `~/.claude/projects/<project-hash>/<session-id>.jsonl`
/// or `~/.claude/projects/<project-hash>/<session-id>/subagents/<agent>.jsonl`
fn extract_path_parts(path: &Path) -> (String, String) {
    let components: Vec<&str> = path
        .components()
        .filter_map(|c| c.as_os_str().to_str())
        .collect();

    // Find "projects" in the path and take the next component as project_hash
    let projects_idx = components.iter().position(|c| *c == "projects");
    let project_hash = projects_idx
        .and_then(|i| components.get(i + 1))
        .unwrap_or(&"unknown")
        .to_string();

    let session_id = path
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("unknown")
        .to_string();

    (project_hash, session_id)
}

/// Parse a single JSONL line into a `CostTurn`, if it's an assistant message
/// with usage data.
fn parse_line(line: &str, project_hash: &str, session_id: &str) -> Option<(CostTurn, Vec<String>)> {
    let v: Value = serde_json::from_str(line).ok()?;

    // Only process assistant messages
    if v.get("type")?.as_str()? != "assistant" {
        return None;
    }

    let msg = v.get("message")?;
    let message_id = msg.get("id")?.as_str()?;
    let model = msg.get("model")?.as_str()?;

    let usage = msg.get("usage")?;
    let input_tokens = usage.get("input_tokens")?.as_u64().unwrap_or(0);
    let output_tokens = usage.get("output_tokens")?.as_u64().unwrap_or(0);
    let cache_write_tokens = usage
        .get("cache_creation_input_tokens")
        .and_then(serde_json::Value::as_u64)
        .unwrap_or(0);
    let cache_read_tokens = usage
        .get("cache_read_input_tokens")
        .and_then(serde_json::Value::as_u64)
        .unwrap_or(0);

    // Parse timestamp from the outer object (ISO 8601)
    let timestamp = parse_timestamp(v.get("timestamp")?.as_str()?)?;

    // Extract tool names and bash commands for classification
    let content = msg.get("content").and_then(|c| c.as_array());
    let mut tool_names_vec: Vec<String> = Vec::new();
    let mut bash_commands: Vec<String> = Vec::new();
    let mut tool_use_ids: Vec<String> = Vec::new();

    if let Some(blocks) = content {
        for block in blocks {
            if block.get("type").and_then(|t| t.as_str()) == Some("tool_use") {
                if let Some(id) = block.get("id").and_then(|i| i.as_str()) {
                    tool_use_ids.push(id.to_string());
                }
                if let Some(name) = block.get("name").and_then(|n| n.as_str()) {
                    tool_names_vec.push(name.to_string());
                    if name == "Bash" {
                        if let Some(cmd) = block
                            .get("input")
                            .and_then(|i| i.get("command"))
                            .and_then(|c| c.as_str())
                        {
                            bash_commands.push(cmd.to_string());
                        }
                    }
                }
            }
        }
    }

    // Classify
    let tool_refs: Vec<&str> = tool_names_vec
        .iter()
        .map(std::string::String::as_str)
        .collect();
    let bash_refs: Vec<&str> = bash_commands
        .iter()
        .map(std::string::String::as_str)
        .collect();
    let category = classifier::classify(&tool_refs, &bash_refs);

    // Compute cost
    let cost_usd = pricing::cost_of_turn(
        model,
        input_tokens,
        output_tokens,
        cache_write_tokens,
        cache_read_tokens,
    );

    let turn = CostTurn {
        message_id: message_id.to_string(),
        project_hash: project_hash.to_string(),
        session_id: session_id.to_string(),
        model: model.to_string(),
        timestamp,
        input_tokens,
        output_tokens,
        cache_write_tokens,
        cache_read_tokens,
        cost_usd,
        category: category.as_str().to_string(),
        tool_names: tool_names_vec.join(","),
        agent: "claude".to_string(),
        credits: None,
    };
    Some((turn, tool_use_ids))
}

/// The tool results carried by one transcript line, as
/// `(tool_use_id, estimated tokens)` pairs.
///
/// A result does not travel with the `tool_use` that asked for it: it arrives
/// in the *following* user message, so the caller keeps the id-to-turn mapping
/// and attributes the size back (#474). Only user messages carry results;
/// every other line yields nothing.
///
/// The size is measured on the result text — the bytes a `Read` or a `Grep`
/// actually put into the conversation, which is the quantity a graph query
/// could have served more cheaply. Content arrives either as a plain string or
/// as a list of blocks, and both forms are measured.
fn parse_tool_results(line: &str) -> Vec<(String, u64)> {
    let Ok(v) = serde_json::from_str::<Value>(line) else {
        return Vec::new();
    };
    if v.get("type").and_then(|t| t.as_str()) != Some("user") {
        return Vec::new();
    }
    let Some(blocks) = v
        .get("message")
        .and_then(|m| m.get("content"))
        .and_then(|c| c.as_array())
    else {
        return Vec::new();
    };

    let mut out = Vec::new();
    for block in blocks {
        if block.get("type").and_then(|t| t.as_str()) != Some("tool_result") {
            continue;
        }
        let Some(id) = block.get("tool_use_id").and_then(|i| i.as_str()) else {
            continue;
        };
        let tokens = block.get("content").map_or(0, measure_result_content);
        if tokens > 0 {
            out.push((id.to_string(), tokens));
        }
    }
    out
}

/// Estimated tokens of one `tool_result` block's content.
fn measure_result_content(content: &Value) -> u64 {
    match content {
        Value::String(text) => u64::from(estimate_tokens(text)),
        Value::Array(blocks) => blocks
            .iter()
            .map(|block| {
                block
                    .get("text")
                    .and_then(|t| t.as_str())
                    .map_or(0, |text| u64::from(estimate_tokens(text)))
            })
            .sum(),
        _ => 0,
    }
}

/// Parse an ISO 8601 timestamp to unix epoch seconds.
pub(crate) fn parse_timestamp(ts: &str) -> Option<u64> {
    // Handle "2026-04-14T10:32:15.039Z" format
    // Simple parsing without pulling in chrono: split on known positions
    if ts.len() < 19 {
        return None;
    }
    let year: i64 = ts.get(0..4)?.parse().ok()?;
    let month: u64 = ts.get(5..7)?.parse().ok()?;
    let day: u64 = ts.get(8..10)?.parse().ok()?;
    let hour: u64 = ts.get(11..13)?.parse().ok()?;
    let min: u64 = ts.get(14..16)?.parse().ok()?;
    let sec: u64 = ts.get(17..19)?.parse().ok()?;

    let bytes = ts.as_bytes();
    if bytes.get(4) != Some(&b'-')
        || bytes.get(7) != Some(&b'-')
        || bytes.get(10) != Some(&b'T')
        || bytes.get(13) != Some(&b':')
        || bytes.get(16) != Some(&b':')
        || year < 1970
        || !(1..=12).contains(&month)
        || hour >= 24
        || min >= 60
        || sec >= 60
    {
        return None;
    }

    let month_days = [0u64, 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
    let max_day = month_days[month as usize] + u64::from(month == 2 && is_leap(year));
    if day == 0 || day > max_day {
        return None;
    }

    // Days from epoch using a simple formula (good enough for 2000-2100)
    let mut days: i64 = 0;
    for y in 1970..year {
        days += if is_leap(y) { 366 } else { 365 };
    }
    for m in 1..month {
        days += month_days[m as usize] as i64;
    }
    if month > 2 && is_leap(year) {
        days += 1;
    }
    days += (day as i64) - 1;

    Some((days as u64) * 86400 + hour * 3600 + min * 60 + sec)
}

fn is_leap(y: i64) -> bool {
    y % 4 == 0 && (y % 100 != 0 || y % 400 == 0)
}

/// Stats returned by the `ingest` function.
pub struct IngestStats {
    /// Number of new Claude turns inserted (receipt-only; does not count historical Droid scans).
    pub turns_inserted: u64,
    /// Number of Droid turns inserted or updated.
    pub turns_updated: u64,
    /// Total cost of newly-inserted Claude turns.
    pub cost_usd: f64,
    /// Total input + output tokens of newly-inserted Claude turns.
    pub tokens_consumed: u64,
    /// Per-source coverage information.
    pub coverage: Vec<SourceCoverage>,
}

impl IngestStats {
    /// Returns an empty stats value with no coverage information.
    pub fn empty() -> Self {
        Self {
            turns_inserted: 0,
            turns_updated: 0,
            cost_usd: 0.0,
            tokens_consumed: 0,
            coverage: Vec::new(),
        }
    }
}

/// Ingest only Claude Code session files from `home`, skipping any Droid scan.
///
/// Shared by the Claude Stop hook (via [`ingest_claude_only`]) and the full
/// coordinated ingest (via [`ingest_from_home`]).  Keeping this as a private
/// helper avoids duplicating the offset-tracking loop.
async fn ingest_claude_from_home(gdb: &GlobalDb, home: &Path) -> IngestStats {
    let (files, mut had_errors) = find_session_files(home);
    let claude_sessions = files.len() as u64;

    let mut total_inserted = 0u64;
    let mut total_cost = 0.0f64;
    let mut total_tokens = 0u64;

    for file_path in &files {
        let path_str = file_path.to_string_lossy().to_string();

        let Ok(meta) = fs::metadata(file_path) else {
            had_errors = true;
            continue;
        };
        let mtime = meta
            .modified()
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map_or(0, |d| d.as_secs());

        let (prev_offset, prev_mtime) = gdb.get_parse_offset(&path_str).await.unwrap_or((0, 0));

        if mtime == prev_mtime && prev_offset > 0 {
            continue;
        }

        let seek_to = if mtime == prev_mtime || (prev_mtime > 0 && mtime > prev_mtime) {
            prev_offset
        } else {
            0
        };

        let (project_hash, session_id) = extract_path_parts(file_path);

        let Ok(f) = fs::File::open(file_path) else {
            had_errors = true;
            continue;
        };
        let mut reader = BufReader::new(f);

        if seek_to > 0 && reader.seek(SeekFrom::Start(seek_to)).is_err() {
            had_errors = true;
            continue;
        }

        // `tool_use_id` -> the message that issued it, so a result found in a
        // later line can be attributed back (#474). Per session file, and
        // entries are removed as they are consumed, so it holds only the calls
        // still awaiting a result.
        let mut issuing_turn: std::collections::HashMap<String, String> =
            std::collections::HashMap::new();

        let mut line = String::new();
        let mut current_offset = seek_to;
        let mut read_failed = false;

        loop {
            line.clear();
            match reader.read_line(&mut line) {
                Ok(0) => break,
                Err(_) => {
                    had_errors = true;
                    read_failed = true;
                    break;
                }
                Ok(n) => {
                    current_offset += n as u64;
                    let trimmed = line.trim();
                    if trimmed.is_empty() {
                        continue;
                    }
                    if let Some((turn, tool_use_ids)) =
                        parse_line(trimmed, &project_hash, &session_id)
                    {
                        let turn_cost = turn.cost_usd;
                        let turn_tokens = turn.input_tokens + turn.output_tokens;
                        // Remember which turn issued each tool call, so the
                        // result blocks in a later user message can be sized
                        // and added back to it (#474).
                        for id in tool_use_ids {
                            issuing_turn.insert(id, turn.message_id.clone());
                        }
                        if gdb.insert_turn(&turn).await {
                            total_inserted += 1;
                            total_cost += turn_cost;
                            total_tokens += turn_tokens;
                        }
                    } else {
                        for (tool_use_id, tokens) in parse_tool_results(trimmed) {
                            // An id with no remembered turn is what a resumed
                            // parse sees when the assistant line fell before
                            // the stored offset; there is nothing to add it to.
                            if let Some(message_id) = issuing_turn.remove(&tool_use_id) {
                                gdb.add_tool_result_tokens(&message_id, tokens).await;
                            }
                        }
                    }
                }
            }
        }

        if !read_failed {
            gdb.set_parse_offset(&path_str, current_offset, mtime).await;
        }
    }

    let claude_state = if had_errors {
        CoverageState::Partial
    } else if files.is_empty() {
        CoverageState::Absent
    } else {
        CoverageState::Complete
    };

    IngestStats {
        turns_inserted: total_inserted,
        turns_updated: 0,
        cost_usd: total_cost,
        tokens_consumed: total_tokens,
        coverage: vec![SourceCoverage {
            agent: "claude",
            state: claude_state,
            sessions: claude_sessions,
        }],
    }
}

/// Ingest Claude sessions from `home` and Droid sessions from
/// `home/.factory/sessions`.  Returns combined [`IngestStats`].
pub(crate) async fn ingest_from_home(gdb: &GlobalDb, home: &Path) -> IngestStats {
    let mut stats = ingest_claude_from_home(gdb, home).await;

    // Droid ingestion from ~/.factory/sessions
    let droid_root = home.join(".factory").join("sessions");
    let droid = crate::accounting::droid::ingest_dir(gdb, &droid_root).await;

    stats.turns_updated = droid.rows_changed;
    stats.coverage.push(droid.coverage);

    stats
}

/// Ingest only Claude Code session files, skipping any Droid scan.
///
/// Use this from the Claude Stop hook to keep the hot path free of Droid
/// filesystem and JSON I/O.  The general [`ingest`] function performs the full
/// Claude + Droid scan and is reserved for cost/monitor/status callers.
pub async fn ingest_claude_only(gdb: &GlobalDb) -> IngestStats {
    let Some(home) = crate::agents::home_dir() else {
        return IngestStats::empty();
    };
    ingest_claude_from_home(gdb, &home).await
}

/// Ingest all Claude Code session files into the global DB.
/// Uses offset tracking to only parse new lines since the last run.
pub async fn ingest(gdb: &GlobalDb) -> IngestStats {
    let Some(home) = crate::agents::home_dir() else {
        return IngestStats::empty();
    };
    ingest_from_home(gdb, &home).await
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    async fn open_test_db(tmp: &TempDir) -> GlobalDb {
        let db_path = tmp.path().join(".tokensave").join("global.db");
        std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
        GlobalDb::open_at(&db_path).await.unwrap()
    }

    /// `ingest_claude_from_home` must not populate Droid coverage or `turns_updated`.
    #[tokio::test]
    async fn claude_only_ingest_has_no_droid_coverage() {
        let home_tmp = TempDir::new().unwrap();
        let db_tmp = TempDir::new().unwrap();
        let gdb = open_test_db(&db_tmp).await;

        let stats = ingest_claude_from_home(&gdb, home_tmp.path()).await;

        assert_eq!(
            stats.turns_updated, 0,
            "claude-only must not update Droid rows"
        );
        assert!(
            !stats.coverage.iter().any(|c| c.agent == "droid"),
            "claude-only must not include droid coverage"
        );
        assert_eq!(
            stats.coverage.len(),
            1,
            "exactly one coverage entry (claude)"
        );
        assert_eq!(stats.coverage[0].agent, "claude");
    }

    #[test]
    fn test_parse_timestamp() {
        // 2026-01-01T00:00:00Z
        let ts = parse_timestamp("2026-01-01T00:00:00.000Z");
        assert!(ts.is_some());
        let epoch = ts.unwrap();
        // 2026-01-01 = 56 years from 1970, roughly 20454 days
        assert!(epoch > 1_700_000_000);
        assert!(epoch < 1_800_000_000);
    }

    #[test]
    fn invalid_timestamp_components_return_none() {
        assert_eq!(parse_timestamp("2026-14-01T00:00:00Z"), None);
        assert_eq!(parse_timestamp("2026-02-30T00:00:00Z"), None);
        assert_eq!(parse_timestamp("2026-01-01T24:00:00Z"), None);
        assert_eq!(parse_timestamp("2026-01-01T00:60:00Z"), None);
        assert_eq!(parse_timestamp("1969-12-31T23:59:59Z"), None);
        assert_eq!(parse_timestamp("2026/01/01 00:00:00Z"), None);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn claude_file_read_error_marks_coverage_partial() {
        let home_tmp = TempDir::new().unwrap();
        let db_tmp = TempDir::new().unwrap();
        let projects = home_tmp.path().join(".claude/projects/project");
        std::fs::create_dir_all(&projects).unwrap();
        std::os::unix::fs::symlink(
            projects.join("missing-target"),
            projects.join("broken.jsonl"),
        )
        .unwrap();
        let gdb = open_test_db(&db_tmp).await;

        let stats = ingest_claude_from_home(&gdb, home_tmp.path()).await;

        assert_eq!(stats.coverage[0].state, CoverageState::Partial);
        assert_eq!(stats.coverage[0].sessions, 1);
    }

    #[test]
    fn test_parse_timestamp_invalid() {
        assert!(parse_timestamp("bad").is_none());
        assert!(parse_timestamp("").is_none());
    }

    #[test]
    fn test_extract_path_parts() {
        let path =
            PathBuf::from("/Users/test/.claude/projects/-Users-test-Code/abc123-session.jsonl");
        let (project, session) = extract_path_parts(&path);
        assert_eq!(project, "-Users-test-Code");
        assert_eq!(session, "abc123-session");
    }

    #[test]
    fn test_parse_line_assistant() {
        let line = r#"{"type":"assistant","message":{"id":"msg_01abc","model":"claude-opus-4-6","role":"assistant","usage":{"input_tokens":1000,"output_tokens":200,"cache_creation_input_tokens":500,"cache_read_input_tokens":800},"content":[{"type":"tool_use","name":"Edit","input":{"file_path":"test.rs"}}]},"timestamp":"2026-04-14T10:00:00.000Z"}"#;
        let turn = parse_line(line, "proj", "sess");
        assert!(turn.is_some());
        let (t, _tool_use_ids) = turn.unwrap();
        assert_eq!(t.message_id, "msg_01abc");
        assert_eq!(t.model, "claude-opus-4-6");
        assert_eq!(t.input_tokens, 1000);
        assert_eq!(t.output_tokens, 200);
        assert_eq!(t.cache_write_tokens, 500);
        assert_eq!(t.cache_read_tokens, 800);
        assert_eq!(t.category, "coding");
        assert_eq!(t.tool_names, "Edit");
        assert!(t.cost_usd > 0.0);
    }

    #[test]
    fn test_parse_line_user_skipped() {
        let line = r#"{"type":"user","message":{"content":"hello"},"timestamp":"2026-04-14T10:00:00.000Z"}"#;
        assert!(parse_line(line, "proj", "sess").is_none());
    }

    #[test]
    fn test_parse_line_malformed() {
        assert!(parse_line("not json at all", "proj", "sess").is_none());
        assert!(parse_line("{}", "proj", "sess").is_none());
    }

    /// The ids a turn issued come back with it, since the results that will
    /// be attributed to it are identified by nothing else (#474).
    #[test]
    fn assistant_turn_reports_the_tool_calls_it_issued() {
        let line = r#"{"type":"assistant","message":{"id":"msg_01abc","model":"claude-opus-4-6","role":"assistant","usage":{"input_tokens":3,"output_tokens":200},"content":[{"type":"tool_use","id":"toolu_1","name":"Read","input":{}},{"type":"tool_use","id":"toolu_2","name":"Grep","input":{}}]},"timestamp":"2026-04-14T10:00:00.000Z"}"#;
        let (turn, ids) = parse_line(line, "proj", "sess").unwrap();
        assert_eq!(turn.tool_names, "Read,Grep");
        assert_eq!(ids, ["toolu_1", "toolu_2"]);
    }

    /// A result block is sized by its text, whether the content is a plain
    /// string or a list of blocks.
    #[test]
    fn tool_results_are_measured_by_their_text() {
        let string_form = r#"{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_1","content":"0123456789012345678901234567890123456789"}]}}"#;
        let results = parse_tool_results(string_form);
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].0, "toolu_1");
        assert_eq!(results[0].1, 10, "40 characters at ~4 per token");

        let block_form = r#"{"type":"user","message":{"content":[{"type":"tool_result","tool_use_id":"toolu_2","content":[{"type":"text","text":"01234567"},{"type":"text","text":"89AB"}]}]}}"#;
        let results = parse_tool_results(block_form);
        assert_eq!(results, [("toolu_2".to_string(), 3)]);
    }

    /// Everything that is not a tool result contributes nothing — an ordinary
    /// user message most of all, which is text the user typed rather than
    /// anything a tool injected.
    #[test]
    fn only_tool_results_are_measured() {
        let typed = r#"{"type":"user","message":{"content":"a long message the user typed out"}}"#;
        assert!(parse_tool_results(typed).is_empty());

        let assistant =
            r#"{"type":"assistant","message":{"content":[{"type":"text","text":"hello"}]}}"#;
        assert!(parse_tool_results(assistant).is_empty());

        assert!(parse_tool_results("not json").is_empty());
        assert!(parse_tool_results("{}").is_empty());
    }
}