codex/codex-rs/cli/src/doctor/thread_inventory.rs

//! Doctor check that compares rollout files against the SQLite thread inventory.

use super::CheckStatus;
use super::Config;
use super::DoctorCheck;
use super::DoctorIssue;
use codex_protocol::protocol::InternalSessionSource;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_rollout::RolloutRecorder;
use codex_state::ThreadStateAuditRow;
use codex_utils_path::normalize_for_path_comparison;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::collections::HashSet;
use std::ffi::OsStr;
use std::path::Path;
use std::path::PathBuf;

const MAX_PARITY_SCAN_FILES: usize = 10_000;
const SAMPLE_LIMIT: usize = 5;
const SUMMARY_LIMIT: usize = 8;
const CHECK_ID: &str = "state.rollout_db_parity";
const CHECK_CATEGORY: &str = "threads";

#[derive(Clone, Debug)]
struct RolloutAuditFile {
    path: PathBuf,
    key: PathBuf,
    archived: bool,
    thread_id: String,
}

#[derive(Default)]
struct RolloutScan {
    files: Vec<RolloutAuditFile>,
    scan_errors: Vec<String>,
    malformed_names: Vec<PathBuf>,
    reached_scan_cap: bool,
}

enum RolloutThreadId {
    Id(String),
    MalformedName,
    Unusable(String),
}

impl RolloutScan {
    fn candidate_count(&self) -> usize {
        self.files.len() + self.malformed_names.len() + self.scan_errors.len()
    }

    fn reached_candidate_cap(&self) -> bool {
        self.candidate_count() >= MAX_PARITY_SCAN_FILES
    }

    fn record_malformed_name(&mut self, path: PathBuf) {
        if self.reached_candidate_cap() {
            self.reached_scan_cap = true;
            return;
        }
        self.malformed_names.push(path);
        self.reached_scan_cap = self.reached_candidate_cap();
    }

    fn record_scan_error(&mut self, message: String) {
        if self.reached_candidate_cap() {
            self.reached_scan_cap = true;
            return;
        }
        self.scan_errors.push(message);
        self.reached_scan_cap = self.reached_candidate_cap();
    }

    fn active_count(&self) -> usize {
        self.files.iter().filter(|file| !file.archived).count()
    }

    fn archived_count(&self) -> usize {
        self.files.iter().filter(|file| file.archived).count()
    }
}

pub(super) async fn thread_inventory_check(config: &Config) -> DoctorCheck {
    thread_inventory_check_for_roots(
        config.codex_home.as_path(),
        config.sqlite_home.as_path(),
        config.model_provider_id.as_str(),
    )
    .await
}

async fn thread_inventory_check_for_roots(
    codex_home: &Path,
    sqlite_home: &Path,
    default_provider: &str,
) -> DoctorCheck {
    let scan = scan_rollout_files(codex_home).await;
    let state_db_path = codex_state::state_db_path(sqlite_home);

    let mut details = vec![
        format!("default model provider: {default_provider}"),
        format!("rollout DB active files: {}", scan.active_count()),
        format!("rollout DB archived files: {}", scan.archived_count()),
        format!("rollout DB scan errors: {}", scan.scan_errors.len()),
        format!(
            "rollout DB malformed file names: {}",
            scan.malformed_names.len()
        ),
        format!("rollout DB scan cap reached: {}", scan.reached_scan_cap),
    ];
    push_samples(
        &mut details,
        "rollout DB scan error sample",
        scan.scan_errors.iter().map(String::as_str),
    );
    push_samples(
        &mut details,
        "rollout DB malformed file sample",
        scan.malformed_names
            .iter()
            .map(|path| path.display().to_string()),
    );

    if !state_db_path.is_file() {
        details.push("rollout DB rows: skipped (state DB missing)".to_string());
        return missing_state_db_check(scan, details);
    }

    let rows = match codex_state::read_thread_state_audit_rows(&state_db_path).await {
        Ok(rows) => rows,
        Err(err) => {
            details.push(format!("rollout DB read error: {err}"));
            return DoctorCheck::new(
                CHECK_ID,
                CHECK_CATEGORY,
                CheckStatus::Warning,
                "state database thread inventory could not be read",
            )
            .details(details)
            .issue(
                DoctorIssue::new(
                    CheckStatus::Warning,
                    "state DB thread rows could not be queried",
                )
                .measured(err.to_string())
                .expected("readable threads table"),
            );
        }
    };

    parity_check_from_scan_and_rows(codex_home, scan, rows, details)
}

