lean_ctx/core/import/
codex.rs

1//! Codex session-history import.
2
3use std::collections::HashSet;
4use std::path::{Path, PathBuf};
5
6use rusqlite::{Connection, types::ValueRef};
7use serde_json::Value;
8
9use super::database::{DatabaseSnapshot, snapshot_history_database};
10use super::{
11    ImportResult, ImportSource, SourceRefusal, finish_result, home_dir, push_error, read_json_file,
12    read_jsonl_files, session_id, staging,
13};
14
15const MAX_SQLITE_ROWS_PER_TABLE: usize = 10_000;
16const MAX_SCAN_DEPTH: usize = 8;
17
18/// Imports Codex history and session files from `~/.codex`.
19#[must_use]
20pub fn import() -> ImportResult {
21    let Some(home) = home_dir() else {
22        return ImportResult::default();
23    };
24    import_from_root(&home.join(".codex"))
25}
26
27/// Imports Codex JSONL, JSON, and SQLite session data from a Codex directory.
28#[must_use]
29pub fn import_from_root(root: &Path) -> ImportResult {
30    if !root.exists() {
31        return ImportResult::default();
32    }
33
34    let mut jsonl = Vec::new();
35    let mut json = Vec::new();
36    let mut sqlite = Vec::new();
37    let mut errors = Vec::new();
38    collect_files(root, 0, &mut jsonl, &mut json, &mut sqlite, &mut errors);
39    jsonl.sort();
40    json.sort();
41    sqlite.sort();
42
43    let mut result = read_jsonl_files(ImportSource::Codex, root, jsonl);
44    for path in json {
45        result.merge(read_json_file(ImportSource::Codex, root, &path));
46    }
47    for path in sqlite {
48        // A database is not text the admitted reader can read; its root
49        // rule is applied here, and every fact mined from its rows is
50        // admitted where it is built (`push_fact`).
51        // Root containment, then the path decision a file read gets, then ONE
52        // open of that file: it is checked on its descriptor and read from a
53        // private snapshot of that descriptor, never from its path again
54        // (`database`). Content is admitted value by value in
55        // `read_sqlite_file`, and the database is one source: all of it, or
56        // nothing but its one refusal count.
57        match snapshot_history_database(root, &path) {
58            Ok(snapshot) => result.merge(read_sqlite_file(root, &snapshot)),
59            Err(refusal) => result.refuse(refusal),
60        }
61    }
62    for error in errors {
63        push_error(&mut result, error);
64    }
65    result
66}
67
68fn collect_files(
69    dir: &Path,
70    depth: usize,
71    jsonl: &mut Vec<PathBuf>,
72    json: &mut Vec<PathBuf>,
73    sqlite: &mut Vec<PathBuf>,
74    errors: &mut Vec<String>,
75) {
76    if depth > MAX_SCAN_DEPTH {
77        return;
78    }
79    let entries = match std::fs::read_dir(dir) {
80        Ok(entries) => entries,
81        Err(error) => {
82            errors.push(format!("{}: {error}", dir.display()));
83            return;
84        }
85    };
86    for entry in entries {
87        let entry = match entry {
88            Ok(entry) => entry,
89            Err(error) => {
90                errors.push(format!("{}: {error}", dir.display()));
91                continue;
92            }
93        };
94        let path = entry.path();
95        if path.is_dir() {
96            collect_files(&path, depth + 1, jsonl, json, sqlite, errors);
97        } else if path.is_file() {
98            match path.extension().and_then(|extension| extension.to_str()) {
99                Some("jsonl") => jsonl.push(path),
100                Some("json") if is_named_session_file(&path) => json.push(path),
101                Some("sqlite" | "sqlite3" | "db") => sqlite.push(path),
102                _ => {}
103            }
104        }
105    }
106}
107
108fn is_named_session_file(path: &Path) -> bool {
109    let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
110        return false;
111    };
112    let name = name.to_ascii_lowercase();
113    ["history", "session", "rollout", "transcript"]
114        .iter()
115        .any(|marker| name.contains(marker))
116}
117
118/// One history database, as ONE source: everything its tables yield is held
119/// pending and taken over only when every table and row was read
120/// (`staging::read_database`, one policy snapshot for the whole read). A
121/// database that fails anywhere counts once (`sources_skipped`, or
122/// `sources_refused` when the policy changed under the read) and contributes
123/// nothing else — no fact of an earlier row or table — but one line: its path
124/// and the fixed sentence of the typed refusal. Never the engine's message:
125/// that names the private snapshot or quotes the schema.
126fn read_sqlite_file(root: &Path, snapshot: &DatabaseSnapshot) -> ImportResult {
127    // The history path names the session; what is opened is the snapshot of
128    // the verified descriptor.
129    let path = snapshot.source_path();
130    let (pending, read) =
131        staging::read_database(root, path, |pending| read_rows(snapshot, pending));
132    let session = session_id(ImportSource::Codex, path, pending.rules());
133    let mut result = ImportResult::default();
134    let mut touched = HashSet::new();
135    let mut seen = HashSet::new();
136    match pending.settle(
137        read,
138        ImportSource::Codex,
139        &session,
140        &mut result,
141        &mut touched,
142        &mut seen,
143    ) {
144        Ok(()) => result.sessions_found = 1,
145        // The reason, as a fixed sentence chosen by the typed refusal, under
146        // the label this importer has always printed for a database.
147        Err(refusal) => {
148            let label = path.display().to_string();
149            push_error(&mut result, staging::not_imported(&label, refusal));
150        }
151    }
152    finish_result(&mut result, &touched);
153    result
154}
155
156/// Stage what every TEXT column value of every table yields. `Err` for any
157/// failure of the read — the caller then takes over nothing.
158fn read_rows(
159    snapshot: &DatabaseSnapshot,
160    pending: &mut staging::Pending,
161) -> Result<(), SourceRefusal> {
162    let unreadable = |_: rusqlite::Error| SourceRefusal::DatabaseUnreadable;
163    let connection = snapshot.open().map_err(unreadable)?;
164    for table in tables(&connection).map_err(unreadable)? {
165        let query = format!(
166            "SELECT * FROM {} LIMIT {MAX_SQLITE_ROWS_PER_TABLE}",
167            quote_identifier(&table)
168        );
169        let mut statement = connection.prepare(&query).map_err(unreadable)?;
170        let columns = statement.column_count();
171        let mut rows = statement.query([]).map_err(unreadable)?;
172        while let Some(row) = rows.next().map_err(unreadable)? {
173            for column in 0..columns {
174                let Ok(ValueRef::Text(bytes)) = row.get_ref(column) else {
175                    continue;
176                };
177                let text = String::from_utf8_lossy(bytes);
178                let original = serde_json::from_str::<Value>(&text)
179                    .unwrap_or_else(|_| Value::String(text.into_owned()));
180                // The ORIGINAL column value is admitted as a whole before
181                // anything is selected from it; a refused value yields
182                // nothing (counted when the database is taken over).
183                let Some(value) = pending.admit(original) else {
184                    continue;
185                };
186                staging::select(&value, |found| pending.stage(found));
187            }
188        }
189    }
190    Ok(())
191}
192
193fn tables(connection: &Connection) -> rusqlite::Result<Vec<String>> {
194    let mut statement = connection.prepare(
195        "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
196    )?;
197    statement
198        .query_map([], |row| row.get(0))?
199        .collect::<rusqlite::Result<Vec<String>>>()
200}
201
202fn quote_identifier(identifier: &str) -> String {
203    format!("\"{}\"", identifier.replace('\"', "\"\""))
204}
lean_ctx/core/import/codex.rs

lean_ctx/core/import/
codex.rs
