mj_controller/import/
codex.rs

1use super::*;
2
3/// Locate a Codex rollout exposed by its native interactive resume picker.
4pub fn locate_codex_session(
5    home: &Path,
6    selection: &CodexSessionSelection,
7) -> Result<LocatedCodexSession> {
8    let mut listed = list_codex_sessions(home)?;
9    // `--latest` follows Codex's own default view, which hides what the user
10    // archived there. Asking for an id by name still finds it.
11    if matches!(selection, CodexSessionSelection::Latest) {
12        listed.retain(|session| !session.natively_archived);
13    }
14    if let CodexSessionSelection::NativeSessionId(session_id) = selection
15        && !listed
16            .iter()
17            .any(|session| session.native_session_id == *session_id)
18    {
19        return locate_unindexed_codex_session(home, session_id);
20    }
21    select_jsonl_session(listed, selection, "Codex")
22}
23
24pub(super) fn locate_unindexed_codex_session(
25    home: &Path,
26    session_id: &str,
27) -> Result<LocatedCodexSession> {
28    validate_id("Codex session", session_id)?;
29    let mut requested = BTreeMap::new();
30    requested.insert(session_id.to_owned(), session_id.to_owned());
31    let mut candidates = Vec::new();
32    let root = home.join("sessions");
33    if root.is_dir() {
34        collect_codex_candidate_paths(&root, &requested, &mut candidates)?;
35    }
36    let titles = codex_native_titles(home)?;
37    let mut matches = Vec::new();
38    for candidate in candidates {
39        let Some(metadata) = codex_session_metadata(&candidate.path)? else {
40            continue;
41        };
42        if metadata.id == session_id {
43            matches.push(LocatedCodexSession {
44                natively_archived: false,
45                title: titles
46                    .get(session_id)
47                    .cloned()
48                    .unwrap_or_else(|| session_id.to_owned()),
49                native_session_id: metadata.id,
50                jsonl_path: candidate.path,
51                modified_at: candidate.modified_at,
52                cwd: metadata.cwd,
53                git_branch: metadata.git_branch,
54                size_bytes: candidate.size_bytes,
55                history_mode: metadata.history_mode,
56            });
57        }
58    }
59    select_jsonl_session(
60        matches,
61        &CodexSessionSelection::NativeSessionId(session_id.to_owned()),
62        "Codex",
63    )
64}
65
66/// List native Codex sessions newest first.
67pub fn list_codex_sessions(home: &Path) -> Result<Vec<LocatedCodexSession>> {
68    let mut sessions = Vec::new();
69    scan_codex_sessions(home, |progress| {
70        if let Some(session) = progress.session {
71            sessions.push(session);
72        }
73    })?;
74    Ok(sessions)
75}
76
77/// Scan native Codex sessions newest first, reporting after every candidate file.
78pub fn scan_codex_sessions(
79    home: &Path,
80    mut report: impl FnMut(SessionScanProgress<LocatedCodexSession>),
81) -> Result<()> {
82    if let Some(sessions) = codex_indexed_sessions(home)? {
83        let total = sessions.len();
84        report(SessionScanProgress {
85            scanned: 0,
86            total,
87            session: None,
88        });
89        for (index, session) in sessions.into_iter().enumerate() {
90            report(SessionScanProgress {
91                scanned: index + 1,
92                total,
93                session: Some(session),
94            });
95        }
96        return Ok(());
97    }
98
99    // Native Codex only indexes threads with a non-empty preview/name. Its
100    // history and session-name index provide the same compact set of IDs,
101    // avoiding an expensive parse of every exec and subagent rollout.
102    let titles = codex_native_titles(home)?;
103    let mut candidates = Vec::new();
104    let root = home.join("sessions");
105    if root.is_dir() {
106        collect_codex_candidate_paths(&root, &titles, &mut candidates)?;
107    }
108    candidates.sort_by(|left, right| {
109        right
110            .modified_at
111            .cmp(&left.modified_at)
112            .then_with(|| right.path.cmp(&left.path))
113    });
114    let total = candidates.len();
115    report(SessionScanProgress {
116        scanned: 0,
117        total,
118        session: None,
119    });
120    for (index, candidate) in candidates.into_iter().enumerate() {
121        let session = codex_session_metadata(&candidate.path)?.map(|metadata| {
122            let session_id = metadata.id;
123            LocatedCodexSession {
124                natively_archived: false,
125                title: titles
126                    .get(&session_id)
127                    .cloned()
128                    .unwrap_or_else(|| session_id.clone()),
129                native_session_id: session_id,
130                jsonl_path: candidate.path,
131                modified_at: candidate.modified_at,
132                cwd: metadata.cwd,
133                git_branch: metadata.git_branch,
134                size_bytes: candidate.size_bytes,
135                history_mode: metadata.history_mode,
136            }
137        });
138        report(SessionScanProgress {
139            scanned: index + 1,
140            total,
141            session,
142        });
143    }
144    Ok(())
145}
146
147pub(super) fn codex_indexed_sessions(home: &Path) -> Result<Option<Vec<LocatedCodexSession>>> {
148    let database = home.join("state_5.sqlite");
149    if !database.is_file() {
150        return Ok(None);
151    }
152    let connection = rusqlite::Connection::open_with_flags(
153        database,
154        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
155    )?;
156    let has_history_mode = connection
157        .prepare("SELECT history_mode FROM threads LIMIT 0")
158        .is_ok();
159    let history_mode_column = if has_history_mode {
160        "history_mode"
161    } else {
162        "'legacy'"
163    };
164    // Codex's own archived threads are listed too, flagged rather than
165    // filtered: the resume dialog hides them until "show archived" is on, and
166    // Hel never writes this database back.
167    let query = format!(
168        "SELECT id, rollout_path, updated_at, COALESCE(NULLIF(name, ''), NULLIF(title, ''), id), cwd, \
169         COALESCE(NULLIF(git_branch, ''), 'HEAD'), {history_mode_column}, archived \
170         FROM threads \
171         WHERE source IN ('cli', 'vscode') \
172           AND preview <> '' \
173           AND rollout_path IS NOT NULL \
174         ORDER BY updated_at DESC, id DESC"
175    );
176    let Ok(mut statement) = connection.prepare(&query) else {
177        return Ok(None);
178    };
179    let rows = statement.query_map([], |row| {
180        Ok((
181            row.get::<_, String>(0)?,
182            row.get::<_, String>(1)?,
183            row.get::<_, i64>(2)?,
184            row.get::<_, String>(3)?,
185            row.get::<_, String>(4)?,
186            row.get::<_, String>(5)?,
187            row.get::<_, String>(6)?,
188            row.get::<_, bool>(7)?,
189        ))
190    })?;
191    let mut sessions = Vec::new();
192    for row in rows {
193        let (session_id, path, updated_at, title, cwd, git_branch, history_mode, natively_archived) =
194            row?;
195        let path = PathBuf::from(path);
196        if validate_id("Codex session", &session_id).is_err() || updated_at.is_negative() {
197            continue;
198        }
199        let Ok(metadata) = fs::symlink_metadata(&path) else {
200            continue;
201        };
202        if metadata.file_type().is_symlink() || !metadata.is_file() {
203            continue;
204        }
205        sessions.push(LocatedCodexSession {
206            native_session_id: session_id.clone(),
207            jsonl_path: path,
208            modified_at: SystemTime::UNIX_EPOCH + Duration::from_secs(updated_at as u64),
209            title: normalize_session_title(&title).unwrap_or(session_id),
210            cwd: PathBuf::from(cwd),
211            git_branch,
212            size_bytes: metadata.len(),
213            history_mode: parse_codex_history_mode(&history_mode)?,
214            natively_archived,
215        });
216    }
217    Ok(Some(sessions))
218}
219
220pub(super) fn collect_codex_candidate_paths(
221    root: &Path,
222    native_titles: &BTreeMap<String, String>,
223    candidates: &mut Vec<FileScanCandidate>,
224) -> Result<()> {
225    for entry in fs::read_dir(root)? {
226        let entry = entry?;
227        let path = entry.path();
228        let metadata = fs::symlink_metadata(&path)?;
229        if metadata.file_type().is_symlink() {
230            continue;
231        }
232        if metadata.is_dir() {
233            collect_codex_candidate_paths(&path, native_titles, candidates)?;
234            continue;
235        }
236        if !metadata.is_file() || path.extension().and_then(|value| value.to_str()) != Some("jsonl")
237        {
238            continue;
239        }
240        if let Some(session_id) = codex_rollout_id_from_path(&path)
241            && !native_titles.contains_key(session_id)
242        {
243            continue;
244        }
245        candidates.push(FileScanCandidate {
246            path,
247            modified_at: metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH),
248            size_bytes: metadata.len(),
249        });
250    }
251    Ok(())
252}
253
254pub(super) fn codex_rollout_id_from_path(path: &Path) -> Option<&str> {
255    let stem = path.file_stem()?.to_str()?;
256    let id = stem.get(stem.len().checked_sub(36)?..)?;
257    (id.as_bytes().get(8) == Some(&b'-')
258        && id.as_bytes().get(13) == Some(&b'-')
259        && id.as_bytes().get(18) == Some(&b'-')
260        && id.as_bytes().get(23) == Some(&b'-'))
261    .then_some(id)
262}
263
264pub(super) fn codex_session_metadata(path: &Path) -> Result<Option<CodexSessionMetadata>> {
265    let file =
266        fs::File::open(path).with_context(|| format!("open Codex session {}", path.display()))?;
267    let mut reader = BufReader::new(file);
268    let mut line = String::new();
269    for _ in 0..8 {
270        line.clear();
271        if reader.read_line(&mut line)? == 0 {
272            break;
273        }
274        let record: Value = serde_json::from_str(&line)
275            .with_context(|| format!("parse Codex session {}", path.display()))?;
276        if record.get("type").and_then(Value::as_str) != Some("session_meta") {
277            continue;
278        }
279        if !codex_source_is_interactive(record.pointer("/payload/source")) {
280            return Ok(None);
281        }
282        // Ephemeral Codex threads normally have no rollout path at all. Keep
283        // this defensive check so a future writer cannot expose one here.
284        if record
285            .pointer("/payload/ephemeral")
286            .and_then(Value::as_bool)
287            == Some(true)
288        {
289            return Ok(None);
290        }
291        // Codex ACP loads a rollout by its payload `id`, which is also the
292        // UUID embedded in the rollout filename. `session_id` can name a
293        // parent thread and therefore is not necessarily resumable itself.
294        let id = record
295            .pointer("/payload/id")
296            .or_else(|| record.pointer("/payload/session_id"))
297            .and_then(Value::as_str)
298            .filter(|id| !id.is_empty())
299            .map(ToOwned::to_owned);
300        if let Some(id) = id {
301            validate_id("Codex session", &id)?;
302            let cwd = record
303                .pointer("/payload/cwd")
304                .and_then(Value::as_str)
305                .filter(|cwd| !cwd.trim().is_empty())
306                .map(PathBuf::from)
307                .unwrap_or_default();
308            let git_branch = record
309                .pointer("/payload/git/branch")
310                .and_then(Value::as_str)
311                .filter(|branch| !branch.trim().is_empty())
312                .unwrap_or("HEAD")
313                .to_owned();
314            let history_mode = record
315                .pointer("/payload/history_mode")
316                .and_then(Value::as_str)
317                .map(parse_codex_history_mode)
318                .transpose()?
319                .unwrap_or(CodexHistoryMode::Legacy);
320            return Ok(Some(CodexSessionMetadata {
321                id,
322                cwd,
323                git_branch,
324                history_mode,
325            }));
326        }
327    }
328    Ok(None)
329}
330
331pub(super) fn parse_codex_history_mode(value: &str) -> Result<CodexHistoryMode> {
332    match value {
333        "legacy" => Ok(CodexHistoryMode::Legacy),
334        "paginated" => Ok(CodexHistoryMode::Paginated),
335        other => bail!("unsupported Codex history mode {other:?}"),
336    }
337}
338
339pub(super) fn codex_source_is_interactive(source: Option<&Value>) -> bool {
340    match source {
341        // Older rollouts predate the source field and came from the TUI.
342        None => true,
343        Some(Value::String(source)) => matches!(source.as_str(), "cli" | "vscode"),
344        // Structured sources identify subagents. Other unexpected shapes are
345        // not sessions offered by the normal interactive resume picker.
346        Some(_) => false,
347    }
348}
349
350pub(super) fn codex_native_titles(home: &Path) -> Result<BTreeMap<String, String>> {
351    let mut titles = BTreeMap::new();
352    // Older Codex stores use history only as their compact interactive-session
353    // index. Keep those IDs discoverable, but do not turn prompt text into a
354    // session name.
355    let history = home.join("history.jsonl");
356    if history.is_file() {
357        for line in BufReader::new(fs::File::open(&history)?).lines() {
358            let record: Value = serde_json::from_str(&line?)?;
359            if let (Some(session_id), Some(text)) = (
360                record.get("session_id").and_then(Value::as_str),
361                record.get("text").and_then(Value::as_str),
362            ) && !text.trim().is_empty()
363            {
364                titles
365                    .entry(session_id.to_owned())
366                    .or_insert_with(|| session_id.to_owned());
367            }
368        }
369    }
370    let index = home.join("session_index.jsonl");
371    if index.is_file() {
372        for line in BufReader::new(fs::File::open(&index)?).lines() {
373            let record: Value = serde_json::from_str(&line?)?;
374            if let (Some(session_id), Some(title)) = (
375                record.get("id").and_then(Value::as_str),
376                record.get("thread_name").and_then(Value::as_str),
377            ) && let Some(title) = normalize_session_title(title)
378            {
379                titles.insert(session_id.to_owned(), title);
380            }
381        }
382    }
383    Ok(titles)
384}
mj_controller/import/codex.rs

mj_controller/import/
codex.rs