fn missing_state_db_check(scan: RolloutScan, details: Vec<String>) -> DoctorCheck {
    if scan.files.is_empty()
        && scan.scan_errors.is_empty()
        && scan.malformed_names.is_empty()
        && !scan.reached_scan_cap
    {
        return DoctorCheck::new(
            CHECK_ID,
            CHECK_CATEGORY,
            CheckStatus::Ok,
            "no rollout/state DB inventory to compare",
        )
        .details(details);
    }

    let summary = if scan.files.is_empty() {
        "rollout scan was incomplete or found bad files"
    } else {
        "state DB is missing while rollout files exist"
    };
    let mut check =
        DoctorCheck::new(CHECK_ID, CHECK_CATEGORY, CheckStatus::Warning, summary).details(details);

    if !scan.files.is_empty() {
        check = check
            .issue(
                DoctorIssue::new(
                    CheckStatus::Warning,
                    "rollout files exist but the state DB is missing",
                )
                .measured(format!("{} rollout files", scan.files.len()))
                .expected("state DB contains matching thread rows")
                .remedy("Start Codex with no state DB present so startup backfill can create it from rollout files."),
        )
            .remediation(
                "Start Codex with no state DB present so startup backfill can create it from rollout files.",
            );
    }
    if !scan.scan_errors.is_empty() || !scan.malformed_names.is_empty() || scan.reached_scan_cap {
        check = check.issue(
            DoctorIssue::new(
                CheckStatus::Warning,
                "rollout scan was incomplete or found bad files",
            )
            .measured(format!(
                "{} scan errors, {} malformed names, scan cap reached: {}",
                scan.scan_errors.len(),
                scan.malformed_names.len(),
                scan.reached_scan_cap
            ))
            .expected("rollout directories are fully scannable")
            .remedy("Check file permissions and unexpected files under CODEX_HOME sessions."),
        );
    }
    check
}

