agent_top_core/harness/
codex.rs

1//! OpenAI Codex CLI: `~/.codex/sessions/YYYY/MM/DD/rollout-<ts>-<id>.jsonl`.
2//!
3//! Format notes (verified on Codex CLI 0.149, 2026-09-03):
4//! * The first line is `session_meta` with `payload.cwd`, `payload.id`,
5//!   `payload.cli_version` and `payload.originator`.
6//! * `event_msg` / `token_count` carries `info.total_token_usage`, which is
7//!   cumulative for the session; `info` is null on rate-limit-only events.
8//!   `input_tokens` includes `cached_input_tokens`.
9//! * `task_started` / `task_complete` / `turn_aborted` bracket a turn.
10//! * `response_item` with `payload.type` `function_call` or
11//!   `custom_tool_call` is one tool call; the matching `*_output` item
12//!   carries the same `payload.call_id`, and the two lines' timestamps
13//!   bracket the call. That pairing is the trace.
14//! * `task_started` and `task_complete` bracket a turn span. An inference
15//!   span runs from a user `message` item or a `*_output` item to the next
16//!   thing the model produced: a call, a `reasoning` item, a
17//!   `web_search_call`, or an assistant `message`.
18//! * `response_item` `web_search_call` is one server-side web search.
19//!
20//! Codex model prices are not in the static table, so cost is reported as
21//! unpriced tokens.
22
23use super::{REFRESH_BUDGET_BYTES, SessionSummary, SessionTracker, SpanRetention, parse_rfc3339_utc};
24use crate::jsonl::TailReader;
25use crate::model::{Activity, Harness, SpanKind, TokenUsage};
26use crate::pricing::{self, Table};
27use serde_json::Value;
28use std::path::{Path, PathBuf};
29use std::time::SystemTime;
30
31pub fn codex_dir() -> Option<PathBuf> {
32    if let Some(d) = std::env::var_os("CODEX_HOME") {
33        return Some(PathBuf::from(d));
34    }
35    std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".codex"))
36}
37
38pub fn sessions_dir() -> Option<PathBuf> {
39    codex_dir().map(|d| d.join("sessions"))
40}
41
42/// Rollout files modified after `since`. Walks `YYYY/MM/DD` and prunes by
43/// directory mtime so the walk stays cheap on a long history.
44pub fn recent_rollouts(since: SystemTime) -> Vec<PathBuf> {
45    let Some(root) = sessions_dir() else { return Vec::new() };
46    let mut out = Vec::new();
47    walk(&root, 0, since, &mut out);
48    out
49}
50
51fn walk(dir: &Path, depth: usize, since: SystemTime, out: &mut Vec<PathBuf>) {
52    let Ok(rd) = std::fs::read_dir(dir) else { return };
53    for e in rd.flatten() {
54        let p = e.path();
55        let Ok(md) = e.metadata() else { continue };
56        if md.is_dir() {
57            if depth < 3 && md.modified().map(|m| m >= since).unwrap_or(true) {
58                walk(&p, depth + 1, since, out);
59            }
60        } else if p.extension().and_then(|x| x.to_str()) == Some("jsonl") && md.modified().map(|m| m >= since).unwrap_or(false) {
61            out.push(p);
62        }
63    }
64}
65
66/// Cheap header read: cwd and start time from the first line only.
67pub fn read_meta(path: &Path) -> Option<(PathBuf, SystemTime)> {
68    use std::io::{BufRead, BufReader};
69    let f = std::fs::File::open(path).ok()?;
70    let mut first = String::new();
71    BufReader::new(f).read_line(&mut first).ok()?;
72    let v: Value = serde_json::from_str(&first).ok()?;
73    if v.get("type").and_then(Value::as_str) != Some("session_meta") {
74        return None;
75    }
76    let cwd = v.pointer("/payload/cwd").and_then(Value::as_str).map(PathBuf::from)?;
77    let ts = v.get("timestamp").and_then(Value::as_str).and_then(parse_rfc3339_utc)?;
78    Some((cwd, ts))
79}
80
81pub struct CodexTranscript {
82    reader: TailReader,
83    prices: &'static Table,
84    summary: SessionSummary,
85    /// Counters naming the turn and inference spans, and the ids of the ones
86    /// currently being extended.
87    turns: u64,
88    inferences: u64,
89    turn: Option<String>,
90    inference: Option<String>,
91}
92
93impl CodexTranscript {
94    pub fn new(path: impl Into<PathBuf>) -> Self {
95        CodexTranscript {
96            reader: TailReader::new(path),
97            prices: pricing::table(),
98            summary: SessionSummary { harness: Some(Harness::Codex), ..Default::default() },
99            turns: 0,
100            inferences: 0,
101            turn: None,
102            inference: None,
103        }
104    }
105
106    /// Something was submitted to the model. One inference at a time: a
107    /// developer message followed by a user message is one submission.
108    fn begin_inference(&mut self, ts: SystemTime) {
109        if self.summary.spans.open_of_kind(SpanKind::Inference).is_some() {
110            return;
111        }
112        // A turn that ended without the model replying (aborted) leaves the
113        // previous inference open; it produced nothing, so it goes.
114        if let Some(id) = self.inference.take() {
115            self.summary.spans.discard_open(&id);
116        }
117        self.inferences += 1;
118        let id = format!("inference:{}", self.inferences);
119        self.summary.spans.open_kind(id.clone(), "inference".into(), ts, false, SpanKind::Inference);
120        self.inference = Some(id);
121    }
122
123    /// The model produced something: the inference in progress ends here.
124    fn end_inference(&mut self, ts: SystemTime) {
125        if let Some(id) = self.inference.take() {
126            self.summary.spans.end_at(&id, ts);
127        }
128    }
129
130    /// See `ClaudeTranscript::with_prices`.
131    pub fn with_prices(mut self, prices: &'static Table) -> Self {
132        self.prices = prices;
133        self
134    }
135
136    /// Keep every span instead of the newest `MAX_SPANS`. See `SpanRetention`.
137    pub fn with_spans(mut self, retention: SpanRetention) -> Self {
138        self.summary.spans = retention.log();
139        self
140    }
141
142    fn ingest(&mut self, line: &str) {
143        let Ok(v) = serde_json::from_str::<Value>(line) else { return };
144        let ts = v.get("timestamp").and_then(Value::as_str).and_then(parse_rfc3339_utc);
145        if let Some(ts) = ts {
146            if self.summary.started_at.is_none() {
147                self.summary.started_at = Some(ts);
148            }
149            self.summary.last_activity = Some(ts);
150        }
151        let kind = v.get("type").and_then(Value::as_str).unwrap_or("");
152        let payload = v.get("payload");
153        let ptype = payload.and_then(|p| p.get("type")).and_then(Value::as_str).unwrap_or("");
154        match kind {
155            "session_meta" => {
156                if let Some(p) = payload {
157                    self.summary.session_id = p.get("id").or(p.get("session_id")).and_then(Value::as_str).map(str::to_string);
158                    self.summary.cwd = p.get("cwd").and_then(Value::as_str).map(PathBuf::from);
159                    self.summary.harness_version = p.get("cli_version").and_then(Value::as_str).map(str::to_string);
160                }
161            }
162            "turn_context" => {
163                if let Some(m) = payload.and_then(|p| p.get("model")).and_then(Value::as_str) {
164                    self.summary.model = Some(m.to_string());
165                }
166            }
167            "event_msg" => match ptype {
168                "token_count" => {
169                    if let Some(total) = payload.and_then(|p| p.pointer("/info/total_token_usage")) {
170                        let g = |k: &str| total.get(k).and_then(Value::as_u64).unwrap_or(0);
171                        self.summary.health.usage_records += 1;
172                        if g("input_tokens") + g("output_tokens") + g("cached_input_tokens") == 0 {
173                            self.summary.health.empty_usage_records += 1;
174                        }
175                        let cached = g("cached_input_tokens");
176                        let usage = TokenUsage {
177                            input: g("input_tokens").saturating_sub(cached),
178                            cache_read: cached,
179                            output: g("output_tokens"),
180                            ..Default::default()
181                        };
182                        self.summary.usage = usage;
183                        let price = self.summary.model.as_deref().and_then(|m| self.prices.lookup(m));
184                        match price {
185                            Some(p) => {
186                                self.summary.cost_usd = p.cost(&usage);
187                                self.summary.unpriced_tokens = 0;
188                            }
189                            None => {
190                                self.summary.cost_usd = 0.0;
191                                self.summary.unpriced_tokens = usage.total();
192                            }
193                        }
194                    }
195                }
196                "task_started" => {
197                    self.summary.activity = Activity::Working;
198                    if let Some(ts) = ts {
199                        self.turns += 1;
200                        let id = format!("turn:{}", self.turns);
201                        self.summary.spans.open_kind(id.clone(), "turn".into(), ts, false, SpanKind::Turn);
202                        self.turn = Some(id);
203                    }
204                }
205                "user_message" => self.summary.activity = Activity::Working,
206                "task_complete" | "turn_aborted" | "error" => {
207                    self.summary.activity = Activity::Waiting;
208                    if let (Some(ts), Some(id)) = (ts, self.turn.take()) {
209                        self.summary.spans.end_at(&id, ts);
210                    }
211                    if let Some(id) = self.inference.take() {
212                        self.summary.spans.discard_open(&id);
213                    }
214                }
215                _ => {}
216            },
217            "response_item" => match ptype {
218                "function_call" | "custom_tool_call" | "local_shell_call" => {
219                    self.summary.tool_calls += 1;
220                    if let (Some(ts), Some(p)) = (ts, payload) {
221                        self.end_inference(ts);
222                        let id = call_id(p);
223                        let name = p.get("name").and_then(Value::as_str).unwrap_or(ptype);
224                        self.summary.spans.open(id, name.to_string(), ts, false);
225                    }
226                }
227                "function_call_output" | "custom_tool_call_output" | "local_shell_call_output" => {
228                    if let (Some(ts), Some(p)) = (ts, payload) {
229                        // Codex reports the result as an opaque string, and
230                        // agent-top does not read tool output, so a failed call
231                        // is not distinguishable from a successful one here.
232                        self.summary.spans.close(&call_id(p), ts, false);
233                        self.begin_inference(ts);
234                    }
235                }
236                // A server-side web search: billed per search by OpenAI, but
237                // at a rate this table does not carry, so counted only.
238                "web_search_call" => {
239                    self.summary.web_searches += 1;
240                    if let Some(ts) = ts {
241                        self.end_inference(ts);
242                    }
243                }
244                "reasoning" => {
245                    if let Some(ts) = ts {
246                        self.end_inference(ts);
247                    }
248                }
249                "message" => match payload.and_then(|p| p.get("role")).and_then(Value::as_str) {
250                    Some("assistant") => {
251                        self.summary.turns += 1;
252                        self.summary.health.billable_messages += 1;
253                        if let Some(ts) = ts {
254                            self.end_inference(ts);
255                        }
256                    }
257                    Some("user") => {
258                        if let Some(ts) = ts {
259                            self.begin_inference(ts);
260                        }
261                    }
262                    _ => {}
263                },
264                _ => {}
265            },
266            _ => {}
267        }
268    }
269}
270
271/// `call_id` on function calls, `id` on the shell-call variants.
272fn call_id(payload: &Value) -> String {
273    payload.get("call_id").or_else(|| payload.get("id")).and_then(Value::as_str).unwrap_or_default().to_string()
274}
275
276impl SessionTracker for CodexTranscript {
277    fn refresh(&mut self) -> anyhow::Result<bool> {
278        let (lines, more) = self.reader.read_new_lines(REFRESH_BUDGET_BYTES)?;
279        for l in &lines {
280            self.ingest(l);
281        }
282        Ok(more)
283    }
284
285    fn summary(&self) -> &SessionSummary {
286        &self.summary
287    }
288
289    fn path(&self) -> &Path {
290        self.reader.path()
291    }
292}
293
294#[cfg(test)]
295mod tests {
296    use super::*;
297    use std::io::Write;
298
299    #[test]
300    fn reads_cumulative_usage_and_state() {
301        let dir = std::env::temp_dir().join(format!("agent-top-codex-{}", std::process::id()));
302        std::fs::create_dir_all(&dir).unwrap();
303        let path = dir.join("rollout.jsonl");
304        let mut f = std::fs::File::create(&path).unwrap();
305        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:20.787Z","type":"session_meta","payload":{{"id":"01a0","cwd":"/tmp/p","cli_version":"0.149.1"}}}}"#).unwrap();
306        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:21.000Z","type":"turn_context","payload":{{"model":"gpt-5-codex"}}}}"#).unwrap();
307        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:22.000Z","type":"event_msg","payload":{{"type":"task_started"}}}}"#).unwrap();
308        writeln!(
309            f,
310            r#"{{"timestamp":"2026-08-28T08:53:23.000Z","type":"response_item","payload":{{"type":"function_call","call_id":"call_1","name":"shell"}}}}"#
311        )
312        .unwrap();
313        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:24.000Z","type":"event_msg","payload":{{"type":"token_count","info":{{"total_token_usage":{{"input_tokens":14778,"cached_input_tokens":12672,"output_tokens":241,"total_tokens":15019}}}}}}}}"#).unwrap();
314        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:25.000Z","type":"event_msg","payload":{{"type":"token_count","info":null}}}}"#)
315            .unwrap();
316        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:25.500Z","type":"response_item","payload":{{"type":"web_search_call","status":"completed"}}}}"#).unwrap();
317        let mut t = CodexTranscript::new(&path);
318        t.refresh().unwrap();
319        let s = t.summary();
320        assert_eq!(s.session_id.as_deref(), Some("01a0"));
321        assert_eq!(s.model.as_deref(), Some("gpt-5-codex"));
322        assert_eq!(s.usage.input, 14778 - 12672);
323        assert_eq!(s.usage.cache_read, 12672);
324        assert_eq!(s.usage.total(), 15019);
325        assert_eq!(s.unpriced_tokens, 15019);
326        assert_eq!(s.tool_calls, 1);
327        assert_eq!(s.activity, Activity::Working);
328        assert_eq!(read_meta(&path).unwrap().0, PathBuf::from("/tmp/p"));
329        assert_eq!(s.web_searches, 1);
330        let tools: Vec<_> = s.spans.iter().filter(|sp| sp.kind == SpanKind::Tool).collect();
331        assert_eq!(tools.len(), 1);
332        assert_eq!(tools[0].name, "shell");
333        assert!(tools[0].is_open(), "no output item yet");
334        let turns: Vec<_> = s.spans.iter().filter(|sp| sp.kind == SpanKind::Turn).collect();
335        assert_eq!(turns.len(), 1);
336        assert!(turns[0].is_open(), "task_started with no task_complete");
337        let _ = std::fs::remove_dir_all(&dir);
338    }
339
340    #[test]
341    fn pairs_calls_with_their_outputs() {
342        let dir = std::env::temp_dir().join(format!("agent-top-codex-spans-{}", std::process::id()));
343        std::fs::create_dir_all(&dir).unwrap();
344        let path = dir.join("rollout.jsonl");
345        let mut f = std::fs::File::create(&path).unwrap();
346        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:23.000Z","type":"response_item","payload":{{"type":"function_call","call_id":"call_1","name":"exec_command"}}}}"#).unwrap();
347        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:23.100Z","type":"response_item","payload":{{"type":"custom_tool_call","call_id":"call_2","name":"apply_patch"}}}}"#).unwrap();
348        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:24.000Z","type":"response_item","payload":{{"type":"function_call_output","call_id":"call_1","output":"ok"}}}}"#).unwrap();
349        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:26.100Z","type":"response_item","payload":{{"type":"custom_tool_call_output","call_id":"call_2","output":"ok"}}}}"#).unwrap();
350        // The model answers the outputs 1.5 s after the last one, then the turn completes.
351        writeln!(
352            f,
353            r#"{{"timestamp":"2026-08-28T08:53:27.600Z","type":"response_item","payload":{{"type":"message","role":"assistant"}}}}"#
354        )
355        .unwrap();
356        writeln!(f, r#"{{"timestamp":"2026-08-28T08:53:27.700Z","type":"event_msg","payload":{{"type":"task_complete"}}}}"#).unwrap();
357        let mut t = CodexTranscript::new(&path);
358        t.refresh().unwrap();
359        let all = t.summary().spans.to_vec();
360        let spans: Vec<_> = all.iter().filter(|sp| sp.kind == SpanKind::Tool).collect();
361        assert_eq!(spans.len(), 2);
362        assert_eq!(spans[0].name, "exec_command");
363        assert_eq!(spans[0].duration_ms, Some(1_000));
364        assert_eq!(spans[1].name, "apply_patch");
365        assert_eq!(spans[1].duration_ms, Some(3_000));
366        assert_eq!(t.summary().tool_calls, 2);
367        // One inference: opened by the first output at :24, not re-opened by the
368        // second at :26.1, ended by the assistant message at :27.6.
369        let inf: Vec<_> = all.iter().filter(|sp| sp.kind == SpanKind::Inference).collect();
370        assert_eq!(inf.len(), 1);
371        assert_eq!(inf[0].duration_ms, Some(3_600));
372        assert!(all.iter().all(|sp| sp.kind != SpanKind::Turn), "no task_started in this file");
373        let _ = std::fs::remove_dir_all(&dir);
374    }
375}
agent_top_core/harness/codex.rs

agent_top_core/harness/
codex.rs
