mi6_core/input/codex_session/

scanner.rs

//! Codex session file scanner with storage integration.
//!
//! Provides high-level scanning functionality that orchestrates parsing,
//! deduplication, and storage operations.

use std::path::{Path, PathBuf};

use crate::model::Storage;
use crate::model::error::{ScanError, TranscriptError};

use super::{CodexSessionParser, FilePosition, extract_session_id_from_filename};

/// Result of scanning a Codex session file.
#[derive(Debug, Clone, Default)]
pub struct ScanResult {
    /// Number of events inserted
    pub inserted: usize,
    /// Number of lines that failed to parse
    pub parse_errors: u64,
    /// Session ID from the file
    pub session_id: Option<String>,
}

/// Orchestrates Codex session file scanning with storage integration.
///
/// `CodexSessionScanner` provides a high-level API for scanning Codex CLI
/// session files and inserting events into storage. It handles:
/// - Incremental parsing (only reads new content since last scan)
/// - Deduplication by timestamp+type key
/// - Position tracking for future scans
pub struct CodexSessionScanner<'a, S: Storage> {
    storage: &'a S,
    machine_id: String,
}

impl<'a, S: Storage> CodexSessionScanner<'a, S> {
    /// Create a new scanner with the given storage and machine ID.
    pub fn new(storage: &'a S, machine_id: impl Into<String>) -> Self {
        Self {
            storage,
            machine_id: machine_id.into(),
        }
    }

    /// Scan a Codex session file and insert events into storage.
    ///
    /// This method:
    /// 1. Extracts session ID from filename
    /// 2. Loads the last scanned position for the file
    /// 3. Parses new entries since that position
    /// 4. Inserts new events
    /// 5. Updates the position for future incremental scans
    pub fn scan_file(&self, path: &Path) -> Result<ScanResult, ScanError> {
        // Extract session ID from filename
        let filename = path.file_name().and_then(|n| n.to_str()).ok_or_else(|| {
            ScanError::Parse(TranscriptError::Io(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "Invalid filename",
            )))
        })?;

        let session_id = extract_session_id_from_filename(filename).ok_or_else(|| {
            ScanError::Parse(TranscriptError::Io(std::io::Error::new(
                std::io::ErrorKind::InvalidInput,
                "Could not extract session ID from filename",
            )))
        })?;

        // Get last scanned position (reuse transcript position tracking)
        let start_position = self
            .storage
            .get_transcript_position(path)?
            .map(|pos| FilePosition {
                offset: pos.offset,
                line_number: pos.line_number,
                last_key: pos.last_uuid,
            })
            .unwrap_or_default();

        // Parse the session file
        let parser = CodexSessionParser::new(&self.machine_id);
        let result = parser.parse_incremental(path, &session_id, &start_position)?;

        let parse_errors = result.parse_errors;

        // Insert new events
        let mut inserted = 0;
        for event in result.events {
            self.storage.insert(&event)?;
            inserted += 1;
        }

        // Update the session's transcript_path with the file path.
        // This is needed because Codex hooks don't provide transcript_path,
        // but we know it from scanning the session file.
        if let Some(path_str) = path.to_str() {
            // Ignore errors - the session may not exist yet if no events were inserted
            let _ = self.storage.update_session_transcript_path(
                &self.machine_id,
                &session_id,
                path_str,
            );
        }

        // Update position for next scan (reuse transcript position)
        let transcript_position = crate::input::transcript::FilePosition {
            offset: result.position.offset,
            line_number: result.position.line_number,
            last_uuid: result.position.last_key,
        };
        self.storage
            .set_transcript_position(path, &transcript_position)?;

        Ok(ScanResult {
            inserted,
            parse_errors,
            session_id: Some(session_id),
        })
    }

    /// Scan a session file by session ID.
    ///
    /// Looks up the session file path in the standard Codex sessions directory.
    /// After scanning, also updates the transcript_path for the provided session_id,
    /// which may differ from the UUID extracted from the filename. This ensures
    /// sessions created by hooks (using thread-id) get their transcript_path set.
    pub fn scan_session(&self, session_id: &str) -> Result<ScanResult, ScanError> {
        let path = find_session_file(session_id)?;
        let result = self.scan_file(&path)?;

        // Also update transcript_path for the original session_id.
        // This is important because the hook may use a different session_id
        // (thread-id) than the UUID in the filename. We need to update both
        // so the session created by hooks gets the transcript path.
        if let Some(path_str) = path.to_str() {
            let _ =
                self.storage
                    .update_session_transcript_path(&self.machine_id, session_id, path_str);
        }

        Ok(result)
    }
}