fn parity_check_from_scan_and_rows(
    codex_home: &Path,
    scan: RolloutScan,
    rows: Vec<ThreadStateAuditRow>,
    mut details: Vec<String>,
) -> DoctorCheck {
    let rollout_by_key = scan
        .files
        .iter()
        .map(|file| (file.key.clone(), file))
        .collect::<HashMap<_, _>>();
    let mut rows_by_key: HashMap<PathBuf, Vec<&ThreadStateAuditRow>> = HashMap::new();
    for row in &rows {
        rows_by_key
            .entry(path_key(&row.rollout_path))
            .or_default()
            .push(row);
    }

    let missing_active = missing_rollout_paths(&scan.files, &rows_by_key, /*archived*/ false);
    let missing_archived = missing_rollout_paths(&scan.files, &rows_by_key, /*archived*/ true);
    let scan_complete = !scan.reached_scan_cap;
    let stale_rows = if scan_complete {
        rows.iter()
            .filter(|row| !row.rollout_path.is_file())
            .collect::<Vec<_>>()
    } else {
        Vec::new()
    };
    let archive_mismatches = if scan_complete {
        rows.iter()
            .filter_map(|row| {
                let expected_archived = rollout_by_key
                    .get(&path_key(&row.rollout_path))
                    .map(|file| file.archived)
                    .or_else(|| {
                        row.rollout_path
                            .is_file()
                            .then(|| archived_from_rollout_path(codex_home, &row.rollout_path))
                            .flatten()
                    })?;
                (expected_archived != row.archived).then_some(row)
            })
            .collect::<Vec<_>>()
    } else {
        Vec::new()
    };
    let duplicate_rollout_thread_ids = duplicate_rollout_thread_ids(&scan.files);
    let duplicate_db_paths = duplicate_db_paths(&rows_by_key);
    let archived_rows = rows.iter().filter(|row| row.archived).count();
    let active_rows = rows.len() - archived_rows;

    details.extend([
        format!("rollout DB rows: {}", rows.len()),
        format!("rollout DB active rows: {active_rows}"),
        format!("rollout DB archived rows: {archived_rows}"),
        format!("rollout DB missing active rows: {}", missing_active.len()),
        format!(
            "rollout DB missing archived rows: {}",
            missing_archived.len()
        ),
        format!(
            "rollout DB stale rows: {}",
            count_or_skipped(stale_rows.len(), scan_complete)
        ),
        format!(
            "rollout DB archive mismatches: {}",
            count_or_skipped(archive_mismatches.len(), scan_complete)
        ),
        format!(
            "rollout DB duplicate rollout thread ids: {}",
            duplicate_rollout_thread_ids.len()
        ),
        format!(
            "rollout DB duplicate DB paths: {}",
            duplicate_db_paths.len()
        ),
        format!(
            "rollout DB model providers: {}",
            count_summary(rows.iter().map(|row| row.model_provider.as_str()))
        ),
        format!(
            "rollout DB sources: {}",
            count_summary(rows.iter().map(|row| source_category(&row.source)))
        ),
    ]);
    push_path_samples(
        &mut details,
        "rollout DB missing active sample",
        missing_active.iter().copied(),
    );
    push_path_samples(
        &mut details,
        "rollout DB missing archived sample",
        missing_archived.iter().copied(),
    );
    push_path_samples(
        &mut details,
        "rollout DB stale row sample",
        stale_rows.iter().map(|row| row.rollout_path.as_path()),
    );
    push_path_samples(
        &mut details,
        "rollout DB archive mismatch sample",
        archive_mismatches
            .iter()
            .map(|row| row.rollout_path.as_path()),
    );
    push_samples(
        &mut details,
        "rollout DB duplicate rollout thread id sample",
        duplicate_rollout_thread_ids.iter().map(String::as_str),
    );
    push_path_samples(
        &mut details,
        "rollout DB duplicate DB path sample",
        duplicate_db_paths.iter().map(PathBuf::as_path),
    );

    let status = if scan.scan_errors.is_empty()
        && scan.malformed_names.is_empty()
        && !scan.reached_scan_cap
        && missing_active.is_empty()
        && missing_archived.is_empty()
        && stale_rows.is_empty()
        && archive_mismatches.is_empty()
        && duplicate_rollout_thread_ids.is_empty()
        && duplicate_db_paths.is_empty()
    {
        CheckStatus::Ok
    } else {
        CheckStatus::Warning
    };

    let summary = if status == CheckStatus::Ok {
        "rollout files and state DB thread inventory agree"
    } else {
        "rollout files and state DB thread inventory differ"
    };
    let mut check = DoctorCheck::new(CHECK_ID, CHECK_CATEGORY, status, summary).details(details);

    if !missing_active.is_empty() || !missing_archived.is_empty() {
        check = check.issue(
            DoctorIssue::new(
                CheckStatus::Warning,
                "rollout files are missing from the state DB",
            )
            .measured(format!(
                "{} active, {} archived",
                missing_active.len(),
                missing_archived.len()
            ))
            .expected("every rollout file has a matching threads row"),
        );
    }
    if !stale_rows.is_empty() {
        check = check.issue(
            DoctorIssue::new(
                CheckStatus::Warning,
                "state DB rows point at missing or unusable rollout files",
            )
            .measured(format!("{} stale rows", stale_rows.len()))
            .expected("every state DB rollout path is a file on disk"),
        );
    }
    if !archive_mismatches.is_empty() {
        check = check.issue(
            DoctorIssue::new(
                CheckStatus::Warning,
                "state DB archive flags disagree with rollout file locations",
            )
            .measured(format!("{} mismatched rows", archive_mismatches.len()))
            .expected(
                "rows under archived_sessions are archived and rows under sessions are active",
            ),
        );
    }
    if !duplicate_rollout_thread_ids.is_empty() || !duplicate_db_paths.is_empty() {
        check = check.issue(
            DoctorIssue::new(
                CheckStatus::Warning,
                "duplicate thread inventory entries found",
            )
            .measured(format!(
                "{} duplicate rollout thread ids, {} duplicate DB paths",
                duplicate_rollout_thread_ids.len(),
                duplicate_db_paths.len()
            ))
            .expected("one rollout path and thread id per thread")
            .remedy("Attach the doctor report to a bug report so support can inspect samples."),
        );
    }
    if !scan.scan_errors.is_empty() || !scan.malformed_names.is_empty() || scan.reached_scan_cap {
        check = check.issue(
            DoctorIssue::new(
                CheckStatus::Warning,
                "rollout scan was incomplete or found bad files",
            )
            .measured(format!(
                "{} scan errors, {} malformed names, scan cap reached: {}",
                scan.scan_errors.len(),
                scan.malformed_names.len(),
                scan.reached_scan_cap
            ))
            .expected("rollout directories are fully scannable")
            .remedy("Check file permissions and unexpected files under CODEX_HOME sessions."),
        );
    }
    check
}

