diff --git a/src/collector/codex.rs b/src/collector/codex.rs index 43a4bfc..0471e7b 100644 --- a/src/collector/codex.rs +++ b/src/collector/codex.rs @@ -1,7 +1,7 @@ use super::process::{self, ProcInfo}; use crate::model::{ - AgentSession, ChatMessage, ChatRole, ChildProcess, RateLimitInfo, SessionStatus, ToolCall, - MAX_CHAT_MESSAGES, + AgentSession, ChatMessage, ChatRole, ChildProcess, FileAccess, FileOp, RateLimitInfo, + SessionStatus, ToolCall, MAX_CHAT_MESSAGES, MAX_FILE_ACCESSES, }; use serde_json::Value; use std::collections::{HashMap, HashSet}; @@ -664,7 +664,7 @@ impl CodexCollector { tool_calls: result.tool_calls, pending_since_ms: result.pending_since_ms, thinking_since_ms: result.thinking_since_ms, - file_accesses: vec![], + file_accesses: result.file_accesses, config_root: super::abbrev_path( self.sessions_dir .parent() @@ -900,6 +900,9 @@ struct CodexJSONLResult { pending_since_ms: u64, /// Timestamp of the latest user prompt not yet followed by assistant output. thinking_since_ms: u64, + /// Files touched, from the item_completed schema (Codex ≥ ~0.149). + /// The old schema never carried file information. + file_accesses: Vec, } impl CodexJSONLResult { @@ -940,6 +943,135 @@ fn sanitize_tool_arg(arg: &str) -> String { redacted.chars().take(120).collect() } +/// Joined text of an item's `content` array. The schema is not consistent +/// about casing ("Text" on AgentMessage, "text" on UserMessage), so any +/// object with a string `text` field counts. +fn concat_item_text(content: &Value) -> String { + content + .as_array() + .map(|parts| { + parts + .iter() + .filter_map(|p| p["text"].as_str()) + .collect::>() + .join("\n") + }) + .unwrap_or_default() +} + +/// One completed item from the new rollout schema (Codex ≥ ~0.149). +/// +/// UserMessage/AgentMessage carry the chat, CommandExecution and Extension +/// are the tool timeline, and FileChange is the only place file writes appear +/// (the old schema never reported files at all). +fn handle_item_completed(item: &Value, ts: u64, result: &mut CodexJSONLResult) { + match item["type"].as_str() { + Some("UserMessage") => { + result.model_generating = true; + result.thinking_since_ms = ts; + let text = concat_item_text(&item["content"]); + if !text.is_empty() { + if result.initial_prompt.is_empty() { + let truncated: String = text.chars().take(120).collect(); + result.initial_prompt = super::redact_secrets(&truncated); + } + push_chat_message( + &mut result.chat_messages, + ChatRole::User, + clean_chat_text(&text, 500), + ); + } + } + Some("AgentMessage") => { + result.turn_count += 1; + result.model_generating = false; + result.thinking_since_ms = 0; + let text = concat_item_text(&item["content"]); + push_chat_message( + &mut result.chat_messages, + ChatRole::Assistant, + clean_chat_text(&text, 500), + ); + } + Some("CommandExecution") => { + result.model_generating = false; + result.thinking_since_ms = 0; + let started = item["started_at_ms"].as_u64().unwrap_or(ts); + let completed = item["completed_at_ms"].as_u64().unwrap_or(started); + // parsed_cmd names the intent (read/write/search/…) and often a + // path; fall back to the raw argv when it is absent. + let parsed = &item["parsed_cmd"][0]; + let name = parsed["type"].as_str().unwrap_or("exec").to_string(); + let arg_raw = parsed["cmd"] + .as_str() + .map(str::to_string) + .or_else(|| value_to_tool_arg(&item["command"])) + .unwrap_or_default(); + if result.tool_calls.len() < 500 { + result.tool_calls.push(ToolCall { + name, + arg: sanitize_tool_arg(&arg_raw), + duration_ms: completed.saturating_sub(started), + }); + } + if let Some(entries) = item["parsed_cmd"].as_array() { + for pc in entries { + let Some(path) = pc["path"].as_str() else { continue }; + let op = match pc["type"].as_str() { + Some("write") => FileOp::Write, + _ => FileOp::Read, + }; + push_file_access(result, path, op); + } + } + } + Some("FileChange") => { + result.model_generating = false; + if let Some(changes) = item["changes"].as_object() { + for (path, change) in changes { + let op = match change["type"].as_str() { + Some("add") => FileOp::Write, + _ => FileOp::Edit, + }; + push_file_access(result, path, op); + if result.tool_calls.len() < 500 { + let short = path.rsplit('/').next().unwrap_or(path); + result.tool_calls.push(ToolCall { + name: "edit".to_string(), + arg: sanitize_tool_arg(short), + duration_ms: 0, + }); + } + } + } + } + Some("Extension") => { + let name = item["kind"].as_str().unwrap_or("extension").to_string(); + let arg = item["query"].as_str().unwrap_or_default(); + if result.tool_calls.len() < 500 { + result.tool_calls.push(ToolCall { + name, + arg: sanitize_tool_arg(arg), + duration_ms: 0, + }); + } + } + _ => {} + } +} + +fn push_file_access(result: &mut CodexJSONLResult, path: &str, op: FileOp) { + result.file_accesses.push(FileAccess { + path: path.to_string(), + operation: op, + turn_index: result.turn_count, + }); + let len = result.file_accesses.len(); + if len > MAX_FILE_ACCESSES { + result.file_accesses.drain(..len - MAX_FILE_ACCESSES); + } +} + fn push_chat_message(messages: &mut Vec, role: ChatRole, text: String) { if text.is_empty() { return; @@ -1082,6 +1214,7 @@ fn parse_codex_jsonl(path: &Path) -> Option { tool_calls: Vec::new(), pending_since_ms: 0, thinking_since_ms: 0, + file_accesses: Vec::new(), }; let mut call_indices: HashMap = HashMap::new(); let mut call_starts: HashMap = HashMap::new(); @@ -1164,6 +1297,24 @@ fn parse_codex_jsonl(path: &Path) -> Option { if let Some(cw) = payload["model_context_window"].as_u64() { result.context_window = cw; } + // New schema (Codex ≥ ~0.149) has no user_message + // event; a turn starting is the "model is working" + // signal until an AgentMessage or task_complete lands. + result.model_generating = true; + if result.thinking_since_ms == 0 { + result.thinking_since_ms = event_timestamp_ms(&val).unwrap_or(0); + } + } + // Newer Codex versions replaced the + // user_message/agent_message/function_call vocabulary + // with item_completed wrappers. + Some("item_completed") => { + let ts = event_timestamp_ms(&val).unwrap_or(0); + handle_item_completed(&payload["item"], ts, &mut result); + } + Some("turn_aborted") => { + result.model_generating = false; + result.thinking_since_ms = 0; } Some("user_message") => { result.model_generating = true; @@ -2202,4 +2353,78 @@ mod tests { let file = tempfile::NamedTempFile::new().unwrap(); assert!(parse_codex_jsonl(file.path()).is_none()); } + + #[test] + fn test_item_completed_chat_and_turns() { + // Codex ≥ ~0.149: chat arrives as item_completed wrappers. Note the + // casing drift — "text" on UserMessage content, "Text" on AgentMessage. + let mut file = tempfile::NamedTempFile::new().unwrap(); + write_lines( + &mut file, + &[ + SESSION_META, + r#"{"type":"event_msg","timestamp":"2026-08-21T05:45:27Z","payload":{"type":"item_completed","item":{"type":"UserMessage","id":"u1","content":[{"type":"text","text":"please fix the build"}]}}}"#, + r#"{"type":"event_msg","timestamp":"2026-08-21T05:45:54Z","payload":{"type":"item_completed","item":{"type":"AgentMessage","id":"a1","content":[{"type":"Text","text":"done"}],"phase":"final_answer"}}}"#, + ], + ); + let result = parse_codex_jsonl(file.path()).unwrap(); + assert_eq!(result.chat_messages.len(), 2); + assert_eq!(result.chat_messages[0].text, "please fix the build"); + assert_eq!(result.chat_messages[1].text, "done"); + assert_eq!(result.turn_count, 1); + assert!(!result.model_generating, "AgentMessage ends the turn"); + assert_eq!(result.initial_prompt, "please fix the build"); + } + + #[test] + fn test_item_completed_trailing_user_message_marks_generating() { + let mut file = tempfile::NamedTempFile::new().unwrap(); + write_lines( + &mut file, + &[ + SESSION_META, + r#"{"type":"event_msg","timestamp":"2026-08-21T05:45:27Z","payload":{"type":"item_completed","item":{"type":"UserMessage","id":"u1","content":[{"type":"text","text":"go"}]}}}"#, + ], + ); + let result = parse_codex_jsonl(file.path()).unwrap(); + assert!(result.model_generating, "unanswered UserMessage means the model is working"); + } + + #[test] + fn test_item_completed_command_execution_tools_and_files() { + let mut file = tempfile::NamedTempFile::new().unwrap(); + write_lines( + &mut file, + &[ + SESSION_META, + r#"{"type":"event_msg","timestamp":"2026-08-21T05:46:11Z","payload":{"type":"item_completed","item":{"type":"CommandExecution","id":"exec-1","command":["/bin/zsh","-lc","sed -n '1,240p' README.md"],"parsed_cmd":[{"type":"read","cmd":"sed -n '1,240p' README.md","name":"README.md","path":"/work/README.md"}],"started_at_ms":1787291167583,"completed_at_ms":1787291168442}}}"#, + ], + ); + let result = parse_codex_jsonl(file.path()).unwrap(); + assert_eq!(result.tool_calls.len(), 1); + assert_eq!(result.tool_calls[0].name, "read"); + assert_eq!(result.tool_calls[0].duration_ms, 859); + assert_eq!(result.file_accesses.len(), 1); + assert_eq!(result.file_accesses[0].path, "/work/README.md"); + assert!(matches!(result.file_accesses[0].operation, FileOp::Read)); + } + + #[test] + fn test_item_completed_file_change_records_accesses() { + let mut file = tempfile::NamedTempFile::new().unwrap(); + write_lines( + &mut file, + &[ + SESSION_META, + r#"{"type":"event_msg","timestamp":"2026-08-21T06:10:59Z","payload":{"type":"item_completed","item":{"type":"FileChange","id":"exec-2","changes":{"/work/new.md":{"type":"add"},"/work/old.rs":{"type":"update"}}}}}"#, + ], + ); + let result = parse_codex_jsonl(file.path()).unwrap(); + assert_eq!(result.file_accesses.len(), 2); + let write = result.file_accesses.iter().find(|f| f.path == "/work/new.md").unwrap(); + let edit = result.file_accesses.iter().find(|f| f.path == "/work/old.rs").unwrap(); + assert!(matches!(write.operation, FileOp::Write)); + assert!(matches!(edit.operation, FileOp::Edit)); + assert_eq!(result.tool_calls.len(), 2, "each change also lands on the tool timeline"); + } }