/// Find a Codex session file by session ID.
///
/// Searches in `~/.codex/sessions/` for a file matching the session UUID.
pub fn find_session_file(session_id: &str) -> Result<PathBuf, ScanError> {
    let sessions_dir = dirs::home_dir()
        .ok_or_else(|| {
            ScanError::Parse(TranscriptError::Io(std::io::Error::new(
                std::io::ErrorKind::NotFound,
                "Could not determine home directory",
            )))
        })?
        .join(".codex")
        .join("sessions");

    if !sessions_dir.exists() {
        return Err(ScanError::Parse(TranscriptError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            "Codex sessions directory not found",
        ))));
    }

    // Search through date directories
    // Format: ~/.codex/sessions/YYYY/MM/DD/rollout-...-<UUID>.jsonl
    let year_entries =
        std::fs::read_dir(&sessions_dir).map_err(|e| ScanError::Parse(TranscriptError::Io(e)))?;

    for year_entry in year_entries {
        let year_path = year_entry
            .map_err(|e| ScanError::Parse(TranscriptError::Io(e)))?
            .path();
        if !year_path.is_dir() {
            continue;
        }

        let month_entries =
            std::fs::read_dir(&year_path).map_err(|e| ScanError::Parse(TranscriptError::Io(e)))?;

        for month_entry in month_entries {
            let month_path = month_entry
                .map_err(|e| ScanError::Parse(TranscriptError::Io(e)))?
                .path();
            if !month_path.is_dir() {
                continue;
            }

            let day_entries = std::fs::read_dir(&month_path)
                .map_err(|e| ScanError::Parse(TranscriptError::Io(e)))?;

            for day_entry in day_entries {
                let day_path = day_entry
                    .map_err(|e| ScanError::Parse(TranscriptError::Io(e)))?
                    .path();
                if !day_path.is_dir() {
                    continue;
                }

                let file_entries = std::fs::read_dir(&day_path)
                    .map_err(|e| ScanError::Parse(TranscriptError::Io(e)))?;

                for file_entry in file_entries {
                    let file_path = file_entry
                        .map_err(|e| ScanError::Parse(TranscriptError::Io(e)))?
                        .path();
                    if let Some(filename) = file_path.file_name().and_then(|n| n.to_str())
                        && filename.ends_with(".jsonl")
                        && filename.contains(session_id)
                    {
                        return Ok(file_path);
                    }
                }
            }
        }
    }

    Err(ScanError::Parse(TranscriptError::Io(std::io::Error::new(
        std::io::ErrorKind::NotFound,
        format!("Session file not found for ID: {}", session_id),
    ))))
}

/// Get the Codex sessions directory path.
pub fn codex_sessions_dir() -> Option<PathBuf> {
    dirs::home_dir().map(|h| h.join(".codex").join("sessions"))
}

/// List all session files in a date directory.
///
/// Returns (session_id, path) tuples for all session files found.
pub fn list_session_files_in_dir(dir: &Path) -> Vec<(String, PathBuf)> {
    let mut results = Vec::new();

    if let Ok(entries) = std::fs::read_dir(dir) {
        for entry in entries.flatten() {
            let path = entry.path();
            if path.is_file()
                && let Some(filename) = path.file_name().and_then(|n| n.to_str())
                && filename.ends_with(".jsonl")
                && let Some(session_id) = extract_session_id_from_filename(filename)
            {
                results.push((session_id, path));
            }
        }
    }

    results
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_list_session_files_in_dir_empty() {
        let temp_dir = tempfile::tempdir().unwrap();
        let results = list_session_files_in_dir(temp_dir.path());
        assert!(results.is_empty());
    }

    #[test]
    fn test_list_session_files_in_dir_with_files() {
        let temp_dir = tempfile::tempdir().unwrap();

        // Create test files
        std::fs::write(
            temp_dir
                .path()
                .join("rollout-2026-01-08T17-00-06-019ba044-90fa-7b30-be9c-6b7b601599cd.jsonl"),
            "{}",
        )
        .unwrap();
        std::fs::write(
            temp_dir
                .path()
                .join("rollout-2026-01-08T16-00-00-abcd1234-5678-90ab-cdef-1234567890ab.jsonl"),
            "{}",
        )
        .unwrap();
        std::fs::write(temp_dir.path().join("other.txt"), "not a session").unwrap();

        let results = list_session_files_in_dir(temp_dir.path());
        assert_eq!(results.len(), 2);

        let session_ids: Vec<_> = results.iter().map(|(id, _)| id.as_str()).collect();
        assert!(session_ids.contains(&"019ba044-90fa-7b30-be9c-6b7b601599cd"));
        assert!(session_ids.contains(&"abcd1234-5678-90ab-cdef-1234567890ab"));
    }

    #[test]
    fn test_codex_sessions_dir() {
        let dir = codex_sessions_dir();
        assert!(dir.is_some());
        assert!(dir.unwrap().ends_with(".codex/sessions"));
    }
}