async fn scan_rollout_files(codex_home: &Path) -> RolloutScan {
    let mut scan = RolloutScan::default();
    scan_rollout_root(
        &codex_home.join("sessions"),
        /*archived*/ false,
        &mut scan,
    )
    .await;
    scan_rollout_root(
        &codex_home.join("archived_sessions"),
        /*archived*/ true,
        &mut scan,
    )
    .await;
    scan
}

async fn scan_rollout_root(root: &Path, archived: bool, scan: &mut RolloutScan) {
    let mut dirs = vec![root.to_path_buf()];
    while let Some(dir) = dirs.pop() {
        if scan.reached_scan_cap {
            return;
        }
        let entries = match std::fs::read_dir(&dir) {
            Ok(entries) => entries,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue,
            Err(err) => {
                scan.record_scan_error(format!("{} ({err})", dir.display()));
                continue;
            }
        };
        for entry in entries {
            if scan.reached_scan_cap {
                return;
            }
            let entry = match entry {
                Ok(entry) => entry,
                Err(err) => {
                    scan.record_scan_error(format!("{} ({err})", dir.display()));
                    continue;
                }
            };
            let path = entry.path();
            let file_type = match entry.file_type() {
                Ok(file_type) => file_type,
                Err(err) => {
                    scan.record_scan_error(format!("{} ({err})", path.display()));
                    continue;
                }
            };
            if file_type.is_dir() {
                dirs.push(path);
                continue;
            }
            if !file_type.is_file() || !is_rollout_file(&path) {
                continue;
            }
            if scan.candidate_count() >= MAX_PARITY_SCAN_FILES {
                scan.reached_scan_cap = true;
                return;
            }
            let thread_id = match thread_id_from_rollout(&path).await {
                RolloutThreadId::Id(thread_id) => thread_id,
                RolloutThreadId::MalformedName => {
                    scan.record_malformed_name(path.clone());
                    continue;
                }
                RolloutThreadId::Unusable(reason) => {
                    scan.record_scan_error(format!("{} ({reason})", path.display()));
                    continue;
                }
            };
            scan.files.push(RolloutAuditFile {
                key: path_key(&path),
                path,
                archived,
                thread_id,
            });
        }
    }
}

async fn thread_id_from_rollout(path: &Path) -> RolloutThreadId {
    let items = match RolloutRecorder::load_rollout_items(path).await {
        Ok((items, _, _)) => items,
        Err(err) => return RolloutThreadId::Unusable(err.to_string()),
    };
    if items.is_empty() {
        return RolloutThreadId::Unusable("no parseable rollout items".to_string());
    }
    codex_rollout::builder_from_items(items.as_slice(), path)
        .map(|builder| RolloutThreadId::Id(builder.id.to_string()))
        .unwrap_or(RolloutThreadId::MalformedName)
}

fn is_rollout_file(path: &Path) -> bool {
    path.extension() == Some(OsStr::new("jsonl"))
        && path
            .file_name()
            .and_then(OsStr::to_str)
            .is_some_and(|name| name.starts_with("rollout-"))
}

fn count_or_skipped(count: usize, complete: bool) -> String {
    if complete {
        count.to_string()
    } else {
        "skipped (scan cap reached)".to_string()
    }
}

fn path_key(path: &Path) -> PathBuf {
    normalize_for_path_comparison(path).unwrap_or_else(|_| path.to_path_buf())
}

fn archived_from_rollout_path(codex_home: &Path, path: &Path) -> Option<bool> {
    let key = path_key(path);
    if key.starts_with(path_key(&codex_home.join("archived_sessions"))) {
        return Some(true);
    }
    if key.starts_with(path_key(&codex_home.join("sessions"))) {
        return Some(false);
    }
    None
}

fn missing_rollout_paths<'a>(
    files: &'a [RolloutAuditFile],
    rows_by_key: &HashMap<PathBuf, Vec<&ThreadStateAuditRow>>,
    archived: bool,
) -> Vec<&'a Path> {
    files
        .iter()
        .filter(|file| file.archived == archived && !has_matching_thread_row(file, rows_by_key))
        .map(|file| file.path.as_path())
        .collect()
}

fn has_matching_thread_row(
    file: &RolloutAuditFile,
    rows_by_key: &HashMap<PathBuf, Vec<&ThreadStateAuditRow>>,
) -> bool {
    let Some(rows) = rows_by_key.get(&file.key) else {
        return false;
    };
    rows.iter().any(|row| row.id == file.thread_id.as_str())
}

fn duplicate_rollout_thread_ids(files: &[RolloutAuditFile]) -> Vec<String> {
    let mut seen = HashSet::new();
    let mut duplicates = HashSet::new();
    for thread_id in files.iter().map(|file| file.thread_id.as_str()) {
        if !seen.insert(thread_id) {
            duplicates.insert(thread_id.to_string());
        }
    }
    let mut duplicates = duplicates.into_iter().collect::<Vec<_>>();
    duplicates.sort();
    duplicates
}

fn duplicate_db_paths(rows_by_key: &HashMap<PathBuf, Vec<&ThreadStateAuditRow>>) -> Vec<PathBuf> {
    let mut paths = rows_by_key
        .iter()
        .filter(|(_, rows)| rows.len() > 1)
        .map(|(path, _)| path.clone())
        .collect::<Vec<_>>();
    paths.sort();
    paths
}

fn source_category(source: &str) -> &'static str {
    let parsed = serde_json::from_str::<SessionSource>(source)
        .or_else(|_| serde_json::from_value(serde_json::Value::String(source.to_string())));
    let Ok(source) = parsed else {
        return "unparsable";
    };

    match source {
        SessionSource::Cli => "cli",
        SessionSource::VSCode => "vscode",
        SessionSource::Exec => "exec",
        SessionSource::Mcp => "mcp",
        SessionSource::Custom(_) => "custom",
        SessionSource::Internal(InternalSessionSource::MemoryConsolidation) => {
            "internal:memory_consolidation"
        }
        SessionSource::SubAgent(SubAgentSource::Review) => "subagent:review",
        SessionSource::SubAgent(SubAgentSource::Compact) => "subagent:compact",
        SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. }) => "subagent:thread_spawn",
        SessionSource::SubAgent(SubAgentSource::MemoryConsolidation) => {
            "subagent:memory_consolidation"
        }
        SessionSource::SubAgent(SubAgentSource::Other(_)) => "subagent:other",
        SessionSource::Unknown => "unknown",
    }
}

fn count_summary<I, V>(values: I) -> String
where
    I: Iterator<Item = V>,
    V: Into<String>,
{
    let mut counts = BTreeMap::<String, usize>::new();
    for value in values {
        *counts.entry(value.into()).or_default() += 1;
    }
    if counts.is_empty() {
        return "none".to_string();
    }

    let mut entries = counts.into_iter().collect::<Vec<_>>();
    entries.sort_by(|(left_value, left_count), (right_value, right_count)| {
        right_count
            .cmp(left_count)
            .then_with(|| left_value.cmp(right_value))
    });
    let omitted_categories = entries.len().saturating_sub(SUMMARY_LIMIT);
    let omitted_rows = entries
        .iter()
        .skip(SUMMARY_LIMIT)
        .map(|(_, count)| count)
        .sum::<usize>();
    let mut parts = entries
        .into_iter()
        .take(SUMMARY_LIMIT)
        .map(|(value, count)| format!("{value}={count}"))
        .collect::<Vec<_>>();
    if omitted_categories > 0 {
        parts.push(format!(
            "other={omitted_rows} across {omitted_categories} categories"
        ));
    }
    parts.join(", ")
}

