use std::collections::BTreeMap;
use std::fs::File;
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use serde::Deserialize;
use crate::adapters::Adapter;
use crate::error::CoreError;
use crate::model::{Event, EventKind, Session, SessionRef, ToolCall, Usage};
pub struct Codex;
fn codex_roots() -> Vec<PathBuf> {
match std::env::var("CODEX_HOME") {
Ok(val) if !val.trim().is_empty() => val
.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.map(PathBuf::from)
.collect(),
_ => directories::BaseDirs::new()
.map(|b| vec![b.home_dir().join(".codex")])
.unwrap_or_default(),
}
}
fn session_dirs(root: &Path) -> [PathBuf; 2] {
[root.join("sessions"), root.join("archived_sessions")]
}
impl Adapter for Codex {
fn id(&self) -> &'static str {
"codex"
}
fn detect(&self) -> bool {
codex_roots()
.iter()
.any(|r| session_dirs(r).iter().any(|d| d.is_dir()))
}
fn discover(&self) -> anyhow::Result<Vec<SessionRef>> {
let roots = codex_roots();
if roots.is_empty() {
anyhow::bail!("could not determine the home directory (set CODEX_HOME)");
}
let mut refs = Vec::new();
for root in &roots {
for dir in session_dirs(root) {
if !dir.is_dir() {
continue;
}
for entry in walkdir::WalkDir::new(&dir).follow_links(false) {
let entry = match entry {
Ok(e) => e,
Err(e) => {
tracing::debug!("skipping unreadable directory entry: {e}");
continue;
}
};
if !entry.file_type().is_file()
|| entry.path().extension().and_then(|x| x.to_str()) != Some("jsonl")
{
continue;
}
let meta = entry.metadata().ok();
refs.push(SessionRef {
path: entry.path().to_path_buf(),
agent: self.id().to_string(),
project: None,
size_bytes: meta.as_ref().map_or(0, |m| m.len()),
modified: meta
.and_then(|m| m.modified().ok())
.and_then(|t| jiff::Timestamp::try_from(t).ok()),
});
}
}
}
refs.sort_by_key(|r| std::cmp::Reverse(r.modified));
Ok(refs)
}
fn parse(&self, r: &SessionRef) -> anyhow::Result<Session> {
let file = File::open(&r.path).map_err(|source| CoreError::Io {
path: r.path.clone(),
source,
})?;
let mut session = Session {
id: session_id_for(&r.path),
agent: self.id().to_string(),
project: r.project.clone(),
model: None,
parent_session: None,
started_at: None,
ended_at: None,
events: Vec::new(),
sub_agents: Vec::new(),
skipped_lines: 0,
};
let mut current_model: Option<String> = None;
let mut model_counts: BTreeMap<String, u64> = BTreeMap::new();
for (idx, line) in BufReader::new(file).lines().enumerate() {
let line = match line {
Ok(l) => l,
Err(e) => {
session.skipped_lines += 1;
tracing::debug!("{}:{}: unreadable line: {e}", r.path.display(), idx + 1);
continue;
}
};
if line.trim().is_empty() {
continue;
}
let raw: RawLine = match serde_json::from_str(&line) {
Ok(v) => v,
Err(e) => {
session.skipped_lines += 1;
tracing::debug!("{}:{}: unparseable JSON: {e}", r.path.display(), idx + 1);
continue;
}
};
let ts = parse_ts(&raw.timestamp);
match raw.kind.as_deref() {
Some("session_meta") => apply_session_meta(&raw, &mut session, &mut current_model),
Some("turn_context") => {
if let Some(model) = raw.payload.model.as_deref() {
if !model.is_empty() {
current_model = Some(model.to_string());
}
}
}
Some("response_item") => push_response_item(&raw, ts, &mut session),
Some("event_msg") => {
push_event_msg(&raw, ts, ¤t_model, &mut session, &mut model_counts)
}
other => {
session.skipped_lines += 1;
tracing::debug!(
"{}:{}: unknown line type {:?}",
r.path.display(),
idx + 1,
other
);
}
}
}
session.started_at = session.events.iter().filter_map(|e| e.ts).min();
session.ended_at = session.events.iter().filter_map(|e| e.ts).max();
session.model = model_counts
.into_iter()
.max_by_key(|(_, count)| *count)
.map(|(model, _)| model)
.or(current_model);
Ok(session)
}
}
fn session_id_for(path: &Path) -> String {
path.file_stem()
.map(|s| s.to_string_lossy().into_owned())
.unwrap_or_else(|| path.to_string_lossy().into_owned())
}
fn parse_ts(raw: &Option<String>) -> Option<jiff::Timestamp> {
raw.as_deref().and_then(|t| t.parse().ok())
}
fn snippet(text: &str) -> Option<String> {
let first = text.lines().find(|l| !l.trim().is_empty())?;
let s: String = first.trim().chars().take(80).collect();
(!s.is_empty()).then_some(s)
}
fn message_text(content: &serde_json::Value) -> String {
if let Some(text) = content.as_str() {
return text.to_string();
}
let Some(blocks) = content.as_array() else {
return String::new();
};
let mut out = String::new();
for block in blocks {
if let Some(text) = block.get("text").and_then(|t| t.as_str()) {
if !out.is_empty() {
out.push('\n');
}
out.push_str(text);
}
}
out
}
fn apply_session_meta(raw: &RawLine, session: &mut Session, current_model: &mut Option<String>) {
if session.project.is_none() {
if let Some(cwd) = raw.payload.cwd.as_deref() {
if !cwd.is_empty() {
session.project = Some(cwd.to_string());
}
}
}
if current_model.is_none() {
if let Some(model) = raw.payload.model.as_deref() {
if !model.is_empty() {
*current_model = Some(model.to_string());
}
}
}
}
fn push_response_item(raw: &RawLine, ts: Option<jiff::Timestamp>, session: &mut Session) {
match raw.payload.item_type.as_deref() {
Some("message") => {
let text = message_text(&raw.payload.content);
let kind = match raw.payload.role.as_deref() {
Some("assistant") => EventKind::Assistant,
_ => EventKind::User,
};
session.events.push(Event {
kind,
ts,
request_id: None,
model: None,
usage: None,
tool_calls: Vec::new(),
sidechain: false,
content_summary: snippet(&text),
content_chars: text.chars().count() as u64,
thinking_chars: 0,
has_thinking: false,
tool_use_id: None,
attachment_kind: None,
item_count: 0,
});
}
Some("function_call") => {
let name = raw
.payload
.name
.clone()
.unwrap_or_else(|| "unknown".to_string());
let id = raw.payload.call_id.clone().unwrap_or_default();
let input_bytes = raw
.payload
.arguments
.as_ref()
.map_or(0, |a| a.chars().count() as u64);
session.events.push(Event {
kind: EventKind::ToolCall,
ts,
request_id: None,
model: None,
usage: None,
tool_calls: vec![ToolCall {
server: ToolCall::server_from_name(&name),
id,
name,
input_bytes,
}],
sidechain: false,
content_summary: None,
content_chars: input_bytes,
thinking_chars: 0,
has_thinking: false,
tool_use_id: None,
attachment_kind: None,
item_count: 0,
});
}
_ => {}
}
}
fn push_event_msg(
raw: &RawLine,
ts: Option<jiff::Timestamp>,
current_model: &Option<String>,
session: &mut Session,
model_counts: &mut BTreeMap<String, u64>,
) {
match raw.payload.item_type.as_deref() {
Some("token_count") => {
let Some(info) = raw.payload.info.as_ref() else {
session.skipped_lines += 1;
return;
};
let Some(last) = info.last_token_usage.as_ref() else {
return;
};
let usage = map_last_usage(last);
let has_thinking = last.reasoning_output_tokens.unwrap_or(0) > 0;
if let Some(model) = current_model {
*model_counts.entry(model.clone()).or_default() += 1;
}
session.events.push(Event {
kind: EventKind::Assistant,
ts,
request_id: None,
model: current_model.clone(),
usage: Some(usage),
tool_calls: Vec::new(),
sidechain: false,
content_summary: None,
content_chars: 0,
thinking_chars: 0,
has_thinking,
tool_use_id: None,
attachment_kind: None,
item_count: 0,
});
}
Some("context_compacted") => session.events.push(Event {
kind: EventKind::Compaction,
ts,
request_id: None,
model: None,
usage: None,
tool_calls: Vec::new(),
sidechain: false,
content_summary: Some("context compacted".to_string()),
content_chars: 0,
thinking_chars: 0,
has_thinking: false,
tool_use_id: None,
attachment_kind: None,
item_count: 0,
}),
Some("mcp_tool_call_end") => {
if let Some(call) = mcp_tool_call(raw) {
session.events.push(Event {
kind: EventKind::ToolCall,
ts,
request_id: None,
model: None,
usage: None,
tool_calls: vec![call],
sidechain: false,
content_summary: None,
content_chars: 0,
thinking_chars: 0,
has_thinking: false,
tool_use_id: None,
attachment_kind: None,
item_count: 0,
});
}
}
_ => {}
}
}
fn mcp_tool_call(raw: &RawLine) -> Option<ToolCall> {
let inv = raw.payload.invocation.as_ref();
let server = inv.and_then(|i| i.server.clone());
let tool = inv
.and_then(|i| i.tool.clone())
.or_else(|| raw.payload.tool.clone());
let name = match (&server, &tool) {
(Some(s), Some(t)) => format!("mcp__{s}__{t}"),
(None, Some(t)) => t.clone(),
_ => return None,
};
Some(ToolCall {
server: server.or_else(|| ToolCall::server_from_name(&name)),
id: raw.payload.call_id.clone().unwrap_or_default(),
name,
input_bytes: 0,
})
}
fn map_last_usage(last: &RawTokenUsage) -> Usage {
let cached = last.cached_input_tokens.unwrap_or(0);
let reasoning = last.reasoning_output_tokens.unwrap_or(0);
Usage {
input: last.input_tokens.map(|i| i.saturating_sub(cached)),
output: last.output_tokens.map(|o| o.saturating_sub(reasoning)),
cache_creation: None,
cache_creation_5m: None,
cache_creation_1h: None,
cache_read: last.cached_input_tokens,
thinking: last.reasoning_output_tokens,
}
}
#[derive(Deserialize)]
struct RawLine {
#[serde(rename = "type")]
kind: Option<String>,
timestamp: Option<String>,
#[serde(default)]
payload: RawPayload,
}
#[derive(Deserialize, Default)]
struct RawPayload {
#[serde(rename = "type")]
item_type: Option<String>,
#[allow(dead_code)]
id: Option<String>,
cwd: Option<String>,
model: Option<String>,
role: Option<String>,
#[serde(default)]
content: serde_json::Value,
name: Option<String>,
arguments: Option<String>,
call_id: Option<String>,
info: Option<RawTokenInfo>,
invocation: Option<RawInvocation>,
tool: Option<String>,
}
#[derive(Deserialize)]
struct RawTokenInfo {
#[allow(dead_code)]
total_token_usage: Option<RawTokenUsage>,
last_token_usage: Option<RawTokenUsage>,
}
#[derive(Deserialize)]
struct RawTokenUsage {
input_tokens: Option<u64>,
cached_input_tokens: Option<u64>,
output_tokens: Option<u64>,
reasoning_output_tokens: Option<u64>,
#[allow(dead_code)]
total_tokens: Option<u64>,
}
#[derive(Deserialize)]
struct RawInvocation {
server: Option<String>,
tool: Option<String>,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn last_usage_maps_disjointly_to_total_tokens() {
let last = RawTokenUsage {
input_tokens: Some(2000),
cached_input_tokens: Some(500),
output_tokens: Some(300),
reasoning_output_tokens: Some(100),
total_tokens: Some(2300),
};
let u = map_last_usage(&last);
assert_eq!(u.cache_read, Some(500));
assert_eq!(u.input, Some(1500), "input minus cached subset");
assert_eq!(u.thinking, Some(100));
assert_eq!(u.output, Some(200), "output minus reasoning subset");
assert_eq!(u.cache_creation, None, "Codex has no cache-creation split");
assert_eq!(
u.known_total(),
2300,
"Σ disjoint fields == total_tokens for the request"
);
}
#[test]
fn underflow_saturates_instead_of_panicking() {
let last = RawTokenUsage {
input_tokens: Some(100),
cached_input_tokens: Some(250),
output_tokens: Some(40),
reasoning_output_tokens: Some(90),
total_tokens: Some(140),
};
let u = map_last_usage(&last);
assert_eq!(u.input, Some(0));
assert_eq!(u.output, Some(0));
assert_eq!(u.cache_read, Some(250));
assert_eq!(u.thinking, Some(90));
}
#[test]
fn codex_home_parses_comma_separated_roots() {
std::env::set_var("CODEX_HOME", " /a/one , /b/two ,, /c/three ");
let roots = codex_roots();
std::env::remove_var("CODEX_HOME");
assert_eq!(
roots,
vec![
PathBuf::from("/a/one"),
PathBuf::from("/b/two"),
PathBuf::from("/c/three"),
],
"trimmed, empties dropped"
);
}
#[test]
fn token_count_and_compaction_map_to_events() {
let mut session = Session {
id: "s".into(),
agent: "codex".into(),
project: None,
model: None,
parent_session: None,
started_at: None,
ended_at: None,
events: Vec::new(),
sub_agents: Vec::new(),
skipped_lines: 0,
};
let mut counts: BTreeMap<String, u64> = BTreeMap::new();
let model = Some("gpt-5.5".to_string());
let tc: RawLine = serde_json::from_str(
r#"{"timestamp":"2026-06-15T10:00:00Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":1000,"cached_input_tokens":0,"output_tokens":200,"reasoning_output_tokens":50,"total_tokens":1200},"last_token_usage":{"input_tokens":1000,"cached_input_tokens":0,"output_tokens":200,"reasoning_output_tokens":50,"total_tokens":1200}}}}"#,
)
.unwrap();
push_event_msg(
&tc,
parse_ts(&tc.timestamp),
&model,
&mut session,
&mut counts,
);
let cc: RawLine = serde_json::from_str(
r#"{"timestamp":"2026-06-15T10:00:01Z","type":"event_msg","payload":{"type":"context_compacted"}}"#,
)
.unwrap();
push_event_msg(
&cc,
parse_ts(&cc.timestamp),
&model,
&mut session,
&mut counts,
);
assert_eq!(session.events.len(), 2);
assert_eq!(session.events[0].kind, EventKind::Assistant);
assert_eq!(session.events[0].model.as_deref(), Some("gpt-5.5"));
assert!(session.events[0].has_thinking, "reasoning_output_tokens>0");
assert_eq!(session.events[0].usage.unwrap().known_total(), 1200);
assert_eq!(session.events[1].kind, EventKind::Compaction);
assert_eq!(counts.get("gpt-5.5"), Some(&1));
}
#[test]
fn mcp_tool_call_end_attributes_server() {
let raw: RawLine = serde_json::from_str(
r#"{"timestamp":"2026-06-15T10:00:00Z","type":"event_msg","payload":{"type":"mcp_tool_call_end","invocation":{"server":"acme-db","tool":"query"},"call_id":"c1"}}"#,
)
.unwrap();
let call = mcp_tool_call(&raw).expect("a tool call");
assert_eq!(call.name, "mcp__acme-db__query");
assert_eq!(call.server.as_deref(), Some("acme-db"));
}
}