Skip to main content

mempal_runtime/cowork/
codex.rs

1//! Codex session reader.
2
3use crate::cowork::peek::{PeekError, PeekMessage, days_to_ymd, parse_rfc3339};
4use serde_json::Value;
5use std::fs;
6use std::io::{BufRead, BufReader};
7use std::path::{Path, PathBuf};
8use std::time::{SystemTime, UNIX_EPOCH};
9
10/// Read the first line of a Codex rollout jsonl and extract `payload.cwd`
11/// from the `session_meta` entry. Returns `None` if the file can't be
12/// opened or the first line isn't a valid `session_meta` with a `cwd` field.
13pub fn read_session_cwd(path: &Path) -> Option<String> {
14    let file = fs::File::open(path).ok()?;
15    let mut reader = BufReader::new(file);
16    let mut line = String::new();
17    reader.read_line(&mut line).ok()?;
18    let val: Value = serde_json::from_str(line.trim()).ok()?;
19    if val.get("type").and_then(|v| v.as_str()) != Some("session_meta") {
20        return None;
21    }
22    val.get("payload")?
23        .get("cwd")?
24        .as_str()
25        .map(|s| s.to_string())
26}
27
28/// Scan the **last 7 days** of the Codex sessions directory
29/// (`~/.codex/sessions/YYYY/MM/DD/`) for `rollout-*.jsonl` files whose
30/// `session_meta.payload.cwd` matches `target_cwd`. Returns the latest
31/// matching file by mtime.
32///
33/// The window is anchored at `SystemTime::now()`. Sessions older than 7
34/// days are not considered, per spec (`specs/p6-cowork-peek-and-decide.spec.md`
35/// Decisions line "扫最近 7 天的目录"). This keeps the scan bounded even
36/// when `~/.codex/sessions` has accumulated months of history.
37pub fn find_latest_session_for_cwd(
38    base: &Path,
39    target_cwd: &str,
40) -> Result<Option<(PathBuf, SystemTime)>, PeekError> {
41    let now_secs = SystemTime::now()
42        .duration_since(UNIX_EPOCH)
43        .map(|d| d.as_secs() as i64)
44        .unwrap_or(0);
45    find_latest_session_for_cwd_at(base, target_cwd, now_secs)
46}
47
48/// Testable variant of `find_latest_session_for_cwd` that accepts an
49/// explicit "now" in epoch seconds.
50///
51/// Scans `[UTC_today - 7, UTC_today + 1]` (9 date directories) to cover
52/// "last 7 local days" regardless of the user's timezone offset. Codex
53/// names session directories by the LOCAL wall-clock date, not UTC:
54///
55///   * For **positive** offsets (e.g. Asia/Shanghai +08:00), during
56///     early-morning hours the live session lives in `UTC_today + 1`.
57///   * For **negative** offsets (e.g. America/Los_Angeles -08:00),
58///     during late-evening hours the live session lives in `UTC_today - 1`,
59///     so 7 local days back is `UTC_today - 7`.
60///
61/// The 9-day UTC window is the smallest that provably covers both cases
62/// without parsing timezone data. It replaces an earlier 7-day UTC window
63/// that silently missed live sessions in UTC+ timezones during the
64/// ~8-hour post-UTC-midnight local-morning period (caught by Codex review).
65pub(crate) fn find_latest_session_for_cwd_at(
66    base: &Path,
67    target_cwd: &str,
68    now_epoch_secs: i64,
69) -> Result<Option<(PathBuf, SystemTime)>, PeekError> {
70    let today_days = now_epoch_secs.div_euclid(86400);
71    let mut candidates: Vec<(PathBuf, SystemTime)> = Vec::new();
72
73    for offset in -1i64..=7i64 {
74        let (year, month, day) = days_to_ymd(today_days - offset);
75        let day_dir = base
76            .join(format!("{year:04}"))
77            .join(format!("{month:02}"))
78            .join(format!("{day:02}"));
79        let Ok(entries) = fs::read_dir(&day_dir) else {
80            continue;
81        };
82        for entry in entries.filter_map(|e| e.ok()) {
83            let path = entry.path();
84            if !path.is_file() {
85                continue;
86            }
87            if path.extension().and_then(|s| s.to_str()) != Some("jsonl") {
88                continue;
89            }
90            let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
91            if !name.starts_with("rollout-") {
92                continue;
93            }
94
95            if let Some(cwd) = read_session_cwd(&path) {
96                if cwd == target_cwd {
97                    if let Ok(metadata) = entry.metadata() {
98                        if let Ok(mtime) = metadata.modified() {
99                            candidates.push((path.clone(), mtime));
100                        }
101                    }
102                }
103            }
104        }
105    }
106
107    Ok(candidates.into_iter().max_by_key(|(_, m)| *m))
108}
109
110/// Parse a Codex rollout jsonl. Returns `(messages, truncated)`.
111///
112/// Only `type: "response_item"` entries with `payload.type: "message"` and
113/// `payload.role` in {user, assistant} are processed. Text is concatenated
114/// from all blocks in `payload.content[]` that have a `text` field
115/// (block types include `input_text`, `output_text`, etc.). Filters out
116/// `reasoning` payloads and `event_msg` entries.
117pub fn parse_codex_jsonl(
118    path: &Path,
119    since: Option<&str>,
120    limit: usize,
121) -> Result<(Vec<PeekMessage>, bool), PeekError> {
122    // Pre-parse the `since` cutoff once; see claude.rs for rationale.
123    let since_cutoff: Option<i64> = match since {
124        Some(raw) => Some(parse_rfc3339(raw).ok_or_else(|| {
125            PeekError::Parse(format!("invalid `since` RFC3339 timestamp: {raw}"))
126        })?),
127        None => None,
128    };
129
130    let file = fs::File::open(path)?;
131    let reader = BufReader::new(file);
132    let mut all: Vec<PeekMessage> = Vec::new();
133
134    for line in reader.lines() {
135        let line = line?;
136        if line.trim().is_empty() {
137            continue;
138        }
139        let val: Value = match serde_json::from_str(&line) {
140            Ok(v) => v,
141            Err(_) => continue,
142        };
143
144        if val.get("type").and_then(|v| v.as_str()) != Some("response_item") {
145            continue;
146        }
147        let Some(payload) = val.get("payload") else {
148            continue;
149        };
150        if payload.get("type").and_then(|v| v.as_str()) != Some("message") {
151            continue;
152        }
153        let role = payload.get("role").and_then(|v| v.as_str()).unwrap_or("");
154        if role != "user" && role != "assistant" {
155            continue;
156        }
157
158        let text = match payload.get("content") {
159            Some(Value::Array(blocks)) => {
160                let parts: Vec<String> = blocks
161                    .iter()
162                    .filter_map(|b| b.get("text").and_then(|v| v.as_str()).map(String::from))
163                    .collect();
164                parts.join("\n")
165            }
166            Some(Value::String(s)) => s.clone(),
167            _ => continue,
168        };
169        if text.is_empty() {
170            continue;
171        }
172
173        let at = val
174            .get("timestamp")
175            .and_then(|v| v.as_str())
176            .unwrap_or("")
177            .to_string();
178
179        if let Some(cutoff) = since_cutoff {
180            // Same safety rule as Claude adapter: unparseable msg timestamps
181            // are kept rather than silently dropped.
182            if let Some(msg_ts) = parse_rfc3339(&at) {
183                if msg_ts <= cutoff {
184                    continue;
185                }
186            }
187        }
188
189        all.push(PeekMessage {
190            role: role.to_string(),
191            at,
192            text,
193        });
194    }
195
196    let total = all.len();
197    let truncated = total > limit;
198    let start = total.saturating_sub(limit);
199    let tail = all.split_off(start);
200    Ok((tail, truncated))
201}
202
203#[cfg(test)]
204mod tests {
205    use super::*;
206    use std::path::PathBuf;
207
208    fn fixture_path(relative: &str) -> PathBuf {
209        PathBuf::from(env!("CARGO_MANIFEST_DIR"))
210            .join("../..")
211            .join("tests/fixtures/cowork")
212            .join(relative)
213    }
214
215    fn codex_fixture_dir() -> PathBuf {
216        fixture_path("codex")
217    }
218
219    #[test]
220    fn reads_session_meta_cwd_from_first_line() {
221        let fixture = fixture_path("codex/2026/04/13/rollout-2026-04-13T12-00-00-fake.jsonl");
222        let cwd = read_session_cwd(&fixture).expect("read cwd");
223        assert_eq!(cwd, "/tmp/fake-project");
224    }
225
226    #[test]
227    fn parses_codex_messages_filtering_event_and_reasoning() {
228        let fixture = fixture_path("codex/2026/04/13/rollout-2026-04-13T12-00-00-fake.jsonl");
229        let (messages, truncated) = parse_codex_jsonl(&fixture, None, 30).expect("parse");
230        // 8 lines total; 4 are valid response_item message entries
231        // (2 user + 2 assistant), rest are session_meta / event_msg / reasoning.
232        assert_eq!(messages.len(), 4);
233        assert_eq!(messages[0].text, "codex: hello");
234        assert_eq!(messages[1].text, "codex: hi back");
235        assert_eq!(messages[2].text, "codex: continue");
236        assert_eq!(messages[3].text, "codex: continuing");
237        assert!(messages[0].at < messages[3].at);
238        assert!(!truncated);
239    }
240
241    #[test]
242    fn honors_limit_by_tail_and_sets_truncated_codex() {
243        let fixture = fixture_path("codex/2026/04/13/rollout-2026-04-13T12-00-00-fake.jsonl");
244        let (messages, truncated) = parse_codex_jsonl(&fixture, None, 2).expect("parse");
245        assert_eq!(messages.len(), 2);
246        assert_eq!(messages[0].text, "codex: continue");
247        assert_eq!(messages[1].text, "codex: continuing");
248        assert!(truncated);
249    }
250
251    // "Now" for the deterministic unit tests: 2026-04-13T12:00:00Z.
252    // Pick this so the 7-day window covers 2026-04-07..2026-04-13
253    // (both existing fixtures 2026/04/12 and 2026/04/13 are in range).
254    const FIXED_NOW_EPOCH_2026_04_13: i64 = {
255        // days_from_civil(2026, 4, 13) = 20556 → * 86400 + 12h
256        20556_i64 * 86400 + 12 * 3600
257    };
258
259    #[test]
260    fn walks_codex_dir_filtering_by_cwd() {
261        let base = codex_fixture_dir();
262        let result =
263            find_latest_session_for_cwd_at(&base, "/tmp/fake-project", FIXED_NOW_EPOCH_2026_04_13)
264                .expect("find session");
265        assert!(result.is_some());
266        let (path, _mtime) = result.unwrap();
267        assert!(
268            path.to_string_lossy()
269                .contains("rollout-2026-04-13T12-00-00-fake.jsonl")
270        );
271    }
272
273    #[test]
274    fn walks_codex_dir_excludes_other_projects() {
275        let base = codex_fixture_dir();
276        let result =
277            find_latest_session_for_cwd_at(&base, "/tmp/fake-project", FIXED_NOW_EPOCH_2026_04_13)
278                .expect("find session");
279        let path = result.unwrap().0;
280        assert!(!path.to_string_lossy().contains("otherproject"));
281    }
282
283    #[test]
284    fn scan_window_covers_utc_tomorrow_for_positive_offset_tz() {
285        // Regression for the UTC-vs-local-date bug caught by Codex review:
286        // Codex session directories are named by LOCAL wall-clock date,
287        // not UTC. In Asia/Shanghai (UTC+8) at 05:00 local on 2026-04-13,
288        // the current session's directory is 2026/04/13, but UTC is still
289        // 2026-04-12T21:00. A naive "days since epoch" scan anchored at
290        // UTC_today would scan 2026/04/06..2026/04/12 and MISS the live
291        // 2026/04/13 directory entirely during the ~8 hour early-morning
292        // window in UTC+ timezones.
293        //
294        // The fix widens the scan to [UTC_today - 7, UTC_today + 1] so
295        // "UTC tomorrow" (which is "local today" for positive offsets) is
296        // always included.
297        let base = codex_fixture_dir();
298        // 2026-04-12T21:00Z = 2026-04-13T05:00 in Asia/Shanghai.
299        // days_from_civil(2026, 4, 12) = 20555.
300        let now_epoch = 20555_i64 * 86400 + 21 * 3600;
301        let result = find_latest_session_for_cwd_at(&base, "/tmp/fake-project", now_epoch)
302            .expect("find session");
303        assert!(
304            result.is_some(),
305            "expected to find the 2026/04/13 fixture at UTC 2026-04-12T21:00 \
306             (= 05:00 local in +08:00)"
307        );
308        let (path, _) = result.unwrap();
309        assert!(
310            path.to_string_lossy().contains("2026/04/13"),
311            "expected the 04/13 fixture (local today), got {path:?}"
312        );
313    }
314
315    #[test]
316    fn scan_window_excludes_sessions_outside_the_9_day_span() {
317        // With the new [UTC_today - 7, UTC_today + 1] window, a fixture
318        // at 2026/04/13 falls OUT of the window only when UTC_today is
319        // advanced past 2026-04-21 (2026-04-13 would be UTC_today - 9
320        // on 2026-04-22, i.e. one day older than the window floor of
321        // UTC_today - 7). Use UTC_today = 2026-04-22:
322        //   window = [2026-04-15, 2026-04-23]
323        //   fixture 2026-04-13 is 2 days OLDER than the floor → excluded.
324        let base = codex_fixture_dir();
325        // days_from_civil(2026, 4, 22) = 20565
326        let now_epoch = 20565_i64 * 86400;
327        let result = find_latest_session_for_cwd_at(&base, "/tmp/fake-project", now_epoch)
328            .expect("find session");
329        assert!(
330            result.is_none(),
331            "04/13 fixture must fall outside the 04/15..04/23 window, got {result:?}"
332        );
333    }
334}