Benchmark unified exec through code mode

This commit is contained in:
Channing Conger
2026-07-07 03:53:01 +00:00
parent 7226904ed1
commit 0be58e088a
3 changed files with 587 additions and 29 deletions
+82 -20
View File
@@ -92,6 +92,7 @@ mod plugin_analytics_capture;
mod plugin_analytics_mutation_smoke;
mod plugin_analytics_smoke;
mod request_user_input;
mod unified_exec_benchmark;
const NOTIFICATIONS_TO_OPT_OUT: &[&str] = &[
// v2 item deltas.
@@ -278,6 +279,22 @@ enum CliCommand {
#[arg(long, default_value_t = 15)]
hold_seconds: u64,
},
/// Compare direct and code-mode unified exec using a deterministic local model.
#[command(name = "unified-exec-benchmark")]
UnifiedExecBenchmark {
/// Measured samples per invocation path.
#[arg(long, default_value_t = 20)]
samples: usize,
/// Unmeasured warmup samples per invocation path.
#[arg(long, default_value_t = 3)]
warmups: usize,
/// Duration of the command run by every sample.
#[arg(long, default_value_t = 50)]
command_duration_ms: u64,
/// Existing workspace path used as the turn cwd.
#[arg(long, value_name = "path", default_value = ".")]
workspace: PathBuf,
},
/// Exercise remote plugin analytics through production app-server RPC paths.
#[command(name = "plugin-analytics-smoke")]
PluginAnalyticsSmoke {
@@ -462,6 +479,26 @@ pub async fn run() -> Result<()> {
hold_seconds,
)
}
CliCommand::UnifiedExecBenchmark {
samples,
warmups,
command_duration_ms,
workspace,
} => {
ensure_dynamic_tools_unused(&dynamic_tools, "unified-exec-benchmark")?;
if url.is_some() {
bail!("unified-exec-benchmark requires --codex-bin and does not support --url");
}
let codex_bin = codex_bin.context("unified-exec-benchmark requires --codex-bin")?;
unified_exec_benchmark::run(
&codex_bin,
&config_overrides,
samples,
warmups,
command_duration_ms,
&workspace,
)
}
CliCommand::PluginAnalyticsSmoke {
plugin_id,
capture_file,
@@ -1503,6 +1540,7 @@ struct CodexClient {
unexpected_items_before_helper_done: Vec<ThreadItem>,
last_turn_status: Option<TurnStatus>,
last_turn_error_message: Option<String>,
quiet: bool,
}
#[derive(Debug, Clone, Copy)]
@@ -1592,6 +1630,7 @@ impl CodexClient {
unexpected_items_before_helper_done: Vec::new(),
last_turn_status: None,
last_turn_error_message: None,
quiet: false,
})
}
@@ -1631,6 +1670,7 @@ impl CodexClient {
unexpected_items_before_helper_done: Vec::new(),
last_turn_status: None,
last_turn_error_message: None,
quiet: false,
})
}
@@ -1833,27 +1873,33 @@ impl CodexClient {
match server_notification {
ServerNotification::ThreadStarted(payload) => {
if payload.thread.id == thread_id {
if payload.thread.id == thread_id && !self.quiet {
println!("< thread/started notification: {:?}", payload.thread);
}
}
ServerNotification::TurnStarted(payload) => {
if payload.turn.id == turn_id {
if payload.turn.id == turn_id && !self.quiet {
println!("< turn/started notification: {:?}", payload.turn.status);
}
}
ServerNotification::AgentMessageDelta(delta) => {
print!("{}", delta.delta);
std::io::stdout().flush().ok();
if !self.quiet {
print!("{}", delta.delta);
std::io::stdout().flush().ok();
}
}
ServerNotification::CommandExecutionOutputDelta(delta) => {
self.note_helper_output(&delta.delta);
print!("{}", delta.delta);
std::io::stdout().flush().ok();
if !self.quiet {
print!("{}", delta.delta);
std::io::stdout().flush().ok();
}
}
ServerNotification::TerminalInteraction(delta) => {
println!("[stdin sent: {}]", delta.stdin);
std::io::stdout().flush().ok();
if !self.quiet {
println!("[stdin sent: {}]", delta.stdin);
std::io::stdout().flush().ok();
}
}
ServerNotification::ItemStarted(payload) => {
if matches!(payload.item, ThreadItem::CommandExecution { .. }) {
@@ -1870,7 +1916,9 @@ impl CodexClient {
self.unexpected_items_before_helper_done
.push(payload.item.clone());
}
println!("\n< item started: {:?}", payload.item);
if !self.quiet {
println!("\n< item started: {:?}", payload.item);
}
}
ServerNotification::ItemCompleted(payload) => {
if let ThreadItem::CommandExecution {
@@ -1885,7 +1933,9 @@ impl CodexClient {
self.command_execution_outputs.push(aggregated_output);
}
}
println!("< item completed: {:?}", payload.item);
if !self.quiet {
println!("< item completed: {:?}", payload.item);
}
}
ServerNotification::TurnCompleted(payload) => {
if payload.turn.id == turn_id {
@@ -1898,20 +1948,26 @@ impl CodexClient {
.error
.as_ref()
.map(|error| error.message.clone());
println!("\n< turn/completed notification: {:?}", payload.turn.status);
if payload.turn.status == TurnStatus::Failed
&& let Some(error) = payload.turn.error
{
println!("[turn error] {}", error.message);
if !self.quiet {
println!("\n< turn/completed notification: {:?}", payload.turn.status);
if payload.turn.status == TurnStatus::Failed
&& let Some(error) = payload.turn.error
{
println!("[turn error] {}", error.message);
}
}
break;
}
}
ServerNotification::McpToolCallProgress(payload) => {
println!("< MCP tool progress: {}", payload.message);
if !self.quiet {
println!("< MCP tool progress: {}", payload.message);
}
}
_ => {
println!("[UNKNOWN SERVER NOTIFICATION] {server_notification:?}");
if !self.quiet {
println!("[UNKNOWN SERVER NOTIFICATION] {server_notification:?}");
}
}
}
}
@@ -1955,7 +2011,9 @@ impl CodexClient {
request.trace = current_span_w3c_trace_context();
let request_json = serde_json::to_string(&request)?;
let request_pretty = serde_json::to_string_pretty(&request)?;
print_multiline_with_prefix("> ", &request_pretty);
if !self.quiet {
print_multiline_with_prefix("> ", &request_pretty);
}
self.write_payload(&request_json)
}
@@ -2020,7 +2078,9 @@ impl CodexClient {
let parsed: Value =
serde_json::from_str(trimmed).context("response was not valid JSON-RPC")?;
let pretty = serde_json::to_string_pretty(&parsed)?;
print_multiline_with_prefix("< ", &pretty);
if !self.quiet {
print_multiline_with_prefix("< ", &pretty);
}
let message: JSONRPCMessage = serde_json::from_value(parsed)
.context("response was not a valid JSON-RPC message")?;
return Ok(message);
@@ -2180,7 +2240,9 @@ impl CodexClient {
fn write_jsonrpc_message(&mut self, message: JSONRPCMessage) -> Result<()> {
let payload = serde_json::to_string(&message)?;
let pretty = serde_json::to_string_pretty(&message)?;
print_multiline_with_prefix("> ", &pretty);
if !self.quiet {
print_multiline_with_prefix("> ", &pretty);
}
self.write_payload(&payload)
}
@@ -20,6 +20,20 @@ pub(super) struct LoopbackResponsesServer {
impl LoopbackResponsesServer {
pub(super) fn start() -> Result<Self> {
Self::start_with_responder(|_| {
Ok(concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-plugin-analytics\"}}\n\n",
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-plugin-analytics\",\"usage\":{\"input_tokens\":0,\"input_tokens_details\":null,\"output_tokens\":0,\"output_tokens_details\":null,\"total_tokens\":0}}}\n\n"
)
.to_string())
})
}
pub(super) fn start_with_responder(
mut responder: impl FnMut(&[u8]) -> io::Result<String> + Send + 'static,
) -> Result<Self> {
let listener =
TcpListener::bind("127.0.0.1:0").context("bind loopback Responses API server")?;
listener
@@ -32,7 +46,7 @@ impl LoopbackResponsesServer {
while !thread_shutdown.load(Ordering::Relaxed) {
match listener.accept() {
Ok((stream, _)) => {
if let Err(err) = handle_model_connection(stream) {
if let Err(err) = handle_model_connection(stream, &mut responder) {
eprintln!("loopback Responses API server error: {err}");
}
}
@@ -67,7 +81,10 @@ impl Drop for LoopbackResponsesServer {
}
}
fn handle_model_connection(mut stream: TcpStream) -> io::Result<()> {
fn handle_model_connection(
mut stream: TcpStream,
responder: &mut impl FnMut(&[u8]) -> io::Result<String>,
) -> io::Result<()> {
stream.set_nonblocking(false)?;
stream.set_read_timeout(Some(Duration::from_secs(2)))?;
let request = read_http_request(&mut stream)?;
@@ -77,13 +94,15 @@ fn handle_model_connection(mut stream: TcpStream) -> io::Result<()> {
.and_then(|line| std::str::from_utf8(line).ok())
.unwrap_or_default();
if request_line.starts_with("POST ") && request_line.contains("/responses ") {
let body = concat!(
"event: response.created\n",
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp-plugin-analytics\"}}\n\n",
"event: response.completed\n",
"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-plugin-analytics\",\"usage\":{\"input_tokens\":0,\"input_tokens_details\":null,\"output_tokens\":0,\"output_tokens_details\":null,\"total_tokens\":0}}}\n\n"
);
write_http_response(&mut stream, "200 OK", "text/event-stream", body)
let body = responder(&request)?;
write_http_response(&mut stream, "200 OK", "text/event-stream", &body)
} else if request_line.starts_with("GET ") && request_line.contains("/models") {
write_http_response(
&mut stream,
"200 OK",
"application/json",
r#"{"models":[]}"#,
)
} else {
write_http_response(
&mut stream,
@@ -0,0 +1,477 @@
use super::CodexClient;
use super::loopback_responses_server::LoopbackResponsesServer;
use anyhow::Context;
use anyhow::Result;
use anyhow::bail;
use codex_app_server_protocol::AskForApproval;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::SandboxPolicy;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use serde_json::Value;
use serde_json::json;
use std::collections::VecDeque;
use std::env;
use std::ffi::OsString;
use std::fmt;
use std::fs;
use std::io;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use std::time::Instant;
use uuid::Uuid;
const MOCK_MODEL_SLUG: &str = "unified-exec-benchmark";
const MOCK_PROVIDER_ID: &str = "unified_exec_benchmark";
const OUTPUT_MARKER: &str = "unified_exec_benchmark_marker";
const MAX_COMMAND_DURATION_MS: u64 = 50;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum InvocationPath {
Direct,
CodeMode,
}
impl fmt::Display for InvocationPath {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Direct => f.write_str("direct"),
Self::CodeMode => f.write_str("code_mode"),
}
}
}
#[derive(Clone, Copy, Debug)]
struct PlannedSample {
path: InvocationPath,
warmup: bool,
}
#[derive(Clone, Copy, Debug)]
struct Observation {
path: InvocationPath,
elapsed: Duration,
}
struct BenchmarkResponder {
samples: VecDeque<PlannedSample>,
active_sample: Option<(PlannedSample, Instant)>,
observations: Arc<Mutex<Vec<Observation>>>,
next_response_id: usize,
command: String,
yield_time_ms: u64,
}
struct TemporaryCodexHome {
path: PathBuf,
}
impl TemporaryCodexHome {
fn create() -> Result<Self> {
let path = env::temp_dir().join(format!("codex-unified-exec-benchmark-{}", Uuid::new_v4()));
fs::create_dir(&path)
.with_context(|| format!("create temporary Codex home {}", path.display()))?;
Ok(Self { path })
}
}
impl Drop for TemporaryCodexHome {
fn drop(&mut self) {
if let Err(err) = fs::remove_dir_all(&self.path) {
eprintln!(
"failed to remove temporary Codex home {}: {err}",
self.path.display()
);
}
}
}
impl BenchmarkResponder {
fn respond(&mut self, _request: &[u8]) -> io::Result<String> {
if let Some((sample, started_at)) = self.active_sample.take() {
if !sample.warmup {
self.observations
.lock()
.map_err(|_| io::Error::other("benchmark observations lock poisoned"))?
.push(Observation {
path: sample.path,
elapsed: started_at.elapsed(),
});
}
return Ok(self.completed_response());
}
let sample = self
.samples
.pop_front()
.ok_or_else(|| io::Error::other("unexpected extra Responses API request"))?;
self.active_sample = Some((sample, Instant::now()));
Ok(self.tool_call_response(sample.path))
}
fn tool_call_response(&mut self, path: InvocationPath) -> String {
let response_id = self.response_id();
let call_id = format!("call-{response_id}");
let arguments = json!({
"cmd": &self.command,
"yield_time_ms": self.yield_time_ms,
})
.to_string();
let tool_call = match path {
InvocationPath::Direct => json!({
"type": "response.output_item.done",
"item": {
"type": "function_call",
"call_id": call_id,
"name": "exec_command",
"arguments": arguments,
}
}),
InvocationPath::CodeMode => {
let code = format!(
"const result = await tools.exec_command({arguments});\ntext(result.output);"
);
json!({
"type": "response.output_item.done",
"item": {
"type": "custom_tool_call",
"call_id": call_id,
"name": "exec",
"input": code,
}
})
}
};
sse([
json!({
"type": "response.created",
"response": { "id": &response_id },
}),
tool_call,
completed_event(&response_id),
])
}
fn completed_response(&mut self) -> String {
let response_id = self.response_id();
sse([
json!({
"type": "response.output_item.done",
"item": {
"type": "message",
"role": "assistant",
"id": format!("message-{response_id}"),
"content": [{ "type": "output_text", "text": "done" }],
}
}),
completed_event(&response_id),
])
}
fn response_id(&mut self) -> String {
let response_id = format!("benchmark-{}", self.next_response_id);
self.next_response_id += 1;
response_id
}
}
pub(super) fn run(
codex_bin: &Path,
config_overrides: &[String],
samples: usize,
warmups: usize,
command_duration_ms: u64,
workspace: &Path,
) -> Result<()> {
if cfg!(windows) {
bail!("unified-exec-benchmark currently requires the POSIX sleep utility");
}
if samples == 0 {
bail!("--samples must be greater than zero");
}
if command_duration_ms == 0 || command_duration_ms > MAX_COMMAND_DURATION_MS {
bail!(
"--command-duration-ms must be between 1 and {MAX_COMMAND_DURATION_MS} for the single-call benchmark"
);
}
let workspace = workspace
.canonicalize()
.with_context(|| format!("resolve workspace {}", workspace.display()))?;
let command_duration = Duration::from_millis(command_duration_ms);
let command = format!(
"sleep {:.3}; printf {OUTPUT_MARKER}",
command_duration.as_secs_f64()
);
let schedule = sample_schedule(samples, warmups);
let observations = Arc::new(Mutex::new(Vec::with_capacity(samples * 2)));
let mut responder = BenchmarkResponder {
samples: schedule.clone().into(),
active_sample: None,
observations: Arc::clone(&observations),
next_response_id: 1,
command: command.clone(),
yield_time_ms: command_duration_ms + 1_000,
};
let responses_server =
LoopbackResponsesServer::start_with_responder(move |request| responder.respond(request))?;
let mut overrides = config_overrides.to_vec();
overrides.extend(benchmark_config_overrides(responses_server.base_url())?);
let codex_home = TemporaryCodexHome::create()?;
let environment = [(
OsString::from("CODEX_HOME"),
codex_home.path.as_os_str().to_os_string(),
)];
let mut client = CodexClient::spawn_stdio_with_env(codex_bin, &overrides, &environment)?;
client.quiet = true;
client.initialize()?;
let thread = client.thread_start(ThreadStartParams {
model: Some(MOCK_MODEL_SLUG.to_string()),
model_provider: Some(MOCK_PROVIDER_ID.to_string()),
base_instructions: Some(String::new()),
developer_instructions: Some(String::new()),
ephemeral: Some(true),
..Default::default()
})?;
for sample in schedule {
reset_turn_observations(&mut client);
run_sample(&mut client, &thread.thread.id, &workspace)?;
validate_sample(&client, sample)?;
}
let observations = observations
.lock()
.map_err(|_| anyhow::anyhow!("benchmark observations lock poisoned"))?
.clone();
if observations.len() != samples * 2 {
bail!(
"expected {} measured samples, recorded {}",
samples * 2,
observations.len()
);
}
print_results(
codex_bin,
&command,
samples,
warmups,
command_duration,
&observations,
);
Ok(())
}
fn run_sample(client: &mut CodexClient, thread_id: &str, workspace: &Path) -> Result<()> {
let turn = client.turn_start(TurnStartParams {
thread_id: thread_id.to_string(),
client_user_message_id: None,
input: vec![UserInput::Text {
text: "run the benchmark tool call".to_string(),
text_elements: Vec::new(),
}],
approval_policy: Some(AskForApproval::Never),
sandbox_policy: Some(SandboxPolicy::DangerFullAccess),
cwd: Some(workspace.to_path_buf()),
..Default::default()
})?;
client.stream_turn(thread_id, &turn.turn.id)
}
fn validate_sample(client: &CodexClient, sample: PlannedSample) -> Result<()> {
if client.last_turn_status != Some(TurnStatus::Completed) {
bail!(
"{} sample failed: turn status {:?}, error {:?}",
sample.path,
client.last_turn_status,
client.last_turn_error_message
);
}
if client.command_execution_statuses.last() != Some(&CommandExecutionStatus::Completed) {
bail!(
"{} sample failed: command statuses {:?}",
sample.path,
client.command_execution_statuses
);
}
if !client
.command_execution_outputs
.last()
.is_some_and(|output| output.contains(OUTPUT_MARKER))
{
bail!("{} sample did not emit the expected marker", sample.path);
}
Ok(())
}
fn reset_turn_observations(client: &mut CodexClient) {
client.command_execution_statuses.clear();
client.command_execution_outputs.clear();
client.command_output_stream.clear();
client.command_item_started = false;
client.helper_done_seen = false;
client.turn_completed_before_helper_done = false;
client.unexpected_items_before_helper_done.clear();
client.last_turn_status = None;
client.last_turn_error_message = None;
}
fn sample_schedule(samples: usize, warmups: usize) -> Vec<PlannedSample> {
let mut schedule = Vec::with_capacity((samples + warmups) * 2);
for round in 0..(samples + warmups) {
let paths = if round % 2 == 0 {
[InvocationPath::Direct, InvocationPath::CodeMode]
} else {
[InvocationPath::CodeMode, InvocationPath::Direct]
};
for path in paths {
schedule.push(PlannedSample {
path,
warmup: round < warmups,
});
}
}
schedule
}
fn benchmark_config_overrides(responses_base_url: &str) -> Result<Vec<String>> {
let provider_base_url = quoted(&format!("{responses_base_url}/v1"))?;
Ok(vec![
format!("model={}", quoted(MOCK_MODEL_SLUG)?),
format!("model_provider={}", quoted(MOCK_PROVIDER_ID)?),
format!(
"model_providers.{MOCK_PROVIDER_ID}.name={}",
quoted("Unified exec benchmark mock provider")?
),
format!("model_providers.{MOCK_PROVIDER_ID}.base_url={provider_base_url}"),
format!(
"model_providers.{MOCK_PROVIDER_ID}.wire_api={}",
quoted("responses")?
),
format!("model_providers.{MOCK_PROVIDER_ID}.requires_openai_auth=false"),
format!("model_providers.{MOCK_PROVIDER_ID}.request_max_retries=0"),
format!("model_providers.{MOCK_PROVIDER_ID}.stream_max_retries=0"),
"features.unified_exec=true".to_string(),
"features.code_mode=true".to_string(),
"suppress_unstable_features_warning=true".to_string(),
])
}
fn quoted(value: &str) -> Result<String> {
serde_json::to_string(value).context("serialize config string")
}
fn completed_event(response_id: &str) -> Value {
json!({
"type": "response.completed",
"response": {
"id": response_id,
"usage": {
"input_tokens": 0,
"input_tokens_details": null,
"output_tokens": 0,
"output_tokens_details": null,
"total_tokens": 0,
}
}
})
}
fn sse(events: impl IntoIterator<Item = Value>) -> String {
let mut body = String::new();
for event in events {
let event_type = event
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
body.push_str("event: ");
body.push_str(event_type);
body.push('\n');
body.push_str("data: ");
body.push_str(&event.to_string());
body.push_str("\n\n");
}
body
}
#[derive(Debug)]
struct Summary {
minimum_ms: f64,
p50_ms: f64,
p95_ms: f64,
mean_ms: f64,
}
fn summarize(path: InvocationPath, observations: &[Observation]) -> Summary {
let mut elapsed: Vec<Duration> = observations
.iter()
.filter(|observation| observation.path == path)
.map(|observation| observation.elapsed)
.collect();
elapsed.sort_unstable();
let mean = elapsed.iter().map(Duration::as_secs_f64).sum::<f64>() / elapsed.len() as f64;
Summary {
minimum_ms: elapsed[0].as_secs_f64() * 1_000.0,
p50_ms: percentile(&elapsed, 50).as_secs_f64() * 1_000.0,
p95_ms: percentile(&elapsed, 95).as_secs_f64() * 1_000.0,
mean_ms: mean * 1_000.0,
}
}
fn percentile(sorted: &[Duration], percentile: usize) -> Duration {
let index = ((sorted.len() - 1) * percentile).div_ceil(100);
sorted[index]
}
fn print_results(
codex_bin: &Path,
command: &str,
samples: usize,
warmups: usize,
command_duration: Duration,
observations: &[Observation],
) {
let direct = summarize(InvocationPath::Direct, observations);
let code_mode = summarize(InvocationPath::CodeMode, observations);
let configured_ms = command_duration.as_secs_f64() * 1_000.0;
let delta_ms = code_mode.mean_ms - direct.mean_ms;
let delta_percent = delta_ms / direct.mean_ms * 100.0;
println!("unified exec benchmark");
println!("codex: {}", codex_bin.display());
println!("command: {command}");
println!("warmups: {warmups}/path; samples: {samples}/path");
println!("timing: initial Responses request to follow-up Responses request");
println!();
println!(
"{:<10} {:>10} {:>10} {:>10} {:>10} {:>14}",
"path", "min", "p50", "p95", "mean", "mean-sleep"
);
println!(
"{:<10} {:>9.3}ms {:>9.3}ms {:>9.3}ms {:>9.3}ms {:>13.3}ms",
InvocationPath::Direct,
direct.minimum_ms,
direct.p50_ms,
direct.p95_ms,
direct.mean_ms,
direct.mean_ms - configured_ms,
);
println!(
"{:<10} {:>9.3}ms {:>9.3}ms {:>9.3}ms {:>9.3}ms {:>13.3}ms",
InvocationPath::CodeMode,
code_mode.minimum_ms,
code_mode.p50_ms,
code_mode.p95_ms,
code_mode.mean_ms,
code_mode.mean_ms - configured_ms,
);
println!();
println!("code_mode - direct mean: {delta_ms:+.3}ms ({delta_percent:+.2}%)");
}