fn push_path_samples<'a>(
    details: &mut Vec<String>,
    label: &str,
    paths: impl Iterator<Item = &'a Path>,
) {
    push_samples(details, label, paths.map(|path| path.display().to_string()));
}

fn push_samples<I, V>(details: &mut Vec<String>, label: &str, values: I)
where
    I: Iterator<Item = V>,
    V: ToString,
{
    for value in values.take(SAMPLE_LIMIT) {
        details.push(format!("{label}: {}", value.to_string()));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use codex_protocol::ThreadId;
    use codex_protocol::protocol::RolloutItem;
    use codex_protocol::protocol::RolloutLine;
    use pretty_assertions::assert_eq;
    use sqlx::sqlite::SqliteConnectOptions;
    use sqlx::sqlite::SqlitePoolOptions;
    use tempfile::TempDir;

    #[tokio::test]
    async fn thread_inventory_check_ok_when_rollouts_match_db() {
        let fixture = Fixture::new().await;
        let active_path = fixture.write_rollout(
            /*archived*/ false,
            "2025-01-02T10-00-00",
            "00000000-0000-0000-0000-000000000001",
        );
        let archived_path = fixture.write_rollout(
            /*archived*/ true,
            "2025-01-02T11-00-00",
            "00000000-0000-0000-0000-000000000002",
        );
        fixture
            .insert_thread_row(
                "00000000-0000-0000-0000-000000000001",
                active_path.as_path(),
                /*archived*/ false,
            )
            .await;
        fixture
            .insert_thread_row(
                "00000000-0000-0000-0000-000000000002",
                archived_path.as_path(),
                /*archived*/ true,
            )
            .await;

        let check = thread_inventory_check_for_roots(
            fixture.codex_home.path(),
            fixture.sqlite_home.path(),
            "test-provider",
        )
        .await;

        assert_eq!(check.status, CheckStatus::Ok);
        assert_eq!(check.category, CHECK_CATEGORY);
        assert_detail(&check, "rollout DB missing active rows", "0");
        assert_detail(&check, "rollout DB missing archived rows", "0");
        assert_detail(&check, "rollout DB stale rows", "0");
        assert_detail(&check, "rollout DB archive mismatches", "0");
    }

    #[tokio::test]
    async fn thread_inventory_check_warns_for_missing_stale_and_mismatched_rows() {
        let fixture = Fixture::new().await;
        let missing_path = fixture.write_rollout(
            /*archived*/ false,
            "2025-01-02T10-00-00",
            "00000000-0000-0000-0000-000000000001",
        );
        let mismatched_path = fixture.write_rollout(
            /*archived*/ true,
            "2025-01-02T11-00-00",
            "00000000-0000-0000-0000-000000000002",
        );
        let stale_path = fixture
            .codex_home
            .path()
            .join("sessions/2025/01/02/rollout-2025-01-02T12-00-00-00000000-0000-0000-0000-000000000003.jsonl");
        fixture
            .insert_thread_row(
                "00000000-0000-0000-0000-000000000002",
                mismatched_path.as_path(),
                /*archived*/ false,
            )
            .await;
        fixture
            .insert_thread_row(
                "00000000-0000-0000-0000-000000000003",
                stale_path.as_path(),
                /*archived*/ false,
            )
            .await;

        let check = thread_inventory_check_for_roots(
            fixture.codex_home.path(),
            fixture.sqlite_home.path(),
            "test-provider",
        )
        .await;

        assert_eq!(check.status, CheckStatus::Warning);
        assert_eq!(check.issues.len(), 3);
        assert_detail(&check, "rollout DB missing active rows", "1");
        assert_detail(&check, "rollout DB stale rows", "1");
        assert_detail(&check, "rollout DB archive mismatches", "1");
        assert_eq!(check.remediation, None);
        assert!(check.issues.iter().all(|issue| {
            !issue
                .remedy
                .as_deref()
                .is_some_and(|remedy| remedy.starts_with("Restart Codex"))
        }));
        assert!(
            check
                .details
                .iter()
                .any(|detail| detail.contains(missing_path.to_string_lossy().as_ref()))
        );
    }

    struct Fixture {
        codex_home: TempDir,
        sqlite_home: TempDir,
    }

    impl Fixture {
        async fn new() -> Self {
            let codex_home = TempDir::new().expect("codex home");
            let sqlite_home = TempDir::new().expect("sqlite home");
            let _runtime = codex_state::StateRuntime::init(
                sqlite_home.path().to_path_buf(),
                "test-provider".to_string(),
            )
            .await
            .expect("state runtime");
            Self {
                codex_home,
                sqlite_home,
            }
        }

        fn write_rollout(&self, archived: bool, timestamp: &str, thread_id: &str) -> PathBuf {
            let root = if archived {
                self.codex_home.path().join("archived_sessions")
            } else {
                self.codex_home.path().join("sessions/2025/01/02")
            };
            std::fs::create_dir_all(&root).expect("rollout dir");
            let path = root.join(format!("rollout-{timestamp}-{thread_id}.jsonl"));
            let parsed_thread_id = ThreadId::from_string(thread_id).expect("thread id");
            let rollout_line = RolloutLine {
                timestamp: timestamp.to_string(),
                item: RolloutItem::SessionMeta(codex_protocol::protocol::SessionMetaLine {
                    meta: codex_protocol::protocol::SessionMeta {
                        session_id: parsed_thread_id.into(),
                        id: parsed_thread_id,
                        timestamp: timestamp.to_string(),
                        cwd: self.codex_home.path().to_path_buf(),
                        originator: "test".to_string(),
                        cli_version: "test".to_string(),
                        source: SessionSource::Cli,
                        model_provider: Some("test-provider".to_string()),
                        ..Default::default()
                    },
                    git: None,
                }),
            };
            let contents = serde_json::to_string(&rollout_line).expect("rollout line");
            std::fs::write(&path, format!("{contents}\n")).expect("rollout file");
            path
        }

        async fn insert_thread_row(&self, id: &str, rollout_path: &Path, archived: bool) {
            let state_db_path = codex_state::state_db_path(self.sqlite_home.path());
            let options = SqliteConnectOptions::new()
                .filename(state_db_path)
                .create_if_missing(false);
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(options)
                .await
                .expect("sqlite pool");
            sqlx::query(
                r#"
INSERT INTO threads (
    id,
    rollout_path,
    created_at,
    updated_at,
    source,
    model_provider,
    cwd,
    title,
    sandbox_policy,
    approval_mode,
    archived,
    archived_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                "#,
            )
            .bind(id)
            .bind(rollout_path.display().to_string())
            .bind(1_i64)
            .bind(1_i64)
            .bind("cli")
            .bind("test-provider")
            .bind(self.codex_home.path().display().to_string())
            .bind("test title")
            .bind("read-only")
            .bind("on-request")
            .bind(if archived { 1_i64 } else { 0_i64 })
            .bind(archived.then_some(1_i64))
            .execute(&pool)
            .await
            .expect("insert thread row");
            pool.close().await;
        }
    }

    fn assert_detail(check: &DoctorCheck, label: &str, expected: &str) {
        let prefix = format!("{label}: ");
        let actual = check
            .details
            .iter()
            .find_map(|detail| detail.strip_prefix(&prefix))
            .expect("detail should exist");
        assert_eq!(actual, expected);
    }

    #[test]
    fn source_category_coarsens_structured_sources() {
        assert_eq!(source_category("cli"), "cli");
        assert_eq!(
            source_category(r#"{"subagent":"memory_consolidation"}"#),
            "subagent:memory_consolidation"
        );
        assert_eq!(
            source_category(
                r#"{"subagent":{"thread_spawn":{"parent_thread_id":"00000000-0000-0000-0000-000000000001","depth":2}}}"#,
            ),
            "subagent:thread_spawn"
        );
    }

    #[test]
    fn count_summary_caps_distinct_values() {
        let summary = count_summary(["a", "b", "c", "d", "e", "f", "g", "h", "i"].into_iter());

        assert_eq!(
            summary,
            "a=1, b=1, c=1, d=1, e=1, f=1, g=1, h=1, other=1 across 1 categories"
        );
    }
}