mirror of
https://github.com/openai/codex.git
synced 2026-10-25 18:28:38 +00:00
rollout: add slow command analysis
This commit is contained in:
Adam Perry
Generated
+1
@@ -3754,6 +3754,7 @@ dependencies = [
|
||||
"codex-login",
|
||||
"codex-otel",
|
||||
"codex-protocol",
|
||||
"codex-shell-command",
|
||||
"codex-state",
|
||||
"codex-utils-path",
|
||||
"pretty_assertions",
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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<SlowCommandsAction>,
|
||||
}
|
||||
|
||||
#[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<String>,
|
||||
wait_seconds: Option<f64>,
|
||||
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<String>,
|
||||
wait_seconds: Option<f64>,
|
||||
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<W>(&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<Vec<SlowCommandCollection>> {
|
||||
let mut collections = Vec::new();
|
||||
let mut current: Option<SlowCommandCollection> = 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<f64>, line_number: usize) -> Result<Option<Duration>> {
|
||||
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()
|
||||
}
|
||||
@@ -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<String> {
|
||||
run_with(codex_home, &[], None)
|
||||
}
|
||||
|
||||
fn run_with(codex_home: &Path, args: &[&str], stdin: Option<&str>) -> Result<String> {
|
||||
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::<Value>(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<_>>(),
|
||||
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::<Value>)
|
||||
.collect::<serde_json::Result<Vec<_>>>()?;
|
||||
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::<Value>)
|
||||
.collect::<serde_json::Result<Vec<_>>>()?;
|
||||
|
||||
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(())
|
||||
}
|
||||
@@ -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 }
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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<SlowCommandAggregate>,
|
||||
pub command_families: Vec<SlowCommandAggregate>,
|
||||
}
|
||||
|
||||
/// 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<SlowCommandDirectCall>,
|
||||
pub continuations: Vec<SlowCommandContinuation>,
|
||||
}
|
||||
|
||||
/// One observed direct `shell_command` or `exec_command` call.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct SlowCommandDirectCall {
|
||||
pub call_id: String,
|
||||
pub command: Option<String>,
|
||||
pub wait: Option<Duration>,
|
||||
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<String>,
|
||||
pub wait: Option<Duration>,
|
||||
pub observed_running: bool,
|
||||
pub observed_terminal: bool,
|
||||
}
|
||||
|
||||
enum PendingCall {
|
||||
Direct,
|
||||
Continuation { session_id: Option<i64> },
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct FileAnalysisState {
|
||||
pending_calls: HashMap<String, PendingCall>,
|
||||
active_processes: HashMap<i64, String>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct AnalysisState {
|
||||
summary: SlowCommandSummary,
|
||||
direct_calls: HashMap<String, SlowCommandDirectCall>,
|
||||
continuations: HashMap<String, SlowCommandContinuation>,
|
||||
relevant_call_ids: HashSet<String>,
|
||||
code_mode_call_ids: HashSet<String>,
|
||||
}
|
||||
|
||||
/// 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<PathBuf> {
|
||||
let mut dirs = vec![
|
||||
codex_home.join(SESSIONS_SUBDIR),
|
||||
codex_home.join(ARCHIVED_SESSIONS_SUBDIR),
|
||||
];
|
||||
let mut files = BTreeMap::<PathBuf, PathBuf>::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::<RolloutLine>(&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<i64>,
|
||||
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::<Vec<_>>();
|
||||
direct_calls.sort_by(|left, right| left.call_id.cmp(&right.call_id));
|
||||
let mut continuations = state.continuations.into_values().collect::<Vec<_>>();
|
||||
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;
|
||||
@@ -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<Item = SlowCommandCollection>,
|
||||
) -> SlowCommandAnalysis {
|
||||
let mut summary = SlowCommandSummary::default();
|
||||
let mut direct_calls = HashMap::<String, SlowCommandDirectCall>::new();
|
||||
let mut continuations = HashMap::<String, SlowCommandContinuation>::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::<Vec<_>>();
|
||||
let continuations = continuations.into_values().collect::<Vec<_>>();
|
||||
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<String>, incoming: Option<String>) {
|
||||
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<Duration>, incoming: Option<Duration>) {
|
||||
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<SlowCommandAggregate> {
|
||||
let mut aggregates = HashMap::<String, SlowCommandAggregate>::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::<Vec<_>>();
|
||||
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
|
||||
}
|
||||
@@ -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<Regex> = LazyLock::new(|| {
|
||||
compile_regex(r"(?m)^(?:Wall time|Duration): (?<seconds>-?[0-9]+(?:\.[0-9]+)?) seconds\r?$")
|
||||
});
|
||||
static RUNNING_SESSION_RE: LazyLock<Regex> = LazyLock::new(|| {
|
||||
compile_regex(r"(?m)^Process running with session ID (?<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<String> {
|
||||
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::<Option<Vec<_>>>()?;
|
||||
(!parts.is_empty()).then(|| {
|
||||
codex_shell_command::parse_command::shlex_join(
|
||||
&parts.into_iter().map(str::to_string).collect::<Vec<_>>(),
|
||||
)
|
||||
})
|
||||
}
|
||||
Value::Null | Value::Bool(_) | Value::Number(_) | Value::Object(_) | Value::String(_) => {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn parse_session_id(arguments: &str) -> Option<i64> {
|
||||
serde_json::from_str::<Value>(arguments)
|
||||
.ok()?
|
||||
.get("session_id")?
|
||||
.as_i64()
|
||||
}
|
||||
|
||||
pub(super) fn parse_wall_time(output: &str) -> Option<Duration> {
|
||||
let seconds = WALL_TIME_RE
|
||||
.captures(output)?
|
||||
.name("seconds")?
|
||||
.as_str()
|
||||
.parse::<f64>()
|
||||
.ok()?;
|
||||
Duration::try_from_secs_f64(seconds).ok()
|
||||
}
|
||||
|
||||
pub(super) fn parse_running_session_id(output: &str) -> Option<i64> {
|
||||
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("<unknown>"))
|
||||
}
|
||||
|
||||
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() {
|
||||
"<unknown>".to_string()
|
||||
} else {
|
||||
executable.to_string()
|
||||
}
|
||||
}
|
||||
@@ -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<String> {
|
||||
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": "<user_shell_command>ignored</user_shell_command>"}],
|
||||
})));
|
||||
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(""), "<unknown>");
|
||||
}
|
||||
Reference in New Issue
Block a user