diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 844a84ef73..b4d61874d5 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3754,6 +3754,7 @@ dependencies = [ "codex-login", "codex-otel", "codex-protocol", + "codex-shell-command", "codex-state", "codex-utils-path", "pretty_assertions", diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index f6c5fbfb98..045f780ca2 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -55,6 +55,7 @@ mod plugin_cmd; mod remote_control_cmd; #[cfg(target_os = "windows")] mod sandbox_setup; +mod slow_commands; mod state_db_recovery; #[cfg(not(windows))] mod wsl_paths; @@ -234,6 +235,9 @@ enum DebugSubcommand { /// Render the model-visible prompt input list as JSON. PromptInput(DebugPromptInputCommand), + /// Analyze direct shell-command wait time in local rollout history. + SlowCommands(slow_commands::SlowCommandsCommand), + /// Replay a rollout trace bundle and write reduced state JSON. #[clap(hide = true)] TraceReduce(DebugTraceReduceCommand), @@ -1517,6 +1521,14 @@ async fn cli_main( ) .await?; } + DebugSubcommand::SlowCommands(cmd) => { + reject_remote_mode_for_subcommand( + root_remote.as_deref(), + root_remote_auth_token_env.as_deref(), + "debug slow-commands", + )?; + slow_commands::run(cmd).await?; + } DebugSubcommand::TraceReduce(cmd) => { reject_remote_mode_for_subcommand( root_remote.as_deref(), diff --git a/codex-rs/cli/src/slow_commands.rs b/codex-rs/cli/src/slow_commands.rs new file mode 100644 index 0000000000..8cd3d94201 --- /dev/null +++ b/codex-rs/cli/src/slow_commands.rs @@ -0,0 +1,354 @@ +use anyhow::Context; +use anyhow::Result; +use clap::Args; +use clap::Subcommand; +use codex_core::config::find_codex_home; +use codex_rollout::SlowCommandAggregate; +use codex_rollout::SlowCommandCollection; +use codex_rollout::SlowCommandContinuation; +use codex_rollout::SlowCommandDirectCall; +use codex_rollout::SlowCommandSummary; +use codex_rollout::analyze_slow_commands; +use codex_rollout::collect_slow_commands; +use codex_rollout::merge_slow_command_collections; +use serde::Deserialize; +use serde::Serialize; +use serde_json::Value; +use std::io::BufRead; +use std::io::BufWriter; +use std::io::Write; +use std::time::Duration; + +const SCHEMA_VERSION: u32 = 1; + +#[derive(Debug, Args)] +pub(crate) struct SlowCommandsCommand { + #[command(subcommand)] + action: Option, +} + +#[derive(Debug, Subcommand)] +enum SlowCommandsAction { + /// Emit mergeable low-level command observations as JSONL. + Collect, + /// Read concatenated collection JSONL from stdin and emit one analysis. + Merge, +} + +#[derive(Deserialize, Serialize)] +struct SummaryRecord { + #[serde(rename = "type")] + record_type: String, + schema_version: u32, + rollout_files_seen: usize, + rollout_files_analyzed: usize, + rollout_files_skipped: usize, + internal_rollout_files_skipped: usize, + rollout_lines_skipped: usize, + direct_shell_calls_seen: usize, + direct_shell_calls_analyzed: usize, + direct_shell_calls_skipped: usize, + continuation_calls_seen: usize, + continuation_calls_attributed: usize, + continuation_calls_skipped: usize, + duplicate_call_ids: usize, + open_processes_at_eof: usize, + code_mode_cells_ignored: usize, + total_wait_seconds: f64, +} + +impl SummaryRecord { + fn new(record_type: &str, summary: &SlowCommandSummary) -> Self { + Self { + record_type: record_type.to_string(), + schema_version: SCHEMA_VERSION, + rollout_files_seen: summary.rollout_files_seen, + rollout_files_analyzed: summary.rollout_files_analyzed, + rollout_files_skipped: summary.rollout_files_skipped, + internal_rollout_files_skipped: summary.internal_rollout_files_skipped, + rollout_lines_skipped: summary.rollout_lines_skipped, + direct_shell_calls_seen: summary.direct_shell_calls_seen, + direct_shell_calls_analyzed: summary.direct_shell_calls_analyzed, + direct_shell_calls_skipped: summary.direct_shell_calls_skipped, + continuation_calls_seen: summary.continuation_calls_seen, + continuation_calls_attributed: summary.continuation_calls_attributed, + continuation_calls_skipped: summary.continuation_calls_skipped, + duplicate_call_ids: summary.duplicate_call_ids, + open_processes_at_eof: summary.open_processes_at_eof, + code_mode_cells_ignored: summary.code_mode_cells_ignored, + total_wait_seconds: seconds(summary.total_wait), + } + } + + fn into_collection_summary(self) -> SlowCommandSummary { + SlowCommandSummary { + rollout_files_seen: self.rollout_files_seen, + rollout_files_analyzed: self.rollout_files_analyzed, + rollout_files_skipped: self.rollout_files_skipped, + internal_rollout_files_skipped: self.internal_rollout_files_skipped, + rollout_lines_skipped: self.rollout_lines_skipped, + duplicate_call_ids: self.duplicate_call_ids, + code_mode_cells_ignored: self.code_mode_cells_ignored, + ..Default::default() + } + } +} + +#[derive(Serialize)] +struct RankingRecord<'a> { + #[serde(rename = "type")] + record_type: &'static str, + schema_version: u32, + rank: usize, + #[serde(skip_serializing_if = "Option::is_none")] + command: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + executable: Option<&'a str>, + invocation_count: usize, + continuation_wait_count: usize, + total_wait_seconds: f64, + average_wait_seconds: f64, + max_wait_seconds: f64, +} + +#[derive(Deserialize, Serialize)] +struct DirectCallRecord { + #[serde(rename = "type")] + record_type: String, + schema_version: u32, + call_id: String, + command: Option, + wait_seconds: Option, + observed_running: bool, + observed_terminal: bool, +} + +#[derive(Deserialize, Serialize)] +struct ContinuationRecord { + #[serde(rename = "type")] + record_type: String, + schema_version: u32, + call_id: String, + invocation_call_id: Option, + wait_seconds: Option, + observed_running: bool, + observed_terminal: bool, +} + +#[derive(Clone, Copy)] +enum RankingKind { + ExactCommand, + CommandFamily, +} + +struct FixedPrecisionFormatter; + +impl serde_json::ser::Formatter for FixedPrecisionFormatter { + fn write_f64(&mut self, writer: &mut W, value: f64) -> std::io::Result<()> + where + W: ?Sized + Write, + { + write!(writer, "{value:.4}") + } +} + +pub(crate) async fn run(command: SlowCommandsCommand) -> Result<()> { + let mut output = BufWriter::new(std::io::stdout().lock()); + match command.action { + None => { + let codex_home = find_codex_home()?; + let analysis = analyze_slow_commands(codex_home.as_path()).await; + write_analysis(&mut output, &analysis)?; + } + Some(SlowCommandsAction::Collect) => { + let codex_home = find_codex_home()?; + let collection = collect_slow_commands(codex_home.as_path()).await; + write_collection(&mut output, &collection)?; + } + Some(SlowCommandsAction::Merge) => { + let collections = read_collections(std::io::stdin().lock())?; + let analysis = merge_slow_command_collections(collections); + write_analysis(&mut output, &analysis)?; + } + } + output.flush()?; + Ok(()) +} + +fn write_analysis( + output: &mut impl Write, + analysis: &codex_rollout::SlowCommandAnalysis, +) -> Result<()> { + write_json_line(output, &SummaryRecord::new("summary", &analysis.summary))?; + for (index, aggregate) in analysis.exact_commands.iter().enumerate() { + write_ranking(output, RankingKind::ExactCommand, index, aggregate)?; + } + for (index, aggregate) in analysis.command_families.iter().enumerate() { + write_ranking(output, RankingKind::CommandFamily, index, aggregate)?; + } + Ok(()) +} + +fn write_collection(output: &mut impl Write, collection: &SlowCommandCollection) -> Result<()> { + write_json_line( + output, + &SummaryRecord::new("collection_summary", &collection.summary), + )?; + for direct_call in &collection.direct_calls { + write_json_line( + output, + &DirectCallRecord { + record_type: "direct_call".to_string(), + schema_version: SCHEMA_VERSION, + call_id: direct_call.call_id.clone(), + command: direct_call.command.clone(), + wait_seconds: direct_call.wait.map(seconds), + observed_running: direct_call.observed_running, + observed_terminal: direct_call.observed_terminal, + }, + )?; + } + for continuation in &collection.continuations { + write_json_line( + output, + &ContinuationRecord { + record_type: "continuation".to_string(), + schema_version: SCHEMA_VERSION, + call_id: continuation.call_id.clone(), + invocation_call_id: continuation.invocation_call_id.clone(), + wait_seconds: continuation.wait.map(seconds), + observed_running: continuation.observed_running, + observed_terminal: continuation.observed_terminal, + }, + )?; + } + Ok(()) +} + +fn read_collections(reader: impl BufRead) -> Result> { + let mut collections = Vec::new(); + let mut current: Option = None; + for (index, line) in reader.lines().enumerate() { + let line_number = index.saturating_add(1); + let line = line.with_context(|| format!("failed to read collection line {line_number}"))?; + if line.trim().is_empty() { + continue; + } + let value: Value = serde_json::from_str(&line) + .with_context(|| format!("invalid JSON on collection line {line_number}"))?; + let record_type = value + .get("type") + .and_then(Value::as_str) + .with_context(|| format!("missing record type on collection line {line_number}"))?; + match record_type { + "collection_summary" => { + if let Some(collection) = current.take() { + collections.push(collection); + } + let record: SummaryRecord = serde_json::from_value(value) + .with_context(|| format!("invalid collection summary on line {line_number}"))?; + validate_schema_version(record.schema_version, line_number)?; + current = Some(SlowCommandCollection { + summary: record.into_collection_summary(), + ..Default::default() + }); + } + "direct_call" => { + let record: DirectCallRecord = serde_json::from_value(value) + .with_context(|| format!("invalid direct call on line {line_number}"))?; + validate_schema_version(record.schema_version, line_number)?; + current + .as_mut() + .with_context(|| { + format!("direct call before collection summary on line {line_number}") + })? + .direct_calls + .push(SlowCommandDirectCall { + call_id: record.call_id, + command: record.command, + wait: parse_duration(record.wait_seconds, line_number)?, + observed_running: record.observed_running, + observed_terminal: record.observed_terminal, + }); + } + "continuation" => { + let record: ContinuationRecord = serde_json::from_value(value) + .with_context(|| format!("invalid continuation on line {line_number}"))?; + validate_schema_version(record.schema_version, line_number)?; + current + .as_mut() + .with_context(|| { + format!("continuation before collection summary on line {line_number}") + })? + .continuations + .push(SlowCommandContinuation { + call_id: record.call_id, + invocation_call_id: record.invocation_call_id, + wait: parse_duration(record.wait_seconds, line_number)?, + observed_running: record.observed_running, + observed_terminal: record.observed_terminal, + }); + } + other => { + anyhow::bail!("unsupported record type `{other}` on collection line {line_number}") + } + } + } + if let Some(collection) = current { + collections.push(collection); + } + Ok(collections) +} + +fn validate_schema_version(schema_version: u32, line_number: usize) -> Result<()> { + if schema_version != SCHEMA_VERSION { + anyhow::bail!( + "unsupported schema version {schema_version} on collection line {line_number}" + ); + } + Ok(()) +} + +fn parse_duration(seconds: Option, line_number: usize) -> Result> { + seconds + .map(Duration::try_from_secs_f64) + .transpose() + .with_context(|| format!("invalid wait duration on collection line {line_number}")) +} + +fn write_ranking( + output: &mut impl Write, + kind: RankingKind, + index: usize, + aggregate: &SlowCommandAggregate, +) -> Result<()> { + let (record_type, command, executable) = match kind { + RankingKind::ExactCommand => ("exact_command", Some(aggregate.key.as_str()), None), + RankingKind::CommandFamily => ("command_family", None, Some(aggregate.key.as_str())), + }; + let record = RankingRecord { + record_type, + schema_version: SCHEMA_VERSION, + rank: index.saturating_add(1), + command, + executable, + invocation_count: aggregate.invocation_count, + continuation_wait_count: aggregate.continuation_wait_count, + total_wait_seconds: seconds(aggregate.total_wait), + average_wait_seconds: seconds(aggregate.average_wait()), + max_wait_seconds: seconds(aggregate.max_wait), + }; + write_json_line(output, &record) +} + +fn write_json_line(output: &mut impl Write, value: &impl Serialize) -> Result<()> { + let mut serializer = + serde_json::Serializer::with_formatter(&mut *output, FixedPrecisionFormatter); + value.serialize(&mut serializer)?; + output.write_all(b"\n")?; + Ok(()) +} + +fn seconds(duration: Duration) -> f64 { + duration.as_secs_f64() +} diff --git a/codex-rs/cli/tests/debug_slow_commands.rs b/codex-rs/cli/tests/debug_slow_commands.rs new file mode 100644 index 0000000000..635ff96a5f --- /dev/null +++ b/codex-rs/cli/tests/debug_slow_commands.rs @@ -0,0 +1,161 @@ +use anyhow::Result; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::fs; +use std::io::Write; +use std::path::Path; +use tempfile::TempDir; + +fn write_rollout(codex_home: &Path) -> Result<()> { + let id = "019ed917-334f-7393-8070-2b72069cd533"; + let dir = codex_home.join("sessions/2026/06/18"); + fs::create_dir_all(&dir)?; + let mut file = fs::File::create(dir.join(format!("rollout-2026-06-18T12-00-00-{id}.jsonl")))?; + for value in [ + json!({ + "timestamp": "2026-06-18T12:00:00Z", + "type": "session_meta", + "payload": { + "id": id, + "timestamp": "2026-06-18T12:00:00Z", + "cwd": codex_home, + "originator": "test", + "cli_version": "test", + "source": "cli", + "model_provider": null, + }, + }), + call("test", "cargo test -p codex-cli"), + output( + "test", + "Wall time: 1.5000 seconds\nProcess exited with code 0\nOutput:", + ), + call("check", "cargo check"), + output( + "check", + "Wall time: 0.5000 seconds\nProcess exited with code 0\nOutput:", + ), + ] { + writeln!(file, "{value}")?; + } + Ok(()) +} + +fn call(call_id: &str, command: &str) -> Value { + json!({ + "timestamp": "2026-06-18T12:00:01Z", + "type": "response_item", + "payload": { + "type": "function_call", + "name": "exec_command", + "arguments": json!({"cmd": command}).to_string(), + "call_id": call_id, + }, + }) +} + +fn output(call_id: &str, text: &str) -> Value { + json!({ + "timestamp": "2026-06-18T12:00:02Z", + "type": "response_item", + "payload": {"type": "function_call_output", "call_id": call_id, "output": text}, + }) +} + +fn run(codex_home: &Path) -> Result { + run_with(codex_home, &[], None) +} + +fn run_with(codex_home: &Path, args: &[&str], stdin: Option<&str>) -> Result { + let mut command = assert_cmd::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?); + command + .env("CODEX_HOME", codex_home) + .args(["debug", "slow-commands"]) + .args(args); + if let Some(stdin) = stdin { + command.write_stdin(stdin); + } + let output = command.output()?; + assert!(output.status.success(), "{output:?}"); + let stdout = String::from_utf8(output.stdout)?; + for line in stdout.lines() { + serde_json::from_str::(line)?; + } + Ok(stdout) +} + +#[test] +fn slow_commands_prints_versioned_ranked_jsonl() -> Result<()> { + let home = TempDir::new()?; + write_rollout(home.path())?; + + let stdout = run(home.path())?; + + assert_eq!( + stdout.lines().collect::>(), + vec![ + r#"{"type":"summary","schema_version":1,"rollout_files_seen":1,"rollout_files_analyzed":1,"rollout_files_skipped":0,"internal_rollout_files_skipped":0,"rollout_lines_skipped":0,"direct_shell_calls_seen":2,"direct_shell_calls_analyzed":2,"direct_shell_calls_skipped":0,"continuation_calls_seen":0,"continuation_calls_attributed":0,"continuation_calls_skipped":0,"duplicate_call_ids":0,"open_processes_at_eof":0,"code_mode_cells_ignored":0,"total_wait_seconds":2.0000}"#, + r#"{"type":"exact_command","schema_version":1,"rank":1,"command":"cargo test -p codex-cli","invocation_count":1,"continuation_wait_count":0,"total_wait_seconds":1.5000,"average_wait_seconds":1.5000,"max_wait_seconds":1.5000}"#, + r#"{"type":"exact_command","schema_version":1,"rank":2,"command":"cargo check","invocation_count":1,"continuation_wait_count":0,"total_wait_seconds":0.5000,"average_wait_seconds":0.5000,"max_wait_seconds":0.5000}"#, + r#"{"type":"command_family","schema_version":1,"rank":1,"executable":"cargo","invocation_count":2,"continuation_wait_count":0,"total_wait_seconds":2.0000,"average_wait_seconds":1.0000,"max_wait_seconds":1.5000}"#, + ] + ); + assert!(stdout.ends_with('\n')); + Ok(()) +} + +#[test] +fn slow_commands_empty_home_prints_only_summary() -> Result<()> { + let home = TempDir::new()?; + + let stdout = run(home.path())?; + + assert_eq!( + stdout, + concat!( + r#"{"type":"summary","schema_version":1,"rollout_files_seen":0,"rollout_files_analyzed":0,"rollout_files_skipped":0,"internal_rollout_files_skipped":0,"rollout_lines_skipped":0,"direct_shell_calls_seen":0,"direct_shell_calls_analyzed":0,"direct_shell_calls_skipped":0,"continuation_calls_seen":0,"continuation_calls_attributed":0,"continuation_calls_skipped":0,"duplicate_call_ids":0,"open_processes_at_eof":0,"code_mode_cells_ignored":0,"total_wait_seconds":0.0000}"#, + "\n" + ) + ); + Ok(()) +} + +#[test] +fn slow_commands_collections_round_trip_and_deduplicate_across_machines() -> Result<()> { + let first_home = TempDir::new()?; + let second_home = TempDir::new()?; + write_rollout(first_home.path())?; + write_rollout(second_home.path())?; + + let first = run_with(first_home.path(), &["collect"], None)?; + let second = run_with(second_home.path(), &["collect"], None)?; + let collected = first + .lines() + .map(serde_json::from_str::) + .collect::>>()?; + assert_eq!(collected.len(), 3); + assert_eq!(collected[0]["type"], "collection_summary"); + assert_eq!(collected[1]["type"], "direct_call"); + assert_eq!(collected[1]["call_id"], "check"); + assert_eq!(collected[1]["wait_seconds"].as_f64(), Some(0.5)); + + let input = format!("{first}{second}"); + let merged = run_with(first_home.path(), &["merge"], Some(&input))?; + let records = merged + .lines() + .map(serde_json::from_str::) + .collect::>>()?; + + assert_eq!(records.len(), 4); + assert_eq!(records[0]["type"], "summary"); + assert_eq!(records[0]["rollout_files_seen"], 2); + assert_eq!(records[0]["direct_shell_calls_seen"], 2); + assert_eq!(records[0]["duplicate_call_ids"], 2); + assert_eq!(records[0]["total_wait_seconds"].as_f64(), Some(2.0)); + assert_eq!(records[1]["command"], "cargo test -p codex-cli"); + assert_eq!(records[1]["invocation_count"], 1); + assert_eq!(records[3]["executable"], "cargo"); + assert_eq!(records[3]["invocation_count"], 2); + Ok(()) +} diff --git a/codex-rs/rollout/Cargo.toml b/codex-rs/rollout/Cargo.toml index edc9059688..1136845c1e 100644 --- a/codex-rs/rollout/Cargo.toml +++ b/codex-rs/rollout/Cargo.toml @@ -20,6 +20,7 @@ codex-git-utils = { workspace = true } codex-login = { workspace = true } codex-otel = { workspace = true } codex-protocol = { workspace = true } +codex-shell-command = { workspace = true } codex-state = { workspace = true } codex-utils-path = { workspace = true } regex = { workspace = true } diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 23be07996c..3edeafd9eb 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -12,6 +12,7 @@ pub(crate) mod policy; pub(crate) mod recorder; pub(crate) mod search; pub(crate) mod session_index; +mod slow_commands; mod sqlite_metrics; pub mod state_db; @@ -74,6 +75,15 @@ pub use session_index::find_thread_meta_by_name_str; pub use session_index::find_thread_name_by_id; pub use session_index::find_thread_names_by_ids; pub use session_index::remove_thread_name_entries; +pub use slow_commands::SlowCommandAggregate; +pub use slow_commands::SlowCommandAnalysis; +pub use slow_commands::SlowCommandCollection; +pub use slow_commands::SlowCommandContinuation; +pub use slow_commands::SlowCommandDirectCall; +pub use slow_commands::SlowCommandSummary; +pub use slow_commands::analyze_slow_commands; +pub use slow_commands::collect_slow_commands; +pub use slow_commands::merge_slow_command_collections; pub use state_db::StateDbHandle; pub use state_db::sqlite_telemetry_recorder; diff --git a/codex-rs/rollout/src/slow_commands.rs b/codex-rs/rollout/src/slow_commands.rs new file mode 100644 index 0000000000..3db649bc86 --- /dev/null +++ b/codex-rs/rollout/src/slow_commands.rs @@ -0,0 +1,421 @@ +use crate::ARCHIVED_SESSIONS_SUBDIR; +use crate::SESSIONS_SUBDIR; +use crate::compression; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::RolloutLine; +use codex_protocol::protocol::SessionSource; +use std::collections::BTreeMap; +use std::collections::HashMap; +use std::collections::HashSet; +use std::path::Path; +use std::path::PathBuf; +use std::time::Duration; + +mod merge; +mod parsing; + +pub use merge::merge_slow_command_collections; + +use merge::analyze_observations; +use merge::merge_optional_duration; +use merge::merge_optional_string; +use parsing::parse_direct_command; +use parsing::parse_running_session_id; +use parsing::parse_session_id; +use parsing::parse_wall_time; + +/// Coverage and accounting data for a slow-command analysis. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SlowCommandSummary { + pub rollout_files_seen: usize, + pub rollout_files_analyzed: usize, + pub rollout_files_skipped: usize, + pub internal_rollout_files_skipped: usize, + pub rollout_lines_skipped: usize, + pub direct_shell_calls_seen: usize, + pub direct_shell_calls_analyzed: usize, + pub direct_shell_calls_skipped: usize, + pub continuation_calls_seen: usize, + pub continuation_calls_attributed: usize, + pub continuation_calls_skipped: usize, + pub duplicate_call_ids: usize, + pub open_processes_at_eof: usize, + pub code_mode_cells_ignored: usize, + pub total_wait: Duration, +} + +/// Aggregate wait-time statistics for one exact command or executable family. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SlowCommandAggregate { + pub key: String, + pub invocation_count: usize, + pub continuation_wait_count: usize, + pub total_wait: Duration, + pub max_wait: Duration, +} + +impl SlowCommandAggregate { + pub fn average_wait(&self) -> Duration { + if self.invocation_count == 0 { + return Duration::ZERO; + } + Duration::from_secs_f64(self.total_wait.as_secs_f64() / self.invocation_count as f64) + } +} + +/// Slow-command analysis over all eligible rollout files in a Codex home. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SlowCommandAnalysis { + pub summary: SlowCommandSummary, + pub exact_commands: Vec, + pub command_families: Vec, +} + +/// Mergeable low-level observations collected from one Codex home. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SlowCommandCollection { + pub summary: SlowCommandSummary, + pub direct_calls: Vec, + pub continuations: Vec, +} + +/// One observed direct `shell_command` or `exec_command` call. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SlowCommandDirectCall { + pub call_id: String, + pub command: Option, + pub wait: Option, + pub observed_running: bool, + pub observed_terminal: bool, +} + +/// One observed `write_stdin` call and its optional originating command. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SlowCommandContinuation { + pub call_id: String, + pub invocation_call_id: Option, + pub wait: Option, + pub observed_running: bool, + pub observed_terminal: bool, +} + +enum PendingCall { + Direct, + Continuation { session_id: Option }, +} + +#[derive(Default)] +struct FileAnalysisState { + pending_calls: HashMap, + active_processes: HashMap, +} + +#[derive(Default)] +struct AnalysisState { + summary: SlowCommandSummary, + direct_calls: HashMap, + continuations: HashMap, + relevant_call_ids: HashSet, + code_mode_call_ids: HashSet, +} + +/// Analyzes direct agent shell calls in active and archived rollout history. +pub async fn analyze_slow_commands(codex_home: &Path) -> SlowCommandAnalysis { + merge_slow_command_collections([collect_slow_commands(codex_home).await]) +} + +/// Collects mergeable direct-call and continuation observations from a Codex home. +pub async fn collect_slow_commands(codex_home: &Path) -> SlowCommandCollection { + let mut state = AnalysisState::default(); + let paths = discover_rollout_paths(codex_home, &mut state.summary).await; + state.summary.rollout_files_seen = paths.len(); + + for path in paths { + analyze_rollout(&path, &mut state).await; + } + + finish_collection(state) +} + +async fn discover_rollout_paths( + codex_home: &Path, + summary: &mut SlowCommandSummary, +) -> Vec { + let mut dirs = vec![ + codex_home.join(SESSIONS_SUBDIR), + codex_home.join(ARCHIVED_SESSIONS_SUBDIR), + ]; + let mut files = BTreeMap::::new(); + + while let Some(dir) = dirs.pop() { + let mut entries = match tokio::fs::read_dir(&dir).await { + Ok(entries) => entries, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue, + Err(_) => { + summary.rollout_files_skipped = summary.rollout_files_skipped.saturating_add(1); + continue; + } + }; + loop { + let entry = match entries.next_entry().await { + Ok(Some(entry)) => entry, + Ok(None) => break, + Err(_) => { + summary.rollout_files_skipped = summary.rollout_files_skipped.saturating_add(1); + break; + } + }; + let path = entry.path(); + let file_type = match entry.file_type().await { + Ok(file_type) => file_type, + Err(_) => { + summary.rollout_files_skipped = summary.rollout_files_skipped.saturating_add(1); + continue; + } + }; + if file_type.is_dir() { + dirs.push(path); + continue; + } + if !file_type.is_file() { + continue; + } + let Some(rollout_file) = compression::RolloutFile::from_path(path) else { + continue; + }; + let canonical_path = compression::plain_rollout_path(rollout_file.path()); + files.insert(canonical_path, rollout_file.into_path()); + } + } + + files.into_values().collect() +} + +async fn analyze_rollout(path: &Path, state: &mut AnalysisState) { + let mut reader = match compression::open_rollout_line_reader(path).await { + Ok(reader) => reader, + Err(_) => { + state.summary.rollout_files_skipped = + state.summary.rollout_files_skipped.saturating_add(1); + return; + } + }; + let mut file_state = FileAnalysisState::default(); + let mut saw_session_meta = false; + + loop { + let line = match reader.next_line().await { + Ok(Some(line)) => line, + Ok(None) => break, + Err(_) => { + state.summary.rollout_files_skipped = + state.summary.rollout_files_skipped.saturating_add(1); + break; + } + }; + if line.trim().is_empty() { + continue; + } + let rollout_line = match serde_json::from_str::(&line) { + Ok(rollout_line) => rollout_line, + Err(_) => { + state.summary.rollout_lines_skipped = + state.summary.rollout_lines_skipped.saturating_add(1); + continue; + } + }; + match rollout_line.item { + RolloutItem::SessionMeta(meta) if !saw_session_meta => { + saw_session_meta = true; + if matches!(meta.meta.source, SessionSource::Internal(_)) { + state.summary.internal_rollout_files_skipped = state + .summary + .internal_rollout_files_skipped + .saturating_add(1); + return; + } + } + RolloutItem::ResponseItem(item) => { + analyze_response_item(item, &mut file_state, state); + } + RolloutItem::SessionMeta(_) + | RolloutItem::InterAgentCommunication(_) + | RolloutItem::Compacted(_) + | RolloutItem::TurnContext(_) + | RolloutItem::EventMsg(_) => {} + } + } + + state.summary.rollout_files_analyzed = state.summary.rollout_files_analyzed.saturating_add(1); +} + +fn analyze_response_item( + item: ResponseItem, + file_state: &mut FileAnalysisState, + state: &mut AnalysisState, +) { + match item { + ResponseItem::FunctionCall { + name, + namespace, + arguments, + call_id, + .. + } if namespace.is_none() => match name.as_str() { + "shell_command" | "exec_command" => { + record_relevant_call_id(&call_id, state); + let command = parse_direct_command(&name, &arguments); + let direct_call = state + .direct_calls + .entry(call_id.clone()) + .or_insert_with(|| SlowCommandDirectCall { + call_id: call_id.clone(), + command: None, + wait: None, + observed_running: false, + observed_terminal: false, + }); + merge_optional_string(&mut direct_call.command, command); + file_state + .pending_calls + .insert(call_id, PendingCall::Direct); + } + "write_stdin" => { + record_relevant_call_id(&call_id, state); + state + .continuations + .entry(call_id.clone()) + .or_insert_with(|| SlowCommandContinuation { + call_id: call_id.clone(), + invocation_call_id: None, + wait: None, + observed_running: false, + observed_terminal: false, + }); + file_state.pending_calls.insert( + call_id, + PendingCall::Continuation { + session_id: parse_session_id(&arguments), + }, + ); + } + _ => {} + }, + ResponseItem::FunctionCallOutput { + call_id, output, .. + } => { + let Some(pending) = file_state.pending_calls.remove(&call_id) else { + return; + }; + let text = output.body.to_text().unwrap_or_default(); + match pending { + PendingCall::Direct => { + record_direct_output(&call_id, &text, file_state, state); + } + PendingCall::Continuation { session_id } => { + record_continuation_output(&call_id, session_id, &text, file_state, state); + } + } + } + ResponseItem::CustomToolCall { call_id, name, .. } if name == "exec" => { + state.code_mode_call_ids.insert(call_id); + } + ResponseItem::Message { .. } + | ResponseItem::AgentMessage { .. } + | ResponseItem::Reasoning { .. } + | ResponseItem::LocalShellCall { .. } + | ResponseItem::FunctionCall { .. } + | ResponseItem::ToolSearchCall { .. } + | ResponseItem::CustomToolCall { .. } + | ResponseItem::CustomToolCallOutput { .. } + | ResponseItem::ToolSearchOutput { .. } + | ResponseItem::WebSearchCall { .. } + | ResponseItem::ImageGenerationCall { .. } + | ResponseItem::Compaction { .. } + | ResponseItem::CompactionTrigger { .. } + | ResponseItem::ContextCompaction { .. } + | ResponseItem::Other => {} + } +} + +fn record_relevant_call_id(call_id: &str, state: &mut AnalysisState) { + if !state.relevant_call_ids.insert(call_id.to_string()) { + state.summary.duplicate_call_ids = state.summary.duplicate_call_ids.saturating_add(1); + } +} + +fn record_direct_output( + call_id: &str, + output: &str, + file_state: &mut FileAnalysisState, + state: &mut AnalysisState, +) { + let Some(direct_call) = state.direct_calls.get_mut(call_id) else { + return; + }; + + if let Some(duration) = parse_wall_time(output) { + merge_optional_duration(&mut direct_call.wait, Some(duration)); + } + + if let Some(session_id) = parse_running_session_id(output) { + direct_call.observed_running = true; + file_state + .active_processes + .insert(session_id, call_id.to_string()); + } else { + direct_call.observed_terminal = true; + } +} + +fn record_continuation_output( + call_id: &str, + session_id: Option, + output: &str, + file_state: &mut FileAnalysisState, + state: &mut AnalysisState, +) { + let Some(continuation) = state.continuations.get_mut(call_id) else { + return; + }; + + if let Some(duration) = parse_wall_time(output) { + merge_optional_duration(&mut continuation.wait, Some(duration)); + } + let observed_running = parse_running_session_id(output).is_some(); + if observed_running { + continuation.observed_running = true; + } else { + continuation.observed_terminal = true; + } + + let Some(session_id) = session_id else { + return; + }; + let Some(invocation_id) = file_state.active_processes.get(&session_id).cloned() else { + return; + }; + merge_optional_string(&mut continuation.invocation_call_id, Some(invocation_id)); + if !observed_running { + file_state.active_processes.remove(&session_id); + } +} + +fn finish_collection(mut state: AnalysisState) -> SlowCommandCollection { + let mut direct_calls = state.direct_calls.into_values().collect::>(); + direct_calls.sort_by(|left, right| left.call_id.cmp(&right.call_id)); + let mut continuations = state.continuations.into_values().collect::>(); + continuations.sort_by(|left, right| left.call_id.cmp(&right.call_id)); + state.summary.code_mode_cells_ignored = state.code_mode_call_ids.len(); + state.summary = analyze_observations(state.summary, &direct_calls, &continuations).summary; + SlowCommandCollection { + summary: state.summary, + direct_calls, + continuations, + } +} + +#[cfg(test)] +#[path = "slow_commands_tests.rs"] +mod tests; diff --git a/codex-rs/rollout/src/slow_commands/merge.rs b/codex-rs/rollout/src/slow_commands/merge.rs new file mode 100644 index 0000000000..caad617a80 --- /dev/null +++ b/codex-rs/rollout/src/slow_commands/merge.rs @@ -0,0 +1,227 @@ +use super::SlowCommandAggregate; +use super::SlowCommandAnalysis; +use super::SlowCommandCollection; +use super::SlowCommandContinuation; +use super::SlowCommandDirectCall; +use super::SlowCommandSummary; +use super::parsing::first_executable; +use std::collections::HashMap; +use std::collections::HashSet; +use std::time::Duration; + +struct Invocation { + command: String, + executable: String, + continuation_wait_count: usize, + total_wait: Duration, +} + +/// Merges low-level collections, deduplicates calls globally, and ranks their commands. +pub fn merge_slow_command_collections( + collections: impl IntoIterator, +) -> SlowCommandAnalysis { + let mut summary = SlowCommandSummary::default(); + let mut direct_calls = HashMap::::new(); + let mut continuations = HashMap::::new(); + + for collection in collections { + add_coverage(&mut summary, &collection.summary); + for direct_call in collection.direct_calls { + match direct_calls.entry(direct_call.call_id.clone()) { + std::collections::hash_map::Entry::Occupied(mut entry) => { + summary.duplicate_call_ids = summary.duplicate_call_ids.saturating_add(1); + merge_direct_call(entry.get_mut(), direct_call); + } + std::collections::hash_map::Entry::Vacant(entry) => { + entry.insert(direct_call); + } + } + } + for continuation in collection.continuations { + match continuations.entry(continuation.call_id.clone()) { + std::collections::hash_map::Entry::Occupied(mut entry) => { + summary.duplicate_call_ids = summary.duplicate_call_ids.saturating_add(1); + merge_continuation(entry.get_mut(), continuation); + } + std::collections::hash_map::Entry::Vacant(entry) => { + entry.insert(continuation); + } + } + } + } + + let direct_calls = direct_calls.into_values().collect::>(); + let continuations = continuations.into_values().collect::>(); + analyze_observations(summary, &direct_calls, &continuations) +} + +pub(super) fn analyze_observations( + mut summary: SlowCommandSummary, + direct_calls: &[SlowCommandDirectCall], + continuations: &[SlowCommandContinuation], +) -> SlowCommandAnalysis { + let mut continuations_by_invocation = HashMap::<&str, Vec<&SlowCommandContinuation>>::new(); + for continuation in continuations { + if let Some(invocation_call_id) = continuation.invocation_call_id.as_deref() { + continuations_by_invocation + .entry(invocation_call_id) + .or_default() + .push(continuation); + } + } + + let mut invocations = Vec::new(); + let mut attributed_continuation_ids = HashSet::new(); + let mut open_processes = 0usize; + for direct_call in direct_calls { + let Some(command) = direct_call.command.as_ref() else { + continue; + }; + let mut total_wait = direct_call.wait.unwrap_or(Duration::ZERO); + let mut wait_samples = usize::from(direct_call.wait.is_some()); + let mut continuation_wait_count = 0usize; + let mut observed_running = direct_call.observed_running; + let mut observed_terminal = direct_call.observed_terminal; + for continuation in continuations_by_invocation + .get(direct_call.call_id.as_str()) + .into_iter() + .flatten() + { + observed_running |= continuation.observed_running; + observed_terminal |= continuation.observed_terminal; + if let Some(wait) = continuation.wait { + total_wait = total_wait.saturating_add(wait); + wait_samples = wait_samples.saturating_add(1); + continuation_wait_count = continuation_wait_count.saturating_add(1); + attributed_continuation_ids.insert(continuation.call_id.as_str()); + } + } + if observed_running && !observed_terminal { + open_processes = open_processes.saturating_add(1); + } + if wait_samples > 0 { + invocations.push(Invocation { + executable: first_executable(command), + command: command.clone(), + continuation_wait_count, + total_wait, + }); + } + } + + summary.direct_shell_calls_seen = direct_calls.len(); + summary.direct_shell_calls_analyzed = invocations.len(); + summary.direct_shell_calls_skipped = summary + .direct_shell_calls_seen + .saturating_sub(summary.direct_shell_calls_analyzed); + summary.continuation_calls_seen = continuations.len(); + summary.continuation_calls_attributed = attributed_continuation_ids.len(); + summary.continuation_calls_skipped = summary + .continuation_calls_seen + .saturating_sub(summary.continuation_calls_attributed); + summary.open_processes_at_eof = open_processes; + summary.total_wait = invocations + .iter() + .fold(Duration::ZERO, |total, invocation| { + total.saturating_add(invocation.total_wait) + }); + + let exact_commands = + aggregate_invocations(&invocations, |invocation| invocation.command.as_str()); + let command_families = + aggregate_invocations(&invocations, |invocation| invocation.executable.as_str()); + SlowCommandAnalysis { + summary, + exact_commands, + command_families, + } +} + +pub(super) fn merge_optional_string(target: &mut Option, incoming: Option) { + if let Some(incoming) = incoming + && target + .as_ref() + .is_none_or(|current| incoming.as_str() < current.as_str()) + { + *target = Some(incoming); + } +} + +pub(super) fn merge_optional_duration(target: &mut Option, incoming: Option) { + if let Some(incoming) = incoming + && target.is_none_or(|current| incoming > current) + { + *target = Some(incoming); + } +} + +fn add_coverage(target: &mut SlowCommandSummary, source: &SlowCommandSummary) { + target.rollout_files_seen = target + .rollout_files_seen + .saturating_add(source.rollout_files_seen); + target.rollout_files_analyzed = target + .rollout_files_analyzed + .saturating_add(source.rollout_files_analyzed); + target.rollout_files_skipped = target + .rollout_files_skipped + .saturating_add(source.rollout_files_skipped); + target.internal_rollout_files_skipped = target + .internal_rollout_files_skipped + .saturating_add(source.internal_rollout_files_skipped); + target.rollout_lines_skipped = target + .rollout_lines_skipped + .saturating_add(source.rollout_lines_skipped); + target.duplicate_call_ids = target + .duplicate_call_ids + .saturating_add(source.duplicate_call_ids); + target.code_mode_cells_ignored = target + .code_mode_cells_ignored + .saturating_add(source.code_mode_cells_ignored); +} + +fn merge_direct_call(target: &mut SlowCommandDirectCall, incoming: SlowCommandDirectCall) { + merge_optional_string(&mut target.command, incoming.command); + merge_optional_duration(&mut target.wait, incoming.wait); + target.observed_running |= incoming.observed_running; + target.observed_terminal |= incoming.observed_terminal; +} + +fn merge_continuation(target: &mut SlowCommandContinuation, incoming: SlowCommandContinuation) { + merge_optional_string(&mut target.invocation_call_id, incoming.invocation_call_id); + merge_optional_duration(&mut target.wait, incoming.wait); + target.observed_running |= incoming.observed_running; + target.observed_terminal |= incoming.observed_terminal; +} + +fn aggregate_invocations<'a>( + invocations: &'a [Invocation], + key: impl Fn(&'a Invocation) -> &'a str, +) -> Vec { + let mut aggregates = HashMap::::new(); + for invocation in invocations { + let aggregate = aggregates + .entry(key(invocation).to_string()) + .or_insert_with(|| SlowCommandAggregate { + key: key(invocation).to_string(), + invocation_count: 0, + continuation_wait_count: 0, + total_wait: Duration::ZERO, + max_wait: Duration::ZERO, + }); + aggregate.invocation_count = aggregate.invocation_count.saturating_add(1); + aggregate.continuation_wait_count = aggregate + .continuation_wait_count + .saturating_add(invocation.continuation_wait_count); + aggregate.total_wait = aggregate.total_wait.saturating_add(invocation.total_wait); + aggregate.max_wait = aggregate.max_wait.max(invocation.total_wait); + } + let mut aggregates = aggregates.into_values().collect::>(); + aggregates.sort_by(|left, right| { + right + .total_wait + .cmp(&left.total_wait) + .then_with(|| right.invocation_count.cmp(&left.invocation_count)) + .then_with(|| left.key.cmp(&right.key)) + }); + aggregates +} diff --git a/codex-rs/rollout/src/slow_commands/parsing.rs b/codex-rs/rollout/src/slow_commands/parsing.rs new file mode 100644 index 0000000000..1d2c80bc47 --- /dev/null +++ b/codex-rs/rollout/src/slow_commands/parsing.rs @@ -0,0 +1,120 @@ +use codex_shell_command::bash::try_parse_shell; +use regex::Regex; +use serde_json::Value; +use std::sync::LazyLock; +use std::time::Duration; + +static WALL_TIME_RE: LazyLock = LazyLock::new(|| { + compile_regex(r"(?m)^(?:Wall time|Duration): (?-?[0-9]+(?:\.[0-9]+)?) seconds\r?$") +}); +static RUNNING_SESSION_RE: LazyLock = LazyLock::new(|| { + compile_regex(r"(?m)^Process running with session ID (?-?[0-9]+)\r?$") +}); + +fn compile_regex(pattern: &str) -> Regex { + match Regex::new(pattern) { + Ok(regex) => regex, + Err(error) => panic!("invalid slow-command regex `{pattern}`: {error}"), + } +} + +pub(super) fn parse_direct_command(name: &str, arguments: &str) -> Option { + let value: Value = serde_json::from_str(arguments).ok()?; + let command = match name { + "shell_command" => value.get("command"), + "exec_command" => value.get("cmd").or_else(|| value.get("command")), + _ => None, + }?; + match command { + Value::String(command) if !command.is_empty() => Some(command.clone()), + Value::Array(parts) => { + let parts = parts + .iter() + .map(Value::as_str) + .collect::>>()?; + (!parts.is_empty()).then(|| { + codex_shell_command::parse_command::shlex_join( + &parts.into_iter().map(str::to_string).collect::>(), + ) + }) + } + Value::Null | Value::Bool(_) | Value::Number(_) | Value::Object(_) | Value::String(_) => { + None + } + } +} + +pub(super) fn parse_session_id(arguments: &str) -> Option { + serde_json::from_str::(arguments) + .ok()? + .get("session_id")? + .as_i64() +} + +pub(super) fn parse_wall_time(output: &str) -> Option { + let seconds = WALL_TIME_RE + .captures(output)? + .name("seconds")? + .as_str() + .parse::() + .ok()?; + Duration::try_from_secs_f64(seconds).ok() +} + +pub(super) fn parse_running_session_id(output: &str) -> Option { + RUNNING_SESSION_RE + .captures(output)? + .name("session_id")? + .as_str() + .parse() + .ok() +} + +pub(super) fn first_executable(command: &str) -> String { + let parsed = try_parse_shell(command).and_then(|tree| { + let mut nodes = vec![tree.root_node()]; + let mut command_name: Option<(usize, String)> = None; + while let Some(node) = nodes.pop() { + if node.kind() == "command_name" + && command_name + .as_ref() + .is_none_or(|(start_byte, _name)| node.start_byte() < *start_byte) + && let Ok(name) = node.utf8_text(command.as_bytes()) + { + command_name = Some((node.start_byte(), name.to_string())); + } + let mut cursor = node.walk(); + nodes.extend(node.children(&mut cursor)); + } + command_name.map(|(_start_byte, name)| name) + }); + let executable = parsed.or_else(|| { + command + .split_whitespace() + .find(|token| *token != "&" && !looks_like_environment_assignment(token)) + .map(str::to_string) + }); + normalize_executable(executable.as_deref().unwrap_or("")) +} + +fn looks_like_environment_assignment(token: &str) -> bool { + token.split_once('=').is_some_and(|(name, _value)| { + !name.is_empty() + && name + .chars() + .all(|ch| ch == '_' || ch.is_ascii_alphanumeric()) + }) +} + +pub(super) fn normalize_executable(executable: &str) -> String { + let executable = executable.trim_matches(['\'', '"']); + let executable = executable.rsplit(['/', '\\']).next().unwrap_or(executable); + let suffix = executable.get(executable.len().saturating_sub(4)..); + if executable.len() > 4 && suffix.is_some_and(|suffix| suffix.eq_ignore_ascii_case(".exe")) { + executable[..executable.len() - 4].to_string() + } else if executable.is_empty() { + "".to_string() + } else { + executable.to_string() + } +} diff --git a/codex-rs/rollout/src/slow_commands_tests.rs b/codex-rs/rollout/src/slow_commands_tests.rs new file mode 100644 index 0000000000..f0d80753be --- /dev/null +++ b/codex-rs/rollout/src/slow_commands_tests.rs @@ -0,0 +1,398 @@ +use super::*; +use crate::compression::compressed_rollout_path; +use codex_protocol::protocol::InternalSessionSource; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::fs; +use std::io::Write; +use std::path::PathBuf; +use tempfile::TempDir; +use uuid::Uuid; + +struct Fixture { + home: TempDir, +} + +impl Fixture { + fn new() -> Self { + Self { + home: TempDir::new().expect("create temp Codex home"), + } + } + + fn write_rollout(&self, archived: bool, source: SessionSource, lines: &[String]) -> PathBuf { + let id = Uuid::new_v4(); + let root = if archived { + self.home.path().join(ARCHIVED_SESSIONS_SUBDIR) + } else { + self.home.path().join(SESSIONS_SUBDIR).join("2026/06/18") + }; + fs::create_dir_all(&root).expect("create rollout directory"); + let path = root.join(format!("rollout-2026-06-18T12-00-00-{id}.jsonl")); + let mut file = fs::File::create(&path).expect("create rollout"); + writeln!( + file, + "{}", + json!({ + "timestamp": "2026-06-18T12:00:00Z", + "type": "session_meta", + "payload": { + "id": id, + "timestamp": "2026-06-18T12:00:00Z", + "cwd": self.home.path(), + "originator": "test", + "cli_version": "test", + "source": source, + "model_provider": null, + }, + }) + ) + .expect("write session metadata"); + for line in lines { + writeln!(file, "{line}").expect("write rollout line"); + } + path + } +} + +fn response_item(payload: Value) -> String { + json!({ + "timestamp": "2026-06-18T12:00:01Z", + "type": "response_item", + "payload": payload, + }) + .to_string() +} + +fn function_call(call_id: &str, name: &str, arguments: Value) -> String { + response_item(json!({ + "type": "function_call", + "name": name, + "arguments": arguments.to_string(), + "call_id": call_id, + })) +} + +fn function_output(call_id: &str, output: &str) -> String { + response_item(json!({ + "type": "function_call_output", + "call_id": call_id, + "output": output, + })) +} + +fn direct_command(call_id: &str, command: &str, output: &str) -> Vec { + vec![ + function_call(call_id, "exec_command", json!({"cmd": command})), + function_output(call_id, output), + ] +} + +fn aggregate( + key: &str, + invocation_count: usize, + continuation_wait_count: usize, + total_seconds: u64, + max_seconds: u64, +) -> SlowCommandAggregate { + SlowCommandAggregate { + key: key.to_string(), + invocation_count, + continuation_wait_count, + total_wait: Duration::from_secs(total_seconds), + max_wait: Duration::from_secs(max_seconds), + } +} + +#[tokio::test] +async fn analyzes_direct_shell_waits_across_root_and_subagent_rollouts() { + let fixture = Fixture::new(); + let mut root_lines = Vec::new(); + root_lines.extend(direct_command( + "exec-background", + "cargo test -p codex-core", + "Wall time: 10.0000 seconds\nProcess running with session ID 42\nOutput:", + )); + root_lines.push(function_call( + "poll-background", + "write_stdin", + json!({"session_id": 42, "chars": ""}), + )); + root_lines.push(function_output( + "poll-background", + "Wall time: 5.0000 seconds\nProcess exited with code 0\nOutput:", + )); + root_lines.extend(direct_command( + "exec-check", + "cargo check", + "Wall time: 2.0000 seconds\nProcess exited with code 0\nOutput:", + )); + root_lines.extend(direct_command( + "exec-rg", + "/usr/bin/rg TODO", + "Wall time: 3.0000 seconds\nProcess exited with code 0\nOutput:", + )); + root_lines.extend(direct_command( + "exec-open", + "sleep 100", + "Wall time: 1.0000 seconds\nProcess running with session ID 77\nOutput:", + )); + root_lines.extend(direct_command( + "exec-bad-output", + "echo missing-time", + "Process exited with code 0\nOutput:", + )); + root_lines.push(response_item(json!({ + "type": "function_call", + "name": "exec_command", + "arguments": "{malformed", + "call_id": "exec-bad-arguments", + }))); + root_lines.push(function_output( + "exec-bad-arguments", + "Wall time: 8.0000 seconds\nProcess exited with code 0\nOutput:", + )); + root_lines.push(function_call( + "orphan-poll", + "write_stdin", + json!({"session_id": 999, "chars": ""}), + )); + root_lines.push(function_output( + "orphan-poll", + "Wall time: 4.0000 seconds\nProcess exited with code 0\nOutput:", + )); + root_lines.push(response_item(json!({ + "type": "custom_tool_call", + "call_id": "code-cell", + "name": "exec", + "input": "await tools.exec_command({cmd: 'ignored'})", + }))); + root_lines.push(response_item(json!({ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "ignored"}], + }))); + root_lines.push("{malformed".to_string()); + fixture.write_rollout(/*archived*/ false, SessionSource::Cli, &root_lines); + + let subagent_lines = direct_command( + "subagent-test", + "cargo test -p codex-core", + "Duration: 3.0000 seconds\nOutput:", + ); + fixture.write_rollout( + /*archived*/ true, + SessionSource::SubAgent(SubAgentSource::Review), + &subagent_lines, + ); + + let internal_lines = direct_command( + "internal-call", + "cargo test --workspace", + "Wall time: 100.0000 seconds\nProcess exited with code 0\nOutput:", + ); + fixture.write_rollout( + /*archived*/ false, + SessionSource::Internal(InternalSessionSource::MemoryConsolidation), + &internal_lines, + ); + + let analysis = analyze_slow_commands(fixture.home.path()).await; + + assert_eq!( + analysis, + SlowCommandAnalysis { + summary: SlowCommandSummary { + rollout_files_seen: 3, + rollout_files_analyzed: 2, + rollout_files_skipped: 0, + internal_rollout_files_skipped: 1, + rollout_lines_skipped: 1, + direct_shell_calls_seen: 7, + direct_shell_calls_analyzed: 5, + direct_shell_calls_skipped: 2, + continuation_calls_seen: 2, + continuation_calls_attributed: 1, + continuation_calls_skipped: 1, + duplicate_call_ids: 0, + open_processes_at_eof: 1, + code_mode_cells_ignored: 1, + total_wait: Duration::from_secs(24), + }, + exact_commands: vec![ + aggregate("cargo test -p codex-core", 2, 1, 18, 15), + aggregate("/usr/bin/rg TODO", 1, 0, 3, 3), + aggregate("cargo check", 1, 0, 2, 2), + aggregate("sleep 100", 1, 0, 1, 1), + ], + command_families: vec![ + aggregate("cargo", 3, 1, 20, 15), + aggregate("rg", 1, 0, 3, 3), + aggregate("sleep", 1, 0, 1, 1), + ], + } + ); +} + +#[tokio::test] +async fn sums_full_waits_for_overlapping_background_processes() { + let fixture = Fixture::new(); + let mut lines = Vec::new(); + lines.extend(direct_command( + "first", + "cargo test first", + "Wall time: 10.0000 seconds\nProcess running with session ID 1\nOutput:", + )); + lines.extend(direct_command( + "second", + "cargo test second", + "Wall time: 7.0000 seconds\nProcess running with session ID 2\nOutput:", + )); + for (call_id, session_id, seconds) in [("poll-first", 1, 5), ("poll-second", 2, 4)] { + lines.push(function_call( + call_id, + "write_stdin", + json!({"session_id": session_id, "chars": ""}), + )); + lines.push(function_output( + call_id, + &format!("Wall time: {seconds}.0000 seconds\nProcess exited with code 0\nOutput:"), + )); + } + fixture.write_rollout(/*archived*/ false, SessionSource::Cli, &lines); + + let analysis = analyze_slow_commands(fixture.home.path()).await; + + assert_eq!(analysis.summary.total_wait, Duration::from_secs(26)); + assert_eq!( + analysis.command_families, + vec![aggregate("cargo", 2, 2, 26, 15)] + ); +} + +#[tokio::test] +async fn merges_overlapping_collections_by_direct_and_continuation_call_id() { + let first_fixture = Fixture::new(); + let second_fixture = Fixture::new(); + let mut shared_lines = direct_command( + "shared-direct", + "cargo test merge", + "Wall time: 10.0000 seconds\nProcess running with session ID 42\nOutput:", + ); + shared_lines.push(function_call( + "shared-poll", + "write_stdin", + json!({"session_id": 42, "chars": ""}), + )); + shared_lines.push(function_output( + "shared-poll", + "Wall time: 5.0000 seconds\nProcess running with session ID 42\nOutput:", + )); + first_fixture.write_rollout(/*archived*/ false, SessionSource::Cli, &shared_lines); + + let mut completed_lines = shared_lines; + completed_lines.push(function_call( + "terminal-poll", + "write_stdin", + json!({"session_id": 42, "chars": ""}), + )); + completed_lines.push(function_output( + "terminal-poll", + "Wall time: 7.0000 seconds\nProcess exited with code 0\nOutput:", + )); + second_fixture.write_rollout( + /*archived*/ false, + SessionSource::Cli, + &completed_lines, + ); + + let first = collect_slow_commands(first_fixture.home.path()).await; + let second = collect_slow_commands(second_fixture.home.path()).await; + let analysis = merge_slow_command_collections([first, second]); + + assert_eq!( + analysis, + SlowCommandAnalysis { + summary: SlowCommandSummary { + rollout_files_seen: 2, + rollout_files_analyzed: 2, + direct_shell_calls_seen: 1, + direct_shell_calls_analyzed: 1, + continuation_calls_seen: 2, + continuation_calls_attributed: 2, + duplicate_call_ids: 2, + total_wait: Duration::from_secs(22), + ..Default::default() + }, + exact_commands: vec![aggregate("cargo test merge", 1, 2, 22, 22)], + command_families: vec![aggregate("cargo", 1, 2, 22, 22)], + } + ); +} + +#[tokio::test] +async fn reads_compressed_history_and_deduplicates_copied_calls_and_siblings() { + let fixture = Fixture::new(); + let lines = direct_command( + "copied-call", + "just test -p codex-rollout", + "Wall time: 4.0000 seconds\nProcess exited with code 0\nOutput:", + ); + let active_path = fixture.write_rollout(/*archived*/ false, SessionSource::Cli, &lines); + let active_contents = fs::read(&active_path).expect("read active rollout"); + fs::write( + compressed_rollout_path(&active_path), + zstd::stream::encode_all(active_contents.as_slice(), 0).expect("compress active rollout"), + ) + .expect("write compressed sibling"); + + let archived_path = fixture.write_rollout(/*archived*/ true, SessionSource::Cli, &lines); + let archived_contents = fs::read(&archived_path).expect("read archived rollout"); + let archived_compressed = compressed_rollout_path(&archived_path); + fs::write( + &archived_compressed, + zstd::stream::encode_all(archived_contents.as_slice(), 0) + .expect("compress archived rollout"), + ) + .expect("write compressed archived rollout"); + fs::remove_file(archived_path).expect("remove plain archived rollout"); + + let analysis = analyze_slow_commands(fixture.home.path()).await; + + assert_eq!(analysis.summary.rollout_files_seen, 2); + assert_eq!(analysis.summary.rollout_files_analyzed, 2); + assert_eq!(analysis.summary.duplicate_call_ids, 1); + assert_eq!(analysis.summary.direct_shell_calls_seen, 1); + assert_eq!(analysis.summary.direct_shell_calls_analyzed, 1); + assert_eq!(analysis.summary.total_wait, Duration::from_secs(4)); + assert_eq!( + analysis.exact_commands, + vec![aggregate("just test -p codex-rollout", 1, 0, 4, 4)] + ); +} + +#[tokio::test] +async fn empty_home_returns_an_empty_analysis() { + let fixture = Fixture::new(); + + assert_eq!( + analyze_slow_commands(fixture.home.path()).await, + SlowCommandAnalysis::default() + ); +} + +#[test] +fn executable_family_uses_shell_parsing_and_cross_platform_path_normalization() { + assert_eq!( + parsing::first_executable("FOO=bar /usr/local/bin/cargo test"), + "cargo" + ); + assert_eq!( + parsing::normalize_executable(r"C:\Tools\cargo.EXE"), + "cargo" + ); + assert_eq!(parsing::first_executable(""), ""